# -*- 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. from unittest import mock import os import requests import pytest from apprise import ( Apprise, AppriseAsset, AppriseAttachment, NotifyType, PersistentStoreMode) from json import dumps from apprise.plugins.matrix import NotifyMatrix from helpers import AppriseURLTester # Disable logging for a cleaner testing output import logging logging.disable(logging.CRITICAL) MATRIX_GOOD_RESPONSE = dumps({ 'room_id': '!abc123:localhost', 'room_alias': '#abc123:localhost', 'joined_rooms': ['!abc123:localhost', '!def456:localhost'], 'access_token': 'abcd1234', 'home_server': 'localhost', }) # Attachment Directory TEST_VAR_DIR = os.path.join(os.path.dirname(__file__), 'var') # Our Testing URLs apprise_url_tests = ( ################################## # NotifyMatrix ################################## ('matrix://', { 'instance': None, }), ('matrixs://', { 'instance': None, }), ('matrix://localhost?mode=off', { # treats it as a anonymous user to register 'instance': NotifyMatrix, # response is false because we have nothing to notify 'response': False, }), ('matrix://localhost', { # response is TypeError because we'll try to initialize as # a t2bot and fail (localhost is too short of a api key) 'instance': TypeError }), ('matrix://user:pass@localhost/#room1/#room2/#room3', { 'instance': NotifyMatrix, 'response': False, 'requests_response_code': requests.codes.internal_server_error, }), ('matrix://user:pass@localhost/#room1/#room2/!room1', { 'instance': NotifyMatrix, # throw a bizzare code forcing us to fail to look it up 'response': False, 'requests_response_code': 999, }), ('matrix://user:pass@localhost:1234/#room', { 'instance': NotifyMatrix, # Throws a series of connection and transfer exceptions when this flag # is set and tests that we gracfully handle them 'test_requests_exceptions': True, # Our expected url(privacy=True) startswith() response: 'privacy_url': 'matrix://user:****@localhost:1234/', }), # Matrix supports webhooks too; the following tests this now: ('matrix://user:token@localhost?mode=matrix&format=text', { # user and token correctly specified with webhook 'instance': NotifyMatrix, 'response': False, }), ('matrix://user:token@localhost?mode=matrix&format=html', { # user and token correctly specified with webhook 'instance': NotifyMatrix, }), ('matrix://user:token@localhost:123/#general/?version=3', { # Provide version over-ride (using version=) 'instance': NotifyMatrix, # Our response expected server response 'requests_response_text': MATRIX_GOOD_RESPONSE, 'privacy_url': 'matrix://user:****@localhost:123', }), ('matrixs://user:token@localhost/#general?v=2', { # Provide version over-ride (using v=) 'instance': NotifyMatrix, # Our response expected server response 'requests_response_text': MATRIX_GOOD_RESPONSE, 'privacy_url': 'matrixs://user:****@localhost', }), ('matrix://user:token@localhost:123/#general/?v=invalid', { # Invalid version specified 'instance': TypeError }), ('matrix://user:token@localhost?mode=slack&format=text', { # user and token correctly specified with webhook 'instance': NotifyMatrix, }), ('matrixs://user:token@localhost?mode=SLACK&format=markdown', { # user and token specified; slack webhook still detected # despite uppercase characters 'instance': NotifyMatrix, }), ('matrix://user@localhost?mode=SLACK&format=markdown&token=mytoken', { # user and token specified; slack webhook still detected # despite uppercase characters; token also set on URL as arg 'instance': NotifyMatrix, }), ('matrix://_?mode=t2bot&token={}'.format('b' * 64), { # Testing t2bot initialization and setting the password using the # token directive 'instance': NotifyMatrix, # Our expected url(privacy=True) startswith() response: 'privacy_url': 'matrix://b...b/', }), # Image Reference ('matrixs://user:token@localhost?mode=slack&format=markdown&image=True', { # user and token specified; image set to True 'instance': NotifyMatrix, }), ('matrixs://user:token@localhost?mode=slack&format=markdown&image=False', { # user and token specified; image set to True 'instance': NotifyMatrix, }), # A Bunch of bad ports ('matrixs://user:pass@hostname:port/#room_alias', { # Invalid Port specified (was a string) 'instance': TypeError, }), ('matrixs://user:pass@hostname:0/#room_alias', { # Invalid Port specified (was a string) 'instance': TypeError, }), ('matrixs://user:pass@hostname:65536/#room_alias', { # Invalid Port specified (was a string) 'instance': TypeError, }), # More general testing... ('matrixs://user@{}?mode=t2bot&format=markdown&image=True' .format('a' * 64), { # user and token specified; image set to True 'instance': NotifyMatrix}), ('matrix://user@{}?mode=t2bot&format=html&image=False' .format('z' * 64), { # user and token specified; image set to True 'instance': NotifyMatrix}), # This will default to t2bot because no targets were specified and no # password ('matrixs://{}'.format('c' * 64), { 'instance': NotifyMatrix, # Throws a series of connection and transfer exceptions when this flag # is set and tests that we gracfully handle them 'test_requests_exceptions': True, }), # Test Native URL ('https://webhooks.t2bot.io/api/v1/matrix/hook/{}/'.format('d' * 64), { # user and token specified; image set to True 'instance': NotifyMatrix, }), ('matrix://user:token@localhost?mode=On', { # invalid webhook specified (unexpected boolean) 'instance': TypeError, }), ('matrix://token@localhost/?mode=Matrix', { 'instance': NotifyMatrix, 'response': False, 'requests_response_code': requests.codes.internal_server_error, }), ('matrix://user:token@localhost/mode=matrix', { 'instance': NotifyMatrix, # throw a bizzare code forcing us to fail to look it up 'response': False, 'requests_response_code': 999, }), ('matrix://token@localhost:8080/?mode=slack', { 'instance': NotifyMatrix, # Throws a series of connection and transfer exceptions when this flag # is set and tests that we gracfully handle them 'test_requests_exceptions': True, }), ('matrix://{}/?mode=t2bot'.format('b' * 64), { 'instance': NotifyMatrix, # 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_matrix_urls(): """ NotifyMatrix() Apprise URLs """ # Run our general tests AppriseURLTester(tests=apprise_url_tests).run_all() @mock.patch('requests.put') @mock.patch('requests.get') @mock.patch('requests.post') def test_plugin_matrix_general(mock_post, mock_get, mock_put): """ NotifyMatrix() General Tests """ response_obj = { 'room_id': '!abc123:localhost', 'room_alias': '#abc123:localhost', 'joined_rooms': ['!abc123:localhost', '!def456:localhost'], 'access_token': 'abcd1234', 'home_server': 'localhost', } request = mock.Mock() request.content = dumps(response_obj) request.status_code = requests.codes.ok # Prepare Mock mock_get.return_value = request mock_post.return_value = request mock_put.return_value = request # Variation Initializations obj = NotifyMatrix(host='host', targets='#abcd') assert isinstance(obj, NotifyMatrix) assert isinstance(obj.url(), str) # Registration successful assert obj.send(body="test") is True del obj obj = NotifyMatrix(host='host', user='user', targets='#abcd') assert isinstance(obj, NotifyMatrix) assert isinstance(obj.url(), str) # Registration successful assert obj.send(body="test") is True del obj obj = NotifyMatrix(host='host', password='passwd', targets='#abcd') assert isinstance(obj, NotifyMatrix) assert isinstance(obj.url(), str) # A username gets automatically generated in these cases assert obj.send(body="test") is True del obj obj = NotifyMatrix( host='host', user='user', password='passwd', targets='#abcd') assert isinstance(obj.url(), str) assert isinstance(obj, NotifyMatrix) # Registration Successful assert obj.send(body="test") is True del obj # Test sending other format types kwargs = NotifyMatrix.parse_url( 'matrix://user:passwd@hostname/#abcd?format=html') obj = NotifyMatrix(**kwargs) assert isinstance(obj.url(), str) assert isinstance(obj, NotifyMatrix) assert obj.send(body="test") is True assert obj.send(title="title", body="test") is True del obj kwargs = NotifyMatrix.parse_url( 'matrix://user:passwd@hostname/#abcd/#abcd:localhost?format=markdown') obj = NotifyMatrix(**kwargs) assert isinstance(obj.url(), str) assert isinstance(obj, NotifyMatrix) assert obj.send(body="test") is True assert obj.send(title="title", body="test") is True del obj kwargs = NotifyMatrix.parse_url( 'matrix://user:passwd@hostname/#abcd/!abcd:localhost?format=text') obj = NotifyMatrix(**kwargs) assert isinstance(obj.url(), str) assert isinstance(obj, NotifyMatrix) is True obj.send(body="test") is True obj.send(title="title", body="test") is True del obj # Test notice type notifications kwargs = NotifyMatrix.parse_url( 'matrix://user:passwd@hostname/#abcd?msgtype=notice') obj = NotifyMatrix(**kwargs) assert isinstance(obj.url(), str) is True assert isinstance(obj, NotifyMatrix) is True assert obj.send(body="test") is True assert obj.send(title="title", body="test") is True with pytest.raises(TypeError): # invalid message type specified kwargs = NotifyMatrix.parse_url( 'matrix://user:passwd@hostname/#abcd?msgtype=invalid') NotifyMatrix(**kwargs) # Force a failed login ro = response_obj.copy() del ro['access_token'] request.content = dumps(ro) request.status_code = 404 # Fails because we couldn't register because of 404 errors assert obj.send(body="test") is False del obj obj = NotifyMatrix(host='host', user='test', targets='#abcd') assert isinstance(obj, NotifyMatrix) is True # Fails because we still couldn't register assert obj.send(user='test', password='passwd', body="test") is False del obj obj = NotifyMatrix( host='host', user='test', password='passwd', targets='#abcd') assert isinstance(obj, NotifyMatrix) is True # Fails because we still couldn't register assert obj.send(body="test") is False del obj obj = NotifyMatrix(host='host', password='passwd', targets='#abcd') # Fails because we still couldn't register assert isinstance(obj, NotifyMatrix) is True assert obj.send(body="test") is False # Force a empty joined list response ro = response_obj.copy() ro['joined_rooms'] = [] request.content = dumps(ro) assert obj.send(user='test', password='passwd', body="test") is False # Fall back to original template request.content = dumps(response_obj) request.status_code = requests.codes.ok # update our response object so logins now succeed response_obj['user_id'] = '@apprise:localhost' # Login was successful but not get a room_id ro = response_obj.copy() del ro['room_id'] request.content = dumps(ro) assert obj.send(user='test', password='passwd', body="test") is False # Fall back to original template request.content = dumps(response_obj) request.status_code = requests.codes.ok del obj obj = NotifyMatrix(host='host', targets=None) assert isinstance(obj, NotifyMatrix) is True # Force a empty joined list response ro = response_obj.copy() ro['joined_rooms'] = [] request.content = dumps(ro) assert obj.send(user='test', password='passwd', body="test") is False # Fall back to original template request.content = dumps(response_obj) request.status_code = requests.codes.ok # our room list is empty so we'll have retrieved the joined_list # as our backup assert obj.send(user='test', password='passwd', body="test") is True del obj @mock.patch('requests.put') @mock.patch('requests.get') @mock.patch('requests.post') def test_plugin_matrix_fetch(mock_post, mock_get, mock_put): """ NotifyMatrix() Server Fetch/API Tests """ response_obj = { 'room_id': '!abc123:localhost', 'room_alias': '#abc123:localhost', 'joined_rooms': ['!abc123:localhost', '!def456:localhost'], # Login details 'access_token': 'abcd1234', 'user_id': '@apprise:localhost', 'home_server': 'localhost', } def fetch_failed(url, *args, **kwargs): # Default configuration request = mock.Mock() request.status_code = requests.codes.ok request.content = dumps(response_obj) if url.find('/rooms/') > -1: # over-ride on room query request.status_code = 403 request.content = dumps({ u'errcode': u'M_UNKNOWN', u'error': u'Internal server error', }) return request mock_put.side_effect = fetch_failed mock_get.side_effect = fetch_failed mock_post.side_effect = fetch_failed obj = NotifyMatrix( host='host', user='user', password='passwd', include_image=True) assert isinstance(obj, NotifyMatrix) is True # We would hve failed to send our image notification assert obj.send(user='test', password='passwd', body="test") is False del obj # Do the same query with no images to fetch asset = AppriseAsset(image_path_mask=False, image_url_mask=False) obj = NotifyMatrix( host='host', user='user', password='passwd', asset=asset) assert isinstance(obj, NotifyMatrix) is True # We would hve failed to send our notification assert obj.send(user='test', password='passwd', body="test") is False del obj response_obj = { # Registration 'access_token': 'abcd1234', 'user_id': '@apprise:localhost', 'home_server': 'localhost', # For room joining 'room_id': '!abc123:localhost', } # Default configuration mock_get.side_effect = None mock_post.side_effect = None mock_put.side_effect = None request = mock.Mock() request.status_code = requests.codes.ok request.content = dumps(response_obj) mock_post.return_value = request mock_get.return_value = request mock_put.return_value = request obj = NotifyMatrix(host='host', include_image=True) assert isinstance(obj, NotifyMatrix) is True assert obj.access_token is None assert obj._register() is True assert obj.access_token is not None # Cause retries request.status_code = 429 request.content = dumps({ 'retry_after_ms': 1, }) code, response = obj._fetch('/retry/apprise/unit/test') assert code is False request.content = dumps({ 'error': { 'retry_after_ms': 1, } }) code, response = obj._fetch('/retry/apprise/unit/test') assert code is False request.content = dumps({ 'error': {} }) code, response = obj._fetch('/retry/apprise/unit/test') assert code is False del obj @mock.patch('requests.put') @mock.patch('requests.get') @mock.patch('requests.post') def test_plugin_matrix_auth(mock_post, mock_get, mock_put): """ NotifyMatrix() Server Authentication """ response_obj = { # Registration 'access_token': 'abcd1234', 'user_id': '@apprise:localhost', 'home_server': 'localhost', } # Default configuration request = mock.Mock() request.status_code = requests.codes.ok request.content = dumps(response_obj) mock_post.return_value = request mock_get.return_value = request mock_put.return_value = request obj = NotifyMatrix(host='localhost') assert isinstance(obj, NotifyMatrix) is True assert obj.access_token is None # logging out without an access_token is silently a success assert obj._logout() is True assert obj.access_token is None assert obj._register() is True assert obj.access_token is not None # Logging in is silently treated as a success because we # already had success registering assert obj._login() is True assert obj.access_token is not None # However if we log out assert obj._logout() is True assert obj.access_token is None # And set ourselves up for failure request.status_code = 403 assert obj._login() is False assert obj.access_token is None # Reset our token obj.access_token = None # Adjust our response to be invalid - missing access_token in response request.status_code = requests.codes.ok ro = response_obj.copy() del ro['access_token'] request.content = dumps(ro) # Our registration will fail now assert obj._register() is False assert obj.access_token is None del obj # So will login obj = NotifyMatrix(host='host', user='user', password='password') assert isinstance(obj, NotifyMatrix) is True assert obj._login() is False assert obj.access_token is None # Adjust our response to be invalid - invalid json response request.content = "{" # Our registration will fail now assert obj._register() is False assert obj.access_token is None request.status_code = requests.codes.ok request.content = dumps(response_obj) assert obj._register() is True assert obj.access_token is not None # Test logoff when getting a 403 error request.status_code = 403 assert obj._logout() is False assert obj.access_token is not None request.status_code = requests.codes.ok request.content = dumps(response_obj) assert obj._register() is True assert obj.access_token is not None request.status_code = 403 request.content = dumps({ u'errcode': u'M_UNKNOWN_TOKEN', u'error': u'Access Token unknown or expired', }) # Test logoff when getting a 403 error; but if we have the right error # code in the response, then we return a True assert obj._logout() is True assert obj.access_token is None del obj @mock.patch('requests.put') @mock.patch('requests.get') @mock.patch('requests.post') def test_plugin_matrix_rooms(mock_post, mock_get, mock_put): """ NotifyMatrix() Room Testing """ response_obj = { # Registration 'access_token': 'abcd1234', 'user_id': '@apprise:localhost', 'home_server': 'localhost', # For joined_room response 'joined_rooms': ['!abc123:localhost', '!def456:localhost'], # For room joining 'room_id': '!abc123:localhost', } # Default configuration request = mock.Mock() request.status_code = requests.codes.ok request.content = dumps(response_obj) mock_post.return_value = request mock_get.return_value = request mock_put.return_value = request obj = NotifyMatrix(host='host') assert isinstance(obj, NotifyMatrix) is True assert obj.access_token is None # Can't get room listing if we're not connnected assert obj._room_join('#abc123') is None assert obj._register() is True assert obj.access_token is not None assert obj._room_join('!abc123') == response_obj['room_id'] # Use cache to get same results assert obj.store.get('!abc123') is None # However this is how the cache entry gets stored assert obj.store.get('!abc123:localhost') is not None assert obj.store.get('!abc123:localhost')['id'] == response_obj['room_id'] assert obj._room_join('!abc123') == response_obj['room_id'] obj.store.clear() assert obj._room_join('!abc123:localhost') == response_obj['room_id'] assert obj.store.get('!abc123:localhost') is not None assert obj.store.get('!abc123:localhost')['id'] == response_obj['room_id'] # Use cache to get same results assert obj._room_join('!abc123:localhost') == response_obj['room_id'] obj.store.clear() assert obj._room_join('abc123') == response_obj['room_id'] # Use cache to get same results assert obj.store.get('#abc123:localhost') is not None assert obj.store.get('#abc123:localhost')['id'] == response_obj['room_id'] assert obj._room_join('abc123') == response_obj['room_id'] obj.store.clear() assert obj._room_join('abc123:localhost') == response_obj['room_id'] # Use cache to get same results assert obj.store.get('#abc123:localhost') is not None assert obj.store.get('#abc123:localhost')['id'] == response_obj['room_id'] assert obj._room_join('abc123:localhost') == response_obj['room_id'] obj.store.clear() assert obj._room_join('#abc123:localhost') == response_obj['room_id'] # Use cache to get same results assert obj.store.get('#abc123:localhost') is not None assert obj.store.get('#abc123:localhost')['id'] == response_obj['room_id'] assert obj._room_join('#abc123:localhost') == response_obj['room_id'] obj.store.clear() assert obj._room_join('%') is None assert obj._room_join(None) is None # 403 response; this will push for a room creation for alias based rooms # and these will fail request.status_code = 403 obj.store.clear() assert obj._room_join('!abc123') is None obj.store.clear() assert obj._room_join('!abc123:localhost') is None obj.store.clear() assert obj._room_join('abc123') is None obj.store.clear() assert obj._room_join('abc123:localhost') is None obj.store.clear() assert obj._room_join('#abc123:localhost') is None del obj # Room creation request.status_code = requests.codes.ok obj = NotifyMatrix(host='host') assert isinstance(obj, NotifyMatrix) is True assert obj.access_token is None # Can't get room listing if we're not connnected assert obj._room_create('#abc123') is None assert obj._register() is True assert obj.access_token is not None # You can't add room_id's, they must be aliases assert obj._room_create('!abc123') is None assert obj._room_create('!abc123:localhost') is None obj.store.clear() assert obj._room_create('abc123') == response_obj['room_id'] obj.store.clear() assert obj._room_create('abc123:localhost') == response_obj['room_id'] obj.store.clear() assert obj._room_create('#abc123:localhost') == response_obj['room_id'] obj.store.clear() assert obj._room_create('%') is None assert obj._room_create(None) is None # 403 response; this will push for a room creation for alias based rooms # and these will fail request.status_code = 403 obj.store.clear() assert obj._room_create('abc123') is None obj.store.clear() assert obj._room_create('abc123:localhost') is None obj.store.clear() assert obj._room_create('#abc123:localhost') is None request.status_code = 403 request.content = dumps({ u'errcode': u'M_ROOM_IN_USE', u'error': u'Room alias already taken', }) obj.store.clear() # This causes us to look up a channel ID if we get a ROOM_IN_USE response assert obj._room_create('#abc123:localhost') is None del obj # Room detection request.status_code = requests.codes.ok request.content = dumps(response_obj) obj = NotifyMatrix(host='localhost') assert isinstance(obj, NotifyMatrix) is True assert obj.access_token is None # No rooms if we're not connected response = obj._joined_rooms() assert isinstance(response, list) is True assert len(response) == 0 # register our account assert obj._register() is True assert obj.access_token is not None response = obj._joined_rooms() assert isinstance(response, list) is True assert len(response) == len(response_obj['joined_rooms']) for r in response: assert r in response_obj['joined_rooms'] request.status_code = 403 response = obj._joined_rooms() assert isinstance(response, list) is True assert len(response) == 0 del obj # Room id lookup request.status_code = requests.codes.ok obj = NotifyMatrix(host='localhost') assert isinstance(obj, NotifyMatrix) is True assert obj.access_token is None # Can't get room listing if we're not connnected assert obj._room_id('#abc123') is None assert obj._register() is True assert obj.access_token is not None # You can't add room_id's, they must be aliases assert obj._room_id('!abc123') is None assert obj._room_id('!abc123:localhost') is None obj.store.clear() assert obj._room_id('abc123') == response_obj['room_id'] obj.store.clear() assert obj._room_id('abc123:localhost') == response_obj['room_id'] obj.store.clear() assert obj._room_id('#abc123:localhost') == response_obj['room_id'] obj.store.clear() assert obj._room_id('%') is None assert obj._room_id(None) is None # If we can't look the code up, we return None request.status_code = 403 obj.store.clear() assert obj._room_id('#abc123:localhost') is None # Force a object removal (thus a logout call) del obj def test_plugin_matrix_url_parsing(): """ NotifyMatrix() URL Testing """ result = NotifyMatrix.parse_url( 'matrix://user:token@localhost?to=#room') assert isinstance(result, dict) is True assert len(result['targets']) == 1 assert '#room' in result['targets'] result = NotifyMatrix.parse_url( 'matrix://user:token@localhost?to=#room1,#room2,#room3') assert isinstance(result, dict) is True assert len(result['targets']) == 3 assert '#room1' in result['targets'] assert '#room2' in result['targets'] assert '#room3' in result['targets'] @mock.patch('requests.put') @mock.patch('requests.get') @mock.patch('requests.post') def test_plugin_matrix_image_errors(mock_post, mock_get, mock_put): """ NotifyMatrix() Image Error Handling """ def mock_function_handing(url, data, **kwargs): """ dummy function for handling image posts (as a failure) """ response_obj = { 'room_id': '!abc123:localhost', 'room_alias': '#abc123:localhost', 'joined_rooms': ['!abc123:localhost', '!def456:localhost'], 'access_token': 'abcd1234', 'home_server': 'localhost', } request = mock.Mock() request.content = dumps(response_obj) request.status_code = requests.codes.ok if 'm.image' in data: # Fail for images request.status_code = 400 return request # Prepare Mock mock_get.side_effect = mock_function_handing mock_post.side_effect = mock_function_handing mock_put.side_effect = mock_function_handing obj = NotifyMatrix(host='host', include_image=True, version='2') assert isinstance(obj, NotifyMatrix) is True assert obj.access_token is None # Notification was successful, however we could not post image and since # we had post errors (of any kind) we still report a failure. assert obj.notify('test', 'test') is False del obj obj = NotifyMatrix(host='host', include_image=False, version='2') assert isinstance(obj, NotifyMatrix) is True assert obj.access_token is None # We didn't post an image (which was set to fail) and therefore our # post was okay assert obj.notify('test', 'test') is True # Force a object removal (thus a logout call) del obj def mock_function_handing(url, data, **kwargs): """ dummy function for handling image posts (successfully) """ response_obj = { 'room_id': '!abc123:localhost', 'room_alias': '#abc123:localhost', 'joined_rooms': ['!abc123:localhost', '!def456:localhost'], 'access_token': 'abcd1234', 'home_server': 'localhost', } request = mock.Mock() request.content = dumps(response_obj) request.status_code = requests.codes.ok return request # Prepare Mock mock_get.side_effect = mock_function_handing mock_put.side_effect = mock_function_handing mock_post.side_effect = mock_function_handing obj = NotifyMatrix(host='host', include_image=True) assert isinstance(obj, NotifyMatrix) is True assert obj.access_token is None assert obj.notify('test', 'test') is True del obj obj = NotifyMatrix(host='host', include_image=False) assert isinstance(obj, NotifyMatrix) is True assert obj.access_token is None assert obj.notify('test', 'test') is True # Force a object removal (thus a logout call) del obj @mock.patch('requests.put') @mock.patch('requests.get') @mock.patch('requests.post') def test_plugin_matrix_attachments_api_v3(mock_post, mock_get, mock_put): """ NotifyMatrix() Attachment Checks (v3) """ # Prepare a good response response = mock.Mock() response.status_code = requests.codes.ok response.content = MATRIX_GOOD_RESPONSE.encode('utf-8') # Prepare a bad response bad_response = mock.Mock() bad_response.status_code = requests.codes.internal_server_error # Prepare Mock return object mock_post.return_value = response mock_get.return_value = response mock_put.return_value = response # Instantiate our object obj = Apprise.instantiate('matrix://user:pass@localhost/#general?v=3') # attach our content attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is True attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) # Test our call count assert mock_put.call_count == 1 assert mock_post.call_count == 2 assert mock_post.call_args_list[0][0][0] == \ 'http://localhost/_matrix/client/v3/login' assert mock_post.call_args_list[1][0][0] == \ 'http://localhost/_matrix/client/v3/join/%23general%3Alocalhost' assert mock_put.call_args_list[0][0][0] == \ 'http://localhost/_matrix/client/v3/rooms/%21abc123%3Alocalhost/' \ 'send/m.room.message/0' # Attach an unsupported file type (it's just skipped) attach = AppriseAttachment( os.path.join(TEST_VAR_DIR, 'apprise-archive.zip')) assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is True # An invalid attachment will cause a failure path = os.path.join(TEST_VAR_DIR, '/invalid/path/to/an/invalid/file.jpg') attach = AppriseAttachment(path) assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=path) is False # update our attachment to be valid attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) mock_post.return_value = None # Throw an exception on the first call to requests.post() for side_effect in (requests.RequestException(), OSError(), bad_response): mock_post.side_effect = [side_effect] # We'll never fail because files are not attached assert obj.send(body="test", attach=attach) is True # Throw an exception on the second call to requests.post() for side_effect in (requests.RequestException(), OSError(), bad_response): mock_post.side_effect = [response, side_effect] # Attachment support does not exist vor v3 at time, so this will # work nicely assert obj.send(body="test", attach=attach) is True # handle a bad response mock_post.side_effect = [response, bad_response, response] # Attachment support does not exist vor v3 at time, so this will # work nicely assert obj.send(body="test", attach=attach) is True # Force a object removal (thus a logout call) del obj @mock.patch('requests.get') @mock.patch('requests.post') def test_plugin_matrix_attachments_api_v2(mock_post, mock_get): """ NotifyMatrix() Attachment Checks (v2) """ # Prepare a good response response = mock.Mock() response.status_code = requests.codes.ok response.content = MATRIX_GOOD_RESPONSE.encode('utf-8') # Prepare a bad response bad_response = mock.Mock() bad_response.status_code = requests.codes.internal_server_error # Prepare Mock return object mock_post.return_value = response mock_get.return_value = response # Instantiate our object obj = Apprise.instantiate('matrix://user:pass@localhost/#general?v=2') # attach our content attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is True attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) # Attach an unsupported file mock_post.return_value = response mock_get.return_value = response mock_post.side_effect = None mock_get.side_effect = None # Force a object removal (thus a logout call) del obj # Instantiate our object obj = Apprise.instantiate('matrixs://user:pass@localhost/#general?v=2') # Reset our object mock_post.reset_mock() mock_get.reset_mock() assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is True # Test our call count assert mock_post.call_count == 5 assert mock_post.call_args_list[0][0][0] == \ 'https://localhost/_matrix/client/r0/login' assert mock_post.call_args_list[1][0][0] == \ 'https://localhost/_matrix/media/r0/upload' assert mock_post.call_args_list[2][0][0] == \ 'https://localhost/_matrix/client/r0/join/%23general%3Alocalhost' assert mock_post.call_args_list[3][0][0] == \ 'https://localhost/_matrix/client/r0/rooms/%21abc123%3Alocalhost/' \ 'send/m.room.message' assert mock_post.call_args_list[4][0][0] == \ 'https://localhost/_matrix/client/r0/rooms/%21abc123%3Alocalhost/' \ 'send/m.room.message' # Attach an unsupported file type; these are skipped attach = AppriseAttachment( os.path.join(TEST_VAR_DIR, 'apprise-archive.zip')) assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=attach) is True # An invalid attachment will cause a failure path = os.path.join(TEST_VAR_DIR, '/invalid/path/to/an/invalid/file.jpg') attach = AppriseAttachment(path) assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO, attach=path) is False # update our attachment to be valid attach = AppriseAttachment(os.path.join(TEST_VAR_DIR, 'apprise-test.gif')) mock_post.return_value = None mock_get.return_value = None # Throw an exception on the first call to requests.post() for side_effect in (requests.RequestException(), OSError(), bad_response): # Reset our value mock_post.reset_mock() mock_get.reset_mock() mock_post.side_effect = [side_effect, response] mock_get.side_effect = [side_effect, response] assert obj.send(body="test", attach=attach) is False # Throw an exception on the second call to requests.post() for side_effect in (requests.RequestException(), OSError(), bad_response): # Reset our value mock_post.reset_mock() mock_get.reset_mock() mock_post.side_effect = [response, side_effect, side_effect, response] mock_get.side_effect = [side_effect, side_effect, response] # We'll fail now because of our error handling assert obj.send(body="test", attach=attach) is False # handle a bad response mock_post.side_effect = \ [response, bad_response, response, response, response, response] mock_get.side_effect = \ [response, bad_response, response, response, response, response] # We'll fail now because of an internal exception assert obj.send(body="test", attach=attach) is False # Force a object removal (thus a logout call) del obj # Instantiate our object obj = Apprise.instantiate( 'matrixs://user:pass@localhost/#general?v=2&image=y') # Reset our object mock_post.reset_mock() mock_get.reset_mock() mock_post.return_value = None mock_get.return_value = None mock_post.side_effect = \ [response, response, bad_response, response, response, response, response] mock_get.side_effect = \ [response, response, bad_response, response, response, response, response] # image attachment didn't succeed assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO) is False # Error during image post mock_post.return_value = response mock_get.return_value = response mock_post.side_effect = None mock_get.side_effect = None # We'll fail now because of an internal exception assert obj.send(body="test", attach=attach) is True # Force __del__() call del obj @mock.patch('requests.put') @mock.patch('requests.get') @mock.patch('requests.post') def test_plugin_matrix_transaction_ids_api_v3_no_cache( mock_post, mock_get, mock_put): """ NotifyMatrix() Transaction ID Checks (v3) """ # Prepare a good response response = mock.Mock() response.status_code = requests.codes.ok response.content = MATRIX_GOOD_RESPONSE.encode('utf-8') # Prepare a bad response bad_response = mock.Mock() bad_response.status_code = requests.codes.internal_server_error # Prepare Mock return object mock_post.return_value = response mock_get.return_value = response mock_put.return_value = response # For each element is 1 batch that is ran # the number defined is the number of notifications to send batch = [10, 1, 5] for notifications in batch: # Instantiate our object obj = Apprise.instantiate('matrix://user:pass@localhost/#general?v=3') # Ensure mode is flush assert obj.store.mode == PersistentStoreMode.MEMORY # Performs a login assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO ) is True assert mock_get.call_count == 0 assert mock_post.call_count == 2 assert mock_post.call_args_list[0][0][0] == \ 'http://localhost/_matrix/client/v3/login' assert mock_post.call_args_list[1][0][0] == \ 'http://localhost/_matrix/client/v3/join/%23general%3Alocalhost' assert mock_put.call_count == 1 assert mock_put.call_args_list[0][0][0] == \ 'http://localhost/_matrix/client/v3/rooms/' + \ '%21abc123%3Alocalhost/send/m.room.message/0' for no, _ in enumerate(range(notifications), start=1): # Clean our slate mock_post.reset_mock() mock_get.reset_mock() mock_put.reset_mock() assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO ) is True assert mock_get.call_count == 0 assert mock_post.call_count == 0 assert mock_put.call_count == 1 assert mock_put.call_args_list[0][0][0] == \ 'http://localhost/_matrix/client/v3/rooms/' + \ f'%21abc123%3Alocalhost/send/m.room.message/{no}' mock_post.reset_mock() mock_get.reset_mock() mock_put.reset_mock() # Force a object removal (thus a logout call) del obj assert mock_get.call_count == 0 assert mock_post.call_count == 1 assert mock_post.call_args_list[0][0][0] == \ 'http://localhost/_matrix/client/v3/logout' mock_post.reset_mock() assert mock_put.call_count == 0 @mock.patch('requests.put') @mock.patch('requests.get') @mock.patch('requests.post') def test_plugin_matrix_transaction_ids_api_v3_w_cache( mock_post, mock_get, mock_put, tmpdir): """ NotifyMatrix() Transaction ID Checks (v3) """ # Prepare a good response response = mock.Mock() response.status_code = requests.codes.ok response.content = MATRIX_GOOD_RESPONSE.encode('utf-8') # Prepare a bad response bad_response = mock.Mock() bad_response.status_code = requests.codes.internal_server_error # Prepare Mock return object mock_post.return_value = response mock_get.return_value = response mock_put.return_value = response # For each element is 1 batch that is ran # the number defined is the number of notifications to send batch = [10, 1, 5] mock_post.reset_mock() mock_get.reset_mock() mock_put.reset_mock() asset = AppriseAsset( storage_mode=PersistentStoreMode.FLUSH, storage_path=str(tmpdir), ) # Message Counter transaction_id = 1 for no, notifications in enumerate(batch): # Instantiate our object obj = Apprise.instantiate( 'matrix://user:pass@localhost/#general?v=3', asset=asset) # Ensure mode is flush assert obj.store.mode == PersistentStoreMode.FLUSH # Performs a login assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO ) is True assert mock_get.call_count == 0 if no == 0: # first entry assert mock_post.call_count == 2 assert mock_post.call_args_list[0][0][0] == \ 'http://localhost/_matrix/client/v3/login' assert mock_post.call_args_list[1][0][0] == \ 'http://localhost/_matrix/client/v3/' \ 'join/%23general%3Alocalhost' assert mock_put.call_count == 1 assert mock_put.call_args_list[0][0][0] == \ 'http://localhost/_matrix/client/v3/rooms/' + \ '%21abc123%3Alocalhost/send/m.room.message/0' for no, _ in enumerate(range(notifications), start=transaction_id): # Clean our slate mock_post.reset_mock() mock_get.reset_mock() mock_put.reset_mock() assert obj.notify( body='body', title='title', notify_type=NotifyType.INFO ) is True # Increment transaction counter transaction_id += 1 assert mock_get.call_count == 0 assert mock_post.call_count == 0 assert mock_put.call_count == 1 assert mock_put.call_args_list[0][0][0] == \ 'http://localhost/_matrix/client/v3/rooms/' + \ f'%21abc123%3Alocalhost/send/m.room.message/{no}' # Increment transaction counter transaction_id += 1 mock_post.reset_mock() mock_get.reset_mock() mock_put.reset_mock() # Force a object removal # Biggest takeaway is that a logout no longer happens del obj assert mock_get.call_count == 0 assert mock_post.call_count == 0 assert mock_put.call_count == 0