From e3e34c42112c071ab66dd9cc0e76c2213ce74a80 Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Thu, 29 Aug 2024 18:09:15 -0400 Subject: [PATCH] WxPusher Support Added (#1135) --- KEYWORDS | 1 + README.md | 4 +- apprise/plugins/wxpusher.py | 374 +++++++++++++++++++++++++++ packaging/redhat/python-apprise.spec | 2 +- test/test_plugin_wxpusher.py | 351 +++++++++++++++++++++++++ 5 files changed, 730 insertions(+), 2 deletions(-) create mode 100644 apprise/plugins/wxpusher.py create mode 100644 test/test_plugin_wxpusher.py diff --git a/KEYWORDS b/KEYWORDS index b67e068a9..d624638fc 100644 --- a/KEYWORDS +++ b/KEYWORDS @@ -112,6 +112,7 @@ WeCom Bot WhatsApp Windows Workflows +WxPusher XBMC XML Zulip diff --git a/README.md b/README.md index 164296d8b..06a0eff7e 100644 --- a/README.md +++ b/README.md @@ -131,7 +131,8 @@ The table below identifies the services this tool supports and some example serv | [Twist](https://github.com/caronc/apprise/wiki/Notify_twist) | twist:// | (TCP) 443 | twist://pasword:login
twist://password:login/#channel
twist://password:login/#team:channel
twist://password:login/#team:channel1/channel2/#team3:channel | [Webex Teams (Cisco)](https://github.com/caronc/apprise/wiki/Notify_wxteams) | wxteams:// | (TCP) 443 | wxteams://Token | [WeCom Bot](https://github.com/caronc/apprise/wiki/Notify_wecombot) | wecombot:// | (TCP) 443 | wecombot://BotKey -| [WhatsApp](https://github.com/caronc/apprise/wiki/Notify_whatsapp) | whatsapp:// | (TCP) 443 | whatsapp://AccessToken@FromPhoneID/ToPhoneNo
whatsapp://Template:AccessToken@FromPhoneID/ToPhoneNo +| [WhatsApp](https://github.com/caronc/apprise/wiki/Notify_whatsapp) | whatsapp:// | (TCP) 443 | whatsapp://AccessToken@FromPhoneID/ToPhoneNo
whatsapp://Template:AccessToken@FromPhoneID/ToPhoneNo +| [WxPusher](https://github.com/caronc/apprise/wiki/Notify_wxpusher) | wxpusher:// | (TCP) 443 | wxpusher://AppToken@UserID1/UserID2/UserIDN
wxpusher://AppToken@Topic1/Topic2/Topic3
wxpusher://AppToken@UserID1/Topic1/ | [XBMC](https://github.com/caronc/apprise/wiki/Notify_xbmc) | xbmc:// or xbmcs:// | (TCP) 8080 or 443 | xbmc://hostname
xbmc://user@hostname
xbmc://user:password@hostname:port | [Zulip Chat](https://github.com/caronc/apprise/wiki/Notify_zulip) | zulip:// | (TCP) 443 | zulip://botname@Organization/Token
zulip://botname@Organization/Token/Stream
zulip://botname@Organization/Token/Email @@ -534,6 +535,7 @@ aobj.add('foobar://') # Send our notification out through our foobar:// aobj.notify("test") ``` + You can read more about creating your own custom notifications and/or hooks [here](https://github.com/caronc/apprise/wiki/decorator_notify). # Persistent Storage diff --git a/apprise/plugins/wxpusher.py b/apprise/plugins/wxpusher.py new file mode 100644 index 000000000..e7c9bd672 --- /dev/null +++ b/apprise/plugins/wxpusher.py @@ -0,0 +1,374 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2024, Chris Caron +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# 1. Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# +# 2. Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. +# +# Sign-up at https://wxpusher.zjiecode.com/ +# +# Login and acquire your App Token +# - Open the backend of the application: +# https://wxpusher.zjiecode.com/admin/ +# - Find the appToken menu from the left menu bar, here you can reset the +# appToken, please note that after resetting, the old appToken will be +# invalid immediately and the call interface will fail. +import re +import json +import requests +from itertools import chain +from .base import NotifyBase +from ..url import PrivacyMode +from ..common import NotifyType +from ..common import NotifyFormat +from ..utils import parse_list +from ..utils import validate_regex +from ..locale import gettext_lazy as _ + + +# Topics are always numerical +IS_TOPIC = re.compile(r'^\s*(?P[1-9][0-9]{0,20})\s*$') + +# users always start with UID_ +IS_USER = re.compile( + r'^\s*(?P(?PUID_)(?P[^\s]+))\s*$', re.I) + + +WXPUSHER_RESPONSE_CODES = { + 1000: "The request was processed successfully.", + 1001: "The token provided in the request is missing.", + 1002: "The token provided in the request is incorrect or expired.", + 1003: "The body of the message was not provided.", + 1004: "The user or topic you're trying to send the message to does not " + "exist", + 1005: "The app or topic binding process failed.", + 1006: "There was an error in sending the message.", + 1007: "The message content exceeds the allowed length.", + 1008: "The API call frequency is too high and the server rejected the " + "request.", + 1009: "There might be other issues that are not explicitly covered by " + "the above codes", + 1010: "The IP address making the request is not whitelisted.", +} + + +class WxPusherContentType: + """ + Defines the different supported content types + """ + TEXT = 1 + HTML = 2 + MARKDOWN = 3 + + +class SubscriptionType: + # Verify Subscription Time + UNVERIFIED = 0 + PAID_USERS = 1 + UNSUBSCRIBED = 2 + + +class NotifyWxPusher(NotifyBase): + """ + A wrapper for WxPusher Notifications + """ + + # The default descriptive name associated with the Notification + service_name = 'WxPusher' + + # The services URL + service_url = 'https://wxpusher.zjiecode.com/' + + # The default protocol + secure_protocol = 'wxpusher' + + # A URL that takes you to the setup/help of the specific protocol + setup_url = 'https://github.com/caronc/apprise/wiki/Notify_wxpusher' + + # WxPusher notification endpoint + notify_url = 'https://wxpusher.zjiecode.com/api/send/message' + + # Define object templates + templates = ( + '{schema}://{token}/{targets}', + ) + + # Define our template tokens + template_tokens = dict(NotifyBase.template_tokens, **{ + 'token': { + 'name': _('App Token'), + 'type': 'string', + 'required': True, + 'regex': (r'^AT_[^\s]+$', 'i'), + 'private': True, + }, + 'target_topic': { + 'name': _('Target Topic'), + 'type': 'int', + 'map_to': 'targets', + }, + 'target_user': { + 'name': _('Target User ID'), + 'type': 'string', + 'regex': (r'^UID_[^\s]+$', 'i'), + 'map_to': 'targets', + }, + 'targets': { + 'name': _('Targets'), + 'type': 'list:string', + }, + }) + + # Define our template arguments + template_args = dict(NotifyBase.template_args, **{ + 'to': { + 'alias_of': 'targets', + }, + 'token': { + 'alias_of': 'token', + }, + }) + + # Used for mapping the content type to our output since Apprise supports + # The same formats that WxPusher does. + __content_type_map = { + NotifyFormat.MARKDOWN: WxPusherContentType.MARKDOWN, + NotifyFormat.TEXT: WxPusherContentType.TEXT, + NotifyFormat.HTML: WxPusherContentType.HTML, + } + + def __init__(self, token, targets=None, **kwargs): + """ + Initialize WxPusher Object + """ + super().__init__(**kwargs) + + # App Token (associated with WxPusher account) + self.token = validate_regex( + token, *self.template_tokens['token']['regex']) + if not self.token: + msg = 'An invalid WxPusher App Token ' \ + '({}) was specified.'.format(token) + self.logger.warning(msg) + raise TypeError(msg) + + # Used for URL generation afterwards only + self._invalid_targets = list() + + # For storing what is detected + self._users = list() + self._topics = list() + + # Parse our targets + for target in parse_list(targets): + # Validate targets and drop bad ones: + result = IS_USER.match(target) + if result: + # store valid user + self._users.append(result['full']) + continue + + result = IS_TOPIC.match(target) + if result: + # store valid topic + self._topics.append(int(result['topic'])) + continue + + self.logger.warning( + 'Dropped invalid WxPusher user/topic ' + '(%s) specified.' % target, + ) + self._invalid_targets.append(target) + + return + + def send(self, body, title='', notify_type=NotifyType.INFO, **kwargs): + """ + Perform WxPusher Notification + """ + + if not self._users and not self._topics: + # There were no services to notify + self.logger.warning( + 'There were no WxPusher targets to notify') + return False + + # Prepare our headers + headers = { + 'User-Agent': self.app_id, + 'Content-Type': 'application/json', + 'Accept': 'application/json', + } + + # Prepare our payload + payload = { + 'appToken': self.token, + 'content': body, + 'summary': title, + 'contentType': self.__content_type_map[self.notify_format], + 'topicIds': self._topics, + 'uids': self._users, + + # unsupported at this time + # 'verifyPay': False, + # 'verifyPayType': 0, + 'url': None, + } + + # Some Debug Logging + self.logger.debug('WxPusher POST URL: {} (cert_verify={})'.format( + self.notify_url, self.verify_certificate)) + self.logger.debug('WxPusher Payload: {}' .format(payload)) + + # Always call throttle before any remote server i/o is made + self.throttle() + + try: + r = requests.post( + self.notify_url, + data=json.dumps(payload).encode('utf-8'), + headers=headers, + verify=self.verify_certificate, + timeout=self.request_timeout, + ) + + try: + content = json.loads(r.content) + + except (AttributeError, TypeError, ValueError): + # ValueError = r.content is Unparsable + # TypeError = r.content is None + # AttributeError = r is None + content = {} + + # 1000 is the expected return code for a successful query + if r.status_code == requests.codes.ok and \ + content and content.get("code") == 1000: + + # We're good! + self.logger.info( + 'Sent WxPusher notification to %d targets.' % ( + len(self._users) + len(self._topics))) + + else: + error_str = content.get('msg') if content else ( + WXPUSHER_RESPONSE_CODES.get( + content.get("code") if content else None, + "An unknown error occured.")) + + # We had a problem + status_str = \ + NotifyWxPusher.http_response_code_lookup( + r.status_code) if not error_str else error_str + + self.logger.warning( + 'Failed to send WxPusher notification, ' + 'code={}/{}: {}'.format( + r.status_code, + 'unk' if not content else content.get("code"), + status_str)) + + self.logger.debug( + 'Response Details:\r\n{}'.format( + content if content else r.content)) + + # Mark our failure + return False + + except requests.RequestException as e: + self.logger.warning( + 'A Connection error occurred sending WxPusher ' + 'notification.' + ) + self.logger.debug('Socket Exception: %s' % str(e)) + + return False + + return True + + @property + def url_identifier(self): + """ + Returns all of the identifiers that make this URL unique from + another simliar one. Targets or end points should never be identified + here. + """ + return (self.secure_protocol, self.token) + + def url(self, privacy=False, *args, **kwargs): + """ + Returns the URL built dynamically based on specified arguments. + """ + + # Define any URL parameters + params = self.url_parameters(privacy=privacy, *args, **kwargs) + + return '{schema}://{token}/{targets}/?{params}'.format( + schema=self.secure_protocol, + token=self.pprint( + self.token, privacy, mode=PrivacyMode.Secret, safe=''), + targets='/'.join(chain( + [str(t) for t in self._topics], self._users, + [NotifyWxPusher.quote(x, safe='') + for x in self._invalid_targets])), + params=NotifyWxPusher.urlencode(params)) + + @staticmethod + def parse_url(url): + """ + Parses the URL and returns enough arguments that can allow + us to re-instantiate this object. + + """ + results = NotifyBase.parse_url(url, verify_host=False) + if not results: + # We're done early as we couldn't load the results + return results + + # Get our entries; split_path() looks after unquoting content for us + # by default + results['targets'] = NotifyWxPusher.split_path(results['fullpath']) + + # App Token + if 'token' in results['qsd'] and len(results['qsd']['token']): + # Extract the App token from an argument + results['token'] = \ + NotifyWxPusher.unquote(results['qsd']['token']) + # Any host entry defined is actually part of the path + # store it's element (if defined) + if results['host']: + results['targets'].append( + NotifyWxPusher.split_path(results['host'])) + + else: + # The hostname is our source number + results['token'] = NotifyWxPusher.unquote(results['host']) + + # Support the 'to' variable so that we can support rooms this way too + # The 'to' makes it easier to use yaml configuration + if 'to' in results['qsd'] and len(results['qsd']['to']): + results['targets'] += \ + NotifyWxPusher.parse_list(results['qsd']['to']) + + return results diff --git a/packaging/redhat/python-apprise.spec b/packaging/redhat/python-apprise.spec index dd6253be7..e0b6fd756 100644 --- a/packaging/redhat/python-apprise.spec +++ b/packaging/redhat/python-apprise.spec @@ -52,7 +52,7 @@ Reddit, Rocket.Chat, RSyslog, SendGrid, ServerChan, SFR, Signal, SimplePush, Sinch, Slack, SMSEagle, SMS Manager, SMTP2Go, SparkPost, Splunk, Super Toasty, Streamlabs, Stride, Synology Chat, Syslog, Techulus Push, Telegram, Threema Gateway, Twilio, Twitter, Twist, VictorOps, Voipms, Vonage, WeCom Bot, -WhatsApp, Webex Teams, Workflows, XBMC} +WhatsApp, Webex Teams, Workflows, WxPusher, XBMC} Name: python-%{pypi_name} Version: 1.8.1 diff --git a/test/test_plugin_wxpusher.py b/test/test_plugin_wxpusher.py new file mode 100644 index 000000000..e20c45a3d --- /dev/null +++ b/test/test_plugin_wxpusher.py @@ -0,0 +1,351 @@ +# -*- coding: utf-8 -*- +# BSD 2-Clause License +# +# Apprise - Push Notification Library. +# Copyright (c) 2024, Chris Caron +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# +# 1. Redistributions of source code must retain the above copyright notice, +# this list of conditions and the following disclaimer. +# +# 2. Redistributions in binary form must reproduce the above copyright notice, +# this list of conditions and the following disclaimer in the documentation +# and/or other materials provided with the distribution. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE +# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF +# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS +# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN +# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) +# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE +# POSSIBILITY OF SUCH DAMAGE. + +import os +from json import loads, dumps +from unittest import mock + +import requests +from apprise import Apprise +from apprise.plugins.wxpusher import NotifyWxPusher +from helpers import AppriseURLTester + +# Disable logging for a cleaner testing output +import logging +logging.disable(logging.CRITICAL) + +WXPUSHER_GOOD_RESPONSE = dumps({"code": 1000}) +WXPUSHER_BAD_RESPONSE = dumps({"code": 99}) + + +# Attachment Directory +TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var') + +# Our Testing URLs +apprise_url_tests = ( + ('wxpusher://', { + # No token specified + 'instance': TypeError, + }), + ('wxpusher://:@/', { + # invalid url + 'instance': TypeError, + }), + ('wxpusher://invalid', { + # invalid app token + 'instance': TypeError, + }), + ('wxpusher://AT_appid/123/', { + # invalid 'to' phone number + 'instance': NotifyWxPusher, + # Notify will fail because it couldn't send to anyone + 'response': False, + # Our expected url(privacy=True) startswith() response: + 'privacy_url': 'wxpusher://****/123/', + # Our response expected server response + 'requests_response_text': WXPUSHER_GOOD_RESPONSE, + }), + ('wxpusher://AT_appid/%20/%20/', { + # invalid 'to' phone number + 'instance': NotifyWxPusher, + # Notify will fail because it couldn't send to anyone + 'response': False, + # Our expected url(privacy=True) startswith() response: + 'privacy_url': 'wxpusher://****/', + # Our response expected server response + 'requests_response_text': WXPUSHER_GOOD_RESPONSE, + }), + ('wxpusher://AT_appid/123/', { + # one phone number will notify ourselves + 'instance': NotifyWxPusher, + # Our response expected server response + 'requests_response_text': WXPUSHER_GOOD_RESPONSE, + }), + ('wxpusher://123?token=AT_abc1234', { + # pass our token in as an argument and our host actually becomes a + # target + 'instance': NotifyWxPusher, + # Our response expected server response + 'requests_response_text': WXPUSHER_GOOD_RESPONSE, + }), + ('wxpusher://?token=AT_abc1234', { + # slightly different then test above; a token is defined, but + # there are no targets + 'instance': NotifyWxPusher, + # Notify will fail because it couldn't send to anyone + 'response': False, + # Our response expected server response + 'requests_response_text': WXPUSHER_GOOD_RESPONSE, + }), + ('wxpusher://?token=AT_abc1234&to=UID_abc', { + # all kwargs to load url with + 'instance': NotifyWxPusher, + # Our response expected server response + 'requests_response_text': WXPUSHER_GOOD_RESPONSE, + }), + ('wxpusher://AT_appid/UID_abcd/', { + # a valid contact + 'instance': NotifyWxPusher, + # Our expected url(privacy=True) startswith() response: + 'privacy_url': 'wxpusher://****/UID_abcd', + # Our response expected server response + 'requests_response_text': WXPUSHER_GOOD_RESPONSE, + }), + ('wxpusher://AT_appid/@/#/,/', { + # Test case where we provide bad data + 'instance': NotifyWxPusher, + # Our failed response + 'requests_response_text': WXPUSHER_GOOD_RESPONSE, + # as a result, we expect a failed notification + 'response': False, + }), + ('wxpusher://AT_appid/123/', { + # Test case where we get a bad response + 'instance': NotifyWxPusher, + # Our expected url(privacy=True) startswith() response: + 'privacy_url': 'wxpusher://****/123', + # Our failed response + 'requests_response_text': WXPUSHER_BAD_RESPONSE, + # as a result, we expect a failed notification + 'response': False, + }), + ('wxpusher://AT_appid/UID_345/', { + # Test case where we get a bad response + 'instance': NotifyWxPusher, + # Our expected url(privacy=True) startswith() response: + 'privacy_url': 'wxpusher://****/UID_345', + # Our failed response + 'requests_response_text': None, + # as a result, we expect a failed notification + 'response': False, + }), + ('wxpusher://AT_appid/UID_345/', { + # Test case where we get a bad response + 'instance': NotifyWxPusher, + # Our expected url(privacy=True) startswith() response: + 'privacy_url': 'wxpusher://****/UID_345', + # Our failed response (bad json) + 'requests_response_text': '{', + # as a result, we expect a failed notification + 'response': False, + }), + ('wxpusher://AT_appid/?to={},{}'.format( + '2' * 11, '3' * 11), { + # use get args to acomplish the same thing + 'instance': NotifyWxPusher, + # Our response expected server response + 'requests_response_text': WXPUSHER_GOOD_RESPONSE, + }), + ('wxpusher://AT_appid/?to={},{},{}'.format( + '2' * 11, '3' * 11, '5' * 3), { + # 2 good targets and one invalid one + 'instance': NotifyWxPusher, + # Our response expected server response + 'requests_response_text': WXPUSHER_GOOD_RESPONSE, + }), + ('wxpusher://AT_appid/{}/{}/'.format( + '2' * 11, '3' * 11), { + # If we have from= specified, then all elements take on the to= value + 'instance': NotifyWxPusher, + # Our response expected server response + 'requests_response_text': WXPUSHER_GOOD_RESPONSE, + }), + ('wxpusher://AT_appid/{}'.format('3' * 11), { + # use get args to acomplish the same thing (use source instead of from) + 'instance': NotifyWxPusher, + # Our response expected server response + 'requests_response_text': WXPUSHER_GOOD_RESPONSE, + }), + ('wxpusher://AT_appid/{}'.format('4' * 11), { + 'instance': NotifyWxPusher, + # throw a bizzare code forcing us to fail to look it up + 'response': False, + 'requests_response_code': 999, + # Our response expected server response + 'requests_response_text': WXPUSHER_GOOD_RESPONSE, + }), + ('wxpusher://AT_appid/{}'.format('4' * 11), { + 'instance': NotifyWxPusher, + # Throws a series of connection and transfer exceptions when this flag + # is set and tests that we gracfully handle them + 'test_requests_exceptions': True, + }), +) + + +def test_plugin_wxpusher_urls(): + """ + NotifyWxPusher() Apprise URLs + + """ + + # Run our general tests + AppriseURLTester(tests=apprise_url_tests).run_all() + + +@mock.patch('requests.post') +def test_plugin_wxpusher_edge_cases(mock_post): + """ + NotifyWxPusher() Edge Cases + + """ + + # Prepare our response + response = requests.Request() + response.status_code = requests.codes.ok + response.content = WXPUSHER_GOOD_RESPONSE + + # Prepare Mock + mock_post.return_value = response + + # Initialize some generic (but valid) tokens + target = 'UID_abcd' + body = "test body" + title = "My Title" + + aobj = Apprise() + assert aobj.add("wxpusher://AT_appid/{}".format(target)) + assert len(aobj) == 1 + assert aobj.notify(title=title, body=body) + assert mock_post.call_count == 1 + + details = mock_post.call_args_list[0] + assert details[0][0] == 'https://wxpusher.zjiecode.com/api/send/message' + payload = loads(details[1]['data']) + assert payload == { + 'appToken': 'AT_appid', + 'content': 'test body', + 'summary': 'My Title', + 'contentType': 1, + 'topicIds': [], + 'uids': ['UID_abcd'], + 'url': None, + } + + # Reset our mock object + mock_post.reset_mock() + + +@mock.patch('requests.post') +def test_plugin_wxpusher_result_set(mock_post): + """ + NotifyWxPusher() Result Sets + + """ + + # Prepare our response + response = requests.Request() + response.status_code = requests.codes.ok + response.content = WXPUSHER_GOOD_RESPONSE + + # Prepare Mock + mock_post.return_value = response + + body = "test body" + title = "My Title" + + aobj = Apprise() + aobj.add('wxpusher://AT_appid/123/abc/UID_456') + # One bad entry and 2 good + assert len(aobj[0]) == 1 + + assert aobj.notify(title=title, body=body) + + # 2 posts made + assert mock_post.call_count == 1 + + details = mock_post.call_args_list[0] + assert details[0][0] == 'https://wxpusher.zjiecode.com/api/send/message' + payload = loads(details[1]['data']) + assert payload == { + 'appToken': 'AT_appid', + 'content': 'test body', + 'summary': 'My Title', + 'contentType': 1, + 'topicIds': [123], + 'uids': ['UID_456'], + 'url': None, + } + + # Validate our information is also placed back into the assembled URL + assert '/123' in aobj[0].url() + assert '/UID_456' in aobj[0].url() + assert '/abc' in aobj[0].url() + + mock_post.reset_mock() + + aobj = Apprise() + aobj.add('wxpusher://AT_appid//UID_123/UID_abc/123456789') + assert len(aobj[0]) == 1 + + assert aobj.notify(title=title, body=body) + + # If batch is off then there is a post per entry + assert mock_post.call_count == 1 + + details = mock_post.call_args_list[0] + assert details[0][0] == 'https://wxpusher.zjiecode.com/api/send/message' + payload = loads(details[1]['data']) + + assert payload == { + 'appToken': 'AT_appid', + 'content': 'test body', + 'summary': 'My Title', + 'contentType': 1, + 'topicIds': [123456789], + 'uids': ['UID_123', 'UID_abc'], + 'url': None, + } + + assert '/123456789' in aobj[0].url() + assert '/UID_123' in aobj[0].url() + assert '/UID_abc' in aobj[0].url() + + +@mock.patch('requests.post') +def test_notify_wxpusher_plugin_result_list(mock_post): + """ + NotifyWxPusher() Result List Response + + """ + + okay_response = requests.Request() + okay_response.status_code = requests.codes.ok + # We want to test the case where the `result` set returned is a list + + # Invalid JSON response + okay_response.content = '{' + + # Assign our mock object our return value + mock_post.return_value = okay_response + + obj = Apprise.instantiate('wxpusher://AT_apptoken/UID_abcd/') + assert isinstance(obj, NotifyWxPusher) + + # We should now fail + assert obj.notify("test") is False