apprise/test/test_plugin_matrix.py

# -*- 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

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