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