erebion/libervia-backend
Watch
1
0
Fork
You've already forked libervia-backend
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

tests (unit): Fix tests.

Browse source
This commit is contained in:
Goffi 2025-12-03 17:43:06 +01:00
commit e68278a1ac
12 changed files with 88 additions and 102 deletions
Download patch file Download diff file Expand all files Collapse all files

32
tests/unit/conftest.py
View file

@ -1,7 +1,7 @@
#!/usr/bin/env python3
# Libervia: an XMPP client
# Copyright (C) 2009-2021 Jérôme Poisson (goffi@goffi.org)
# Copyright (C) 2009-2025 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
@ -21,6 +21,8 @@ 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 import G
from wokkel.data_form import NS_X_DATA
from libervia.backend.core.main import LiberviaBackend
from libervia.backend.tools import async_trigger as trigger
from libervia.backend.core import xmpp
@ -42,6 +44,7 @@ def storage():
class MockLiberviaBackend(LiberviaBackend):
def __init__(self, bridge, storage):
G.set_host(self)
self._cb_map = {}
self._menus = {}
self._menus_paths = {}
@ -51,7 +54,7 @@ class MockLiberviaBackend(LiberviaBackend):
# map for short name to whole namespace,
# extended by plugins with register_namespace
self.ns_map = {
"x-data": xmpp.NS_X_DATA,
"x-data": NS_X_DATA,
"disco#info": xmpp.NS_DISCO_INFO,
}
self.memory = MagicMock()
@ -73,36 +76,11 @@ class MockLiberviaBackend(LiberviaBackend):
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

24
tests/unit/frontends/test_webrtc.py
View file

@ -24,6 +24,8 @@ try:
except ImportError:
pytest.skip("Gst not available.", allow_module_level=True)
from pytest_twisted import ensureDeferred as ed
from libervia.backend.core import exceptions
from libervia.backend.tools.common import data_format
from libervia.frontends.tools import webrtc as webrtc_mod
@ -95,7 +97,7 @@ class TestWebRtc:
assert webrtc.webrtcbin.emit.call_count == 1
assert webrtc.webrtcbin.emit.call_args[0][0] == "set-remote-description"
@pytest.mark.asyncio
@ed
async def test_answer_call(self, webrtc, monkeypatch):
"""The method correctly answers an incoming call."""
mock_setup_call = AsyncMock()
@ -128,12 +130,11 @@ class TestWebRtc:
mock_pad.has_current_caps.return_value = True
mock_pad.get_current_caps.return_value = mock_caps
mock_caps.__len__.return_value = 1
mock_caps.__getitem__.return_value = mock_structure
mock_caps.get_structure.return_value = mock_structure
mock_structure.get_name.return_value = "video/x-h264"
# We use non-standard resolution as example to trigger the workaround
mock_structure.get_int.side_effect = lambda x: MagicMock(
value=990 if x == "width" else 557
mock_structure.get_int = MagicMock(
side_effect=lambda x: (0, 990) if x == "width" else (0, 557)
)
webrtc.on_remote_decodebin_stream(None, mock_pad)
@ -153,8 +154,7 @@ class TestWebRtc:
mock_pad.has_current_caps.return_value = True
mock_pad.get_current_caps.return_value = mock_caps
mock_caps.__len__.return_value = 1
mock_caps.__getitem__.return_value = mock_structure
mock_caps.get_structure.return_value = mock_structure
mock_structure.get_name.return_value = "audio/x-raw"
webrtc.on_remote_decodebin_stream(None, mock_pad)
@ -163,7 +163,7 @@ class TestWebRtc:
mock_pad.link.assert_called()
@pytest.mark.skipif(Gst is None, reason="GStreamer is not available")
@pytest.mark.asyncio
@ed
async def test_setup_call_correct_role(self, host, webrtc, monkeypatch):
"""Roles are set in setup_call."""
monkeypatch.setattr(Gst, "parse_launch", MagicMock())
@ -181,7 +181,7 @@ class TestWebRtc:
with pytest.raises(AssertionError):
await webrtc.setup_call("invalid_role")
@pytest.mark.asyncio
@ed
async def test_setup_call_test_mode(self, host, webrtc, monkeypatch):
"""Test mode use fake video and audio in setup_call."""
monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
@ -190,7 +190,7 @@ class TestWebRtc:
assert "videotestsrc" in webrtc.gst_pipe_desc
assert "audiotestsrc" in webrtc.gst_pipe_desc
@pytest.mark.asyncio
@ed
async def test_setup_call_normal_mode(self, host, webrtc, monkeypatch):
"""Normal mode use real video and audio in setup_call."""
monkeypatch.setattr(data_format, "deserialise", MagicMock(return_value=[]))
@ -200,7 +200,7 @@ class TestWebRtc:
assert "pulsesrc" in webrtc.gst_pipe_desc
@pytest.mark.skipif(Gst is None, reason="GStreamer is not available")
@pytest.mark.asyncio
@ed
async def test_setup_call_with_stun_and_turn(self, host, webrtc, monkeypatch):
"""STUN and TURN server configurations are done in setup_call."""
mock_pipeline = MagicMock()
@ -244,7 +244,7 @@ class TestWebRtc:
)
@pytest.mark.skipif(Gst is None, reason="GStreamer is not available")
@pytest.mark.asyncio
@ed
async def test_setup_call_gstreamer_pipeline_failure(self, webrtc, monkeypatch):
"""Test setup_call method handling Gstreamer pipeline failure."""
monkeypatch.setattr(Gst, "parse_launch", lambda _: None)

39
tests/unit/test_ap-gateway.py
View file

@ -42,7 +42,7 @@ from libervia.backend.plugins import plugin_comp_ap_gateway
from libervia.backend.plugins.plugin_comp_ap_gateway import constants as ap_const
from libervia.backend.plugins.plugin_comp_ap_gateway import TYPE_ACTOR
from libervia.backend.plugins.plugin_comp_ap_gateway.http_server import HTTPServer
from libervia.backend.plugins.plugin_xep_0277 import NS_ATOM
from libervia.backend.plugins.plugin_xep_0277 import NS_ATOM, MbData
from libervia.backend.plugins.plugin_xep_0424 import NS_MESSAGE_RETRACT
from libervia.backend.plugins.plugin_xep_0465 import NS_PPS
from libervia.backend.tools import xml_tools
@ -1238,12 +1238,12 @@ class TestActivityPubGateway:
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
mb_data = {
"author_jid": TEST_JID.full(),
"content": f"mention of @{TEST_AP_ACCOUNT}",
"service": TEST_JID.full(),
"node": ap_gateway._m.namespace,
}
mb_data = MbData(
author_jid=TEST_JID,
content=f"mention of @{TEST_AP_ACCOUNT}",
service=TEST_JID,
node=ap_gateway._m.namespace,
)
ap_item = await ap_gateway.mb_data_2_ap_item(
ap_gateway.client, mb_data, public=True
)
@ -1266,12 +1266,12 @@ class TestActivityPubGateway:
monkeypatch.setattr(plugin_comp_ap_gateway.treq, "json_content", mock_treq_json)
monkeypatch.setattr(ap_gateway, "ap_get", mock_ap_get)
mb_data = {
"author_jid": TEST_JID.full(),
"content": f"mention of @{TEST_AP_ACCOUNT}",
"service": TEST_JID.full(),
"node": ap_gateway._m.namespace,
}
mb_data = MbData(
author_jid=TEST_JID,
content=f"mention of @{TEST_AP_ACCOUNT}",
service=TEST_JID,
node=ap_gateway._m.namespace,
)
ap_item = await ap_gateway.mb_data_2_ap_item(
ap_gateway.client, mb_data, public=False
)
@ -1448,20 +1448,17 @@ class TestActivityPubGateway:
assert cache_pubsub_items.called
# the microblog data put in cache correspond to the item sent to subscribers
__, __, __, [mb_data] = cache_pubsub_items.call_args.args
extra = mb_data["extra"]
assert "repeated" in extra
repeated = extra["repeated"]
assert (
repeated["by"]
== ap_gateway.get_local_jid_from_account(TEST_AP_ACCOUNT).full()
)
extra = mb_data.extra
repeated = extra.repeated
assert repeated is not None
assert repeated.by == ap_gateway.get_local_jid_from_account(TEST_AP_ACCOUNT)
xmpp_item_xmpp_url = xmpp_uri.build_xmpp_uri(
"pubsub",
path=TEST_JID.full(),
node=ap_gateway._m.namespace,
item=xmpp_item["id"],
)
assert repeated["uri"] == xmpp_item_xmpp_url
assert repeated.uri == xmpp_item_xmpp_url
@ed
async def test_xmpp_attachment_noticed_to_ap_like(self, ap_gateway, monkeypatch):

49
tests/unit/test_email_gateway.py
View file

@ -51,6 +51,19 @@ def email_gw(host):
return email_gw
TEST_CREDENTIALS_KW = {
"user_name": "User Name",
"smtp_host": "smtp.example.org",
"smtp_port": 587,
"smtp_username": "user",
"smtp_password": "password",
"imap_host": "imap.example.org",
"imap_port": 993,
"imap_username": "user",
"imap_password": "password",
}
class TestEmailGatewayComponent:
def test_jid_to_email_gateway_jid(self, email_gw):
@ -58,7 +71,7 @@ class TestEmailGatewayComponent:
client = MagicMock()
client.jid = jid.JID("gateway.example.org")
address_jid = jid.JID(r"user\40some-email-domain.example@gateway.example.org")
credentials = {"user_email": "user@example.org"}
credentials = Credentials(user_email="user@example.org", **TEST_CREDENTIALS_KW)
result = email_gw.jid_to_email(client, address_jid, credentials)
assert result == "user@some-email-domain.example"
@ -67,7 +80,7 @@ class TestEmailGatewayComponent:
client = MagicMock()
client.jid = jid.JID("gateway.example.org")
address_jid = jid.JID("external@example.org")
credentials = {"user_email": "user@example.org"}
credentials = Credentials(user_email="user@example.org", **TEST_CREDENTIALS_KW)
result = email_gw.jid_to_email(client, address_jid, credentials)
assert result == '"xmpp:external@example.org" <user@example.org>'
@ -129,7 +142,7 @@ class TestEmailGatewayComponent:
to_jid = jid.JID("gw-user@example.org")
user_email = "user@example.org"
user_data = UserData(Credentials({"user_email": user_email}))
user_data = UserData(Credentials(user_email=user_email, **TEST_CREDENTIALS_KW))
await email_gw.on_new_email(user_data, to_jid, email)
client.sendMessage.assert_called_once()
@ -138,7 +151,7 @@ class TestEmailGatewayComponent:
assert call_args[1] == {"": "Hello, world!\n"}
assert call_args[2] == None
client.sendMessage.assert_called_once_with(
to_jid, {"": "Hello, world!\n"}, None, extra={}
to_jid, {"": "Hello, world!\n"}, None, extra={}, uid=None
)
@ed
@ -159,7 +172,7 @@ class TestEmailGatewayComponent:
user_jid = jid.JID("gw-user@example.org")
user_email = "user@example.org"
user_data = UserData(Credentials({"user_email": user_email}))
user_data = UserData(Credentials(user_email=user_email, **TEST_CREDENTIALS_KW))
await email_gw.on_new_email(user_data, user_jid, email)
client.sendMessage.assert_called_once_with(
@ -183,6 +196,7 @@ class TestEmailGatewayComponent:
],
}
},
uid=None
)
@ed
@ -192,14 +206,10 @@ class TestEmailGatewayComponent:
email_gw.client.jid = jid.JID("gateway.example.org")
email_gw.storage = MagicMock()
email_gw.storage.get = AsyncMock(
return_value={
"user_email": "user@example.org",
"user_name": "Sender Name",
"smtp_host": "smtp.example.org",
"smtp_port": "587",
"smtp_username": "sender",
"smtp_password": "password",
}
return_value=Credentials(
user_email="user@example.org",
**TEST_CREDENTIALS_KW
)
)
from_jid = jid.JID("user@example.org")
@ -248,7 +258,7 @@ class TestEmailGatewayComponent:
to_jid = jid.JID("gw-user@example.org")
user_email = "user@example.org"
user_data = UserData(Credentials({"user_email": user_email}))
user_data = UserData(Credentials(user_email=user_email, **TEST_CREDENTIALS_KW))
await email_gw.on_new_email(user_data, to_jid, email)
client.sendMessage.assert_called_once_with(
@ -256,6 +266,7 @@ class TestEmailGatewayComponent:
{"": "Hello, world!\n"},
None,
extra={"headers": {"keywords": "test, example", "urgency": "high"}},
uid=None
)
@ -271,13 +282,15 @@ class TestPubsubService:
def pubsub_resource(self, pubsub_service):
return pubsub_service.resource
def test_getNodes(self, pubsub_resource, client):
@ed
async def test_getNodes(self, pubsub_resource, client):
"""XEP-0498 well-known node is returned."""
requestor = client.jid
service = client.pubsub_service
nodeIdentifier = "test_node"
result = pubsub_resource.getNodes(requestor, service, nodeIdentifier)
assert result.result == [pubsub_resource._pfs.namespace]
# node_id must be None to get the nodes.
node_id = None
result = await pubsub_resource.getNodes(requestor, service, node_id)
assert result == [pubsub_resource._pfs.namespace]
@ed
async def test_items(self, pubsub_resource, client, host):

4
tests/unit/test_plugin_xep_0176.py
View file

@ -225,7 +225,7 @@ class TestXEP0176:
transport_elt = await xep_0176.jingle_session_init(client, session, content_name)
expected_transport_elt = xep_0176.build_transport(
session["contents"][content_name]["transport_data"]["local_ice_data"]
session["contents"][content_name]["transport_data"]["local_ice_data"], add_trickle=True
)
assert transport_elt.toXml() == expected_transport_elt.toXml()
@ -245,7 +245,7 @@ class TestXEP0176:
)
expected_transport_elt = xep_0176.build_transport(
session["contents"][content_name]["transport_data"]["local_ice_data"]
session["contents"][content_name]["transport_data"]["local_ice_data"], add_trickle=True
)
assert returned_transport_elt.toXml() == expected_transport_elt.toXml()

9
tests/unit/test_pubsub-cache.py
View file

@ -24,7 +24,6 @@ from libervia.backend.core.constants import Const as C
class TestPubsubCache:
@ed
async def test_cache_is_used_transparently(self, host, client):
"""Cache is used when a pubsub get_items operation is done"""
@ -61,12 +60,16 @@ class TestPubsubCache:
@ed
async def test_cache_is_not_used_when_no_cache(self, host, client):
"""Cache is skipped when 'pubsub_cache_strategy' is set to 'no_cache'"""
with host.use_option_and_reload(None, "pubsub_cache_strategy", "no_cache"):
with patch("libervia.backend.plugins.plugin_pubsub_cache.G") as mock_G:
mock_G.config.common.pubsub_cache_strategy = "no_cache"
items_ret = defer.Deferred()
items_ret.callback(([], {}))
client.pubsub_client.items = MagicMock(return_value=items_ret)
host.memory.storage.get_pubsub_node.return_value = None
host.memory.storage.set_pubsub_node.return_value = PubsubNode(sync_state=None)
host.memory.storage.set_pubsub_node.return_value = PubsubNode(
sync_state=None
)
with patch.object(host.plugins["PUBSUB_CACHE"], "cache_node") as cache_node:
await host.plugins["XEP-0060"].get_items(
client,