123 lines
3.7 KiB
Python
123 lines
3.7 KiB
Python
#!/usr/bin/env python3
|
|
|
|
# Libervia: an XMPP client
|
|
# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
|
|
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as published by
|
|
# the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
from contextlib import contextmanager
|
|
from unittest.mock import MagicMock, AsyncMock
|
|
from pytest import fixture
|
|
from twisted.internet import defer
|
|
from twisted.words.protocols.jabber import jid
|
|
from libervia.backend.core.main import LiberviaBackend
|
|
from libervia.backend.tools import async_trigger as trigger
|
|
from libervia.backend.core import xmpp
|
|
|
|
|
|
@fixture(scope="session")
|
|
def bridge():
|
|
bridge = AsyncMock()
|
|
bridge.add_signal = MagicMock()
|
|
bridge.add_method = MagicMock()
|
|
return bridge
|
|
|
|
|
|
@fixture(scope="session")
|
|
def storage():
|
|
return AsyncMock()
|
|
|
|
|
|
class MockLiberviaBackend(LiberviaBackend):
|
|
|
|
def __init__(self, bridge, storage):
|
|
self._cb_map = {}
|
|
self._menus = {}
|
|
self._menus_paths = {}
|
|
self._test_config = {}
|
|
self.profiles = {}
|
|
self.plugins = {}
|
|
# map for short name to whole namespace,
|
|
# extended by plugins with register_namespace
|
|
self.ns_map = {
|
|
"x-data": xmpp.NS_X_DATA,
|
|
"disco#info": xmpp.NS_DISCO_INFO,
|
|
}
|
|
self.memory = MagicMock()
|
|
self.memory.storage = storage
|
|
self.memory.config_get.side_effect = self.get_test_config
|
|
|
|
self.trigger = trigger.TriggerManager()
|
|
self.bridge = bridge
|
|
defer.ensureDeferred(self._post_init())
|
|
self.common_cache = AsyncMock()
|
|
self._import_plugins()
|
|
self._add_base_menus()
|
|
self.initialised = defer.Deferred()
|
|
self.initialised.callback(None)
|
|
|
|
def get_test_config(self, section, name, default=None):
|
|
return self._test_config.get((section or None, name), default)
|
|
|
|
def set_test_config(self, section, name, value):
|
|
self._test_config[(section or None, name)] = value
|
|
|
|
def clear_test_config(self):
|
|
self._test_config.clear()
|
|
|
|
def register_namespace(self, short_name, namespace):
|
|
# Plugin classes may be instantiated several times in tests, so we do nothing in
|
|
# this method to avoid ConflictError.
|
|
pass
|
|
|
|
@contextmanager
|
|
def use_option_and_reload(self, section, name, value):
|
|
self.set_test_config(section, name, value)
|
|
self.reload_plugins()
|
|
try:
|
|
yield self
|
|
finally:
|
|
self.clear_test_config()
|
|
self.reload_plugins()
|
|
|
|
def reload_plugins(self):
|
|
self.plugins.clear()
|
|
self.trigger._TriggerManager__triggers.clear()
|
|
self.ns_map = {
|
|
"x-data": xmpp.NS_X_DATA,
|
|
"disco#info": xmpp.NS_DISCO_INFO,
|
|
}
|
|
self._import_plugins()
|
|
|
|
def _init(self):
|
|
pass
|
|
|
|
async def _post_init(self):
|
|
pass
|
|
|
|
|
|
@fixture(scope="session")
|
|
def host(bridge, storage):
|
|
host = MockLiberviaBackend(bridge=bridge, storage=storage)
|
|
return host
|
|
|
|
|
|
@fixture
|
|
def client():
|
|
client = MagicMock()
|
|
client.jid = jid.JID("test_user@test.example/123")
|
|
client.server_jid = jid.JID("test.example")
|
|
client.pubsub_service = jid.JID("pubsub.test.example")
|
|
client.pubsub_client = AsyncMock()
|
|
return client
|