github
/
apprise
mirror of https://github.com/caronc/apprise
1
0
Fork 0
Code Issues Releases Wiki Activity
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
898 Commits
25 Branches
56 Tags
29 MiB
 
 
 
Tree: 78f16adbb9
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from '78f16adbb9'
${ noResults }
apprise/test/test_plugin_matrix.py

1554 lines
51 KiB
Raw Blame History

# -*- coding: utf-8 -*-
# BSD 2-Clause License
#
# Apprise - Push Notification Library.
# Copyright (c) 2024, Chris Caron <lead2gold@gmail.com>
#
# 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, loads
from apprise.plugins.matrix import NotifyMatrix
from apprise.plugins.matrix import MatrixDiscoveryException
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',
# Simulate .well-known
"m.homeserver": {
"base_url": "https://matrix.example.com"
},
"m.identity_server": {
"base_url": "https://vector.im"
},
})
# 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_discovery_service(mock_post, mock_get):
"""
NotifyMatrix() Discovery Service
"""
# Prepare a good response
response = mock.Mock()
response.status_code = requests.codes.ok
response.content = MATRIX_GOOD_RESPONSE.encode('utf-8')
# Prepare a good response
bad_response = mock.Mock()
bad_response.status_code = requests.codes.unauthorized
bad_response.content = MATRIX_GOOD_RESPONSE.encode('utf-8')
# Prepare Mock return object
mock_post.return_value = response
mock_get.return_value = response
# Instantiate our object
obj = Apprise.instantiate(
'matrixs://user:pass@example.com/#general?v=2&discovery=yes')
assert obj.notify('body') is True
response = mock.Mock()
response.status_code = requests.codes.unavailable
_resp = loads(MATRIX_GOOD_RESPONSE)
mock_get.return_value = response
mock_post.return_value = response
obj = Apprise.instantiate(
'matrixs://user:pass@example.com/#general?v=2&discovery=yes')
obj.store.clear(
NotifyMatrix.discovery_base_key, NotifyMatrix.discovery_identity_key)
# Invalid host / fallback is to resolve our own host
with pytest.raises(MatrixDiscoveryException):
obj.base_url
# Verify cache is not saved
assert NotifyMatrix.discovery_base_key not in obj.store
assert NotifyMatrix.discovery_identity_key not in obj.store
response.status_code = requests.codes.ok
obj.store.clear(
NotifyMatrix.discovery_base_key, NotifyMatrix.discovery_identity_key)
# bad data
_resp['m.homeserver'] = '!garbage!:303'
response.content = dumps(_resp).encode('utf-8')
obj.store.clear(
NotifyMatrix.discovery_base_key, NotifyMatrix.discovery_identity_key)
with pytest.raises(MatrixDiscoveryException):
obj.base_url
# Verify cache is not saved
assert NotifyMatrix.discovery_base_key not in obj.store
assert NotifyMatrix.discovery_identity_key not in obj.store
# We fail our discovery and therefore can't send our notification
assert obj.notify('hello world') is False
# bad key
_resp['m.homeserver'] = {}
response.content = dumps(_resp).encode('utf-8')
obj.store.clear(
NotifyMatrix.discovery_base_key, NotifyMatrix.discovery_identity_key)
with pytest.raises(MatrixDiscoveryException):
obj.base_url
# Verify cache is not saved
assert NotifyMatrix.discovery_base_key not in obj.store
assert NotifyMatrix.discovery_identity_key not in obj.store
_resp['m.homeserver'] = {'base_url': 'https://nuxref.com/base'}
response.content = dumps(_resp).encode('utf-8')
obj.store.clear(
NotifyMatrix.discovery_base_key, NotifyMatrix.discovery_identity_key)
assert obj.base_url == 'https://nuxref.com/base'
assert obj.identity_url == "https://vector.im"
# Verify cache saved
assert NotifyMatrix.discovery_base_key in obj.store
assert NotifyMatrix.discovery_identity_key in obj.store
# Discovery passes so notifications work too
assert obj.notify('hello world') is True
# bad data
_resp['m.identity_server'] = '!garbage!:303'
response.content = dumps(_resp).encode('utf-8')
obj.store.clear(
NotifyMatrix.discovery_base_key, NotifyMatrix.discovery_identity_key)
with pytest.raises(MatrixDiscoveryException):
obj.base_url
# Verify cache is not saved
assert NotifyMatrix.discovery_base_key not in obj.store
assert NotifyMatrix.discovery_identity_key not in obj.store
# no key
_resp['m.identity_server'] = {}
response.content = dumps(_resp).encode('utf-8')
obj.store.clear(
NotifyMatrix.discovery_base_key, NotifyMatrix.discovery_identity_key)
with pytest.raises(MatrixDiscoveryException):
obj.base_url
# Verify cache is not saved
assert NotifyMatrix.discovery_base_key not in obj.store
assert NotifyMatrix.discovery_identity_key not in obj.store
# remove
del _resp['m.identity_server']
response.content = dumps(_resp).encode('utf-8')
obj.store.clear(
NotifyMatrix.discovery_base_key, NotifyMatrix.discovery_identity_key)
assert obj.base_url == 'https://nuxref.com/base'
assert obj.identity_url == 'https://nuxref.com/base'
# restore
_resp['m.identity_server'] = {'base_url': '"https://vector.im'}
response.content = dumps(_resp).encode('utf-8')
# Not found is an acceptable response (no exceptions thrown)
response.status_code = requests.codes.not_found
obj.store.clear(
NotifyMatrix.discovery_base_key, NotifyMatrix.discovery_identity_key)
assert obj.base_url == 'https://example.com'
assert obj.identity_url == 'https://example.com'
# Verify cache saved
assert NotifyMatrix.discovery_base_key in obj.store
assert NotifyMatrix.discovery_identity_key in obj.store
# Discovery passes so notifications work too
response.status_code = requests.codes.ok
assert obj.notify('hello world') is True
response.status_code = requests.codes.ok
mock_get.return_value = None
mock_get.side_effect = (response, bad_response)
obj.store.clear(
NotifyMatrix.discovery_base_key, NotifyMatrix.discovery_identity_key)
with pytest.raises(MatrixDiscoveryException):
obj.base_url
# Verify cache is not saved
assert NotifyMatrix.discovery_base_key not in obj.store
assert NotifyMatrix.discovery_identity_key not in obj.store
# Test case where ourIdentity URI fails to do it's check
mock_get.side_effect = (response, response, bad_response)
obj.store.clear(
NotifyMatrix.discovery_base_key, NotifyMatrix.discovery_identity_key)
with pytest.raises(MatrixDiscoveryException):
obj.base_url
# Verify cache is not saved
assert NotifyMatrix.discovery_base_key not in obj.store
assert NotifyMatrix.discovery_identity_key not in obj.store
# Test an empty block response
response.status_code = requests.codes.ok
response.content = ''
mock_get.return_value = response
mock_get.side_effect = None
mock_post.return_value = response
mock_post.side_effect = None
obj.store.clear(
NotifyMatrix.discovery_base_key, NotifyMatrix.discovery_identity_key)
assert obj.base_url == 'https://example.com'
assert obj.identity_url == 'https://example.com'
# Verify cache saved
assert NotifyMatrix.discovery_base_key in obj.store
assert NotifyMatrix.discovery_identity_key in obj.store
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://matrix.example.com/_matrix/client/r0/login'
assert mock_post.call_args_list[1][0][0] == \
'https://matrix.example.com/_matrix/media/r0/upload'
assert mock_post.call_args_list[2][0][0] == \
'https://matrix.example.com/_matrix/client/r0/' \
'join/%23general%3Alocalhost'
assert mock_post.call_args_list[3][0][0] == \
'https://matrix.example.com/_matrix/client/r0' \
'/rooms/%21abc123%3Alocalhost/send/m.room.message'
assert mock_post.call_args_list[4][0][0] == \
'https://matrix.example.com/_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 (no discovery required)
obj = Apprise.instantiate(
'matrixs://user:pass@localhost/#general?v=2&discovery=no&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