118 lines
3.4 KiB
Python
118 lines
3.4 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 sat.core.sat_main import SAT
|
||
from sat.tools import async_trigger as trigger
|
||
from sat.core import xmpp
|
||
|
||
|
||
@fixture(scope="session")
|
||
def bridge():
|
||
bridge = AsyncMock()
|
||
bridge.addSignal = MagicMock()
|
||
bridge.addMethod = MagicMock()
|
||
return bridge
|
||
|
||
|
||
@fixture(scope="session")
|
||
def storage():
|
||
return AsyncMock()
|
||
|
||
|
||
class MockSAT(SAT):
|
||
|
||
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 registerNamespace
|
||
self.ns_map = {
|
||
"x-data": xmpp.NS_X_DATA,
|
||
"disco#info": xmpp.NS_DISCO_INFO,
|
||
}
|
||
self.memory = MagicMock()
|
||
self.memory.storage = storage
|
||
self.memory.getConfig.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._addBaseMenus()
|
||
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()
|
||
|
||
@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 = MockSAT(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
|