From 10986b89a2b06c942072c87a6570794973cb966a Mon Sep 17 00:00:00 2001 From: Chris Caron Date: Sat, 1 Jun 2024 10:30:17 -0400 Subject: [PATCH] test case coverage --- apprise/plugins/wxpusher.py | 25 +-- test/test_plugin_wxpusher.py | 357 +++++++++++++++++++++++++++++++++++ 2 files changed, 371 insertions(+), 11 deletions(-) create mode 100644 test/test_plugin_wxpusher.py diff --git a/apprise/plugins/wxpusher.py b/apprise/plugins/wxpusher.py index b34af1c7c9..6a788e3eb2 100644 --- a/apprise/plugins/wxpusher.py +++ b/apprise/plugins/wxpusher.py @@ -25,11 +25,15 @@ # 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 Token +# 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 @@ -85,7 +89,7 @@ class NotifyWxPusher(NotifyBase): setup_url = 'https://github.com/caronc/apprise/wiki/Notify_wxpusher' # WxPusher notification endpoint - notify_url = 'http://wxpusher.zjiecode.com/api/send/message' + notify_url = 'https://wxpusher.zjiecode.com/api/send/message' # Define object templates templates = ( @@ -296,13 +300,6 @@ def url(self, privacy=False, *args, **kwargs): for x in self._invalid_targets])), params=NotifyWxPusher.urlencode(params)) - def __len__(self): - """ - Returns the number of targets associated with this notification - """ - targets = len(self._users) + len(self._topics) - return targets if targets > 0 else 1 - @staticmethod def parse_url(url): """ @@ -324,6 +321,12 @@ def parse_url(url): # 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']) diff --git a/test/test_plugin_wxpusher.py b/test/test_plugin_wxpusher.py new file mode 100644 index 0000000000..ea3ffcb9cc --- /dev/null +++ b/test/test_plugin_wxpusher.py @@ -0,0 +1,357 @@ +# -*- 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({"success": True}) +WXPUSHER_BAD_RESPONSE = dumps({"success": False}) + + +# 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': 0, + 'topicIds': [], + 'uids': ['UID_abcd'], + 'verifyPay': False, + 'verifyPayType': 0, + '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': 0, + 'topicIds': [123], + 'uids': ['UID_456'], + 'verifyPay': False, + 'verifyPayType': 0, + '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': 0, + 'topicIds': [123456789], + 'uids': ['UID_123', 'UID_abc'], + 'verifyPay': False, + 'verifyPayType': 0, + '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