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 + black reformatting.

Browse source
This commit is contained in:
Goffi 2024-07-14 17:42:53 +02:00
commit 067d8f32bf
11 changed files with 229 additions and 194 deletions
Download patch file Download diff file Expand all files Collapse all files

2
tests/unit/conftest.py
View file

@ -48,7 +48,7 @@ class MockLiberviaBackend(LiberviaBackend):
self._test_config = {}
self.profiles = {}
self.plugins = {}
# map for short name to whole namespace,
# map for short name to whole namespace,
# extended by plugins with register_namespace
self.ns_map = {
"x-data": xmpp.NS_X_DATA,

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

@ -29,7 +29,6 @@ from libervia.backend.tools.common import data_format
from libervia.frontends.tools import webrtc as webrtc_mod
@pytest.fixture
def host(monkeypatch):
host = MagicMock()
@ -37,6 +36,7 @@ def host(monkeypatch):
host.app.expand = lambda s: s
return host
@pytest.fixture(scope="function")
def webrtc(host):
"""Fixture for WebRTC instantiation."""
@ -185,7 +185,7 @@ class TestWebRtc:
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=[]))
monkeypatch.setattr(webrtc, "sources", webrtc_mod.SINKS_TEST)
monkeypatch.setattr(webrtc, "sources_data", webrtc_mod.SourcesTest())
await webrtc.setup_call("initiator")
assert "videotestsrc" in webrtc.gst_pipe_desc
assert "audiotestsrc" in webrtc.gst_pipe_desc
@ -194,7 +194,7 @@ class TestWebRtc:
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=[]))
monkeypatch.setattr(webrtc, "sources", webrtc_mod.SOURCES_AUTO)
monkeypatch.setattr(webrtc, "sources_data", webrtc_mod.SourcesAuto())
await webrtc.setup_call("initiator")
assert "v4l2src" in webrtc.gst_pipe_desc
assert "pulsesrc" in webrtc.gst_pipe_desc

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

@ -718,8 +718,7 @@ class TestActivityPubGateway:
first_item["id"] == "https://test.example/_ap/item/some_user@test.example/4"
)
assert (
first_item["actor"]
== "https://test.example/_ap/actor/some_user@test.example"
first_item["actor"] == "https://test.example/_ap/actor/some_user@test.example"
)
assert first_item["type"] == "Create"
first_item_obj = first_item["object"]
@ -803,8 +802,7 @@ class TestActivityPubGateway:
)
assert following["@context"] == ["https://www.w3.org/ns/activitystreams"]
assert (
following["id"]
== "https://test.example/_ap/following/some_user@test.example"
following["id"] == "https://test.example/_ap/following/some_user@test.example"
)
assert following["totalItems"] == len(subscriptions)
assert following["type"] == "OrderedCollection"
@ -841,8 +839,7 @@ class TestActivityPubGateway:
)
assert followers["@context"] == ["https://www.w3.org/ns/activitystreams"]
assert (
followers["id"]
== "https://test.example/_ap/followers/some_user@test.example"
followers["id"] == "https://test.example/_ap/followers/some_user@test.example"
)
assert followers["totalItems"] == len(subscribers)
assert followers["type"] == "OrderedCollection"
@ -1398,7 +1395,9 @@ class TestActivityPubGateway:
assert ap_post.called
url, actor_id, doc = ap_post.call_args.args
assert url == TEST_USER_DATA["endpoints"]["sharedInbox"]
assert actor_id == ap_gateway.build_apurl(ap_const.TYPE_ACTOR, TEST_JID.userhost())
assert actor_id == ap_gateway.build_apurl(
ap_const.TYPE_ACTOR, TEST_JID.userhost()
)
assert doc["type"] == "Announce"
assert ap_const.NS_AP_PUBLIC in doc["to"]
assert doc["object"] == ap_item["id"]
@ -1434,8 +1433,14 @@ class TestActivityPubGateway:
ap_gateway.host.memory.storage, "cache_pubsub_items"
) as cache_pubsub_items:
await ap_gateway.server.resource.handle_announce_activity(
TEST_AP_ACTOR_ID, Request(MagicMock()), announce, None, None, None,
"", TEST_AP_ACTOR_ID
TEST_AP_ACTOR_ID,
Request(MagicMock()),
announce,
None,
None,
None,
"",
TEST_AP_ACTOR_ID,
)
assert cache_pubsub_items.called
@ -1444,7 +1449,10 @@ class TestActivityPubGateway:
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()
assert (
repeated["by"]
== ap_gateway.get_local_jid_from_account(TEST_AP_ACCOUNT).full()
)
xmpp_item_xmpp_url = xmpp_uri.build_xmpp_uri(
"pubsub",
path=TEST_JID.full(),
@ -1487,7 +1495,9 @@ class TestActivityPubGateway:
assert ap_post.called
url, actor_id, doc = ap_post.call_args.args
assert url == TEST_USER_DATA["endpoints"]["sharedInbox"]
assert actor_id == ap_gateway.build_apurl(ap_const.TYPE_ACTOR, TEST_JID.userhost())
assert actor_id == ap_gateway.build_apurl(
ap_const.TYPE_ACTOR, TEST_JID.userhost()
)
assert doc["type"] == "Like"
assert ap_const.NS_AP_PUBLIC in doc["cc"]
assert doc["object"] == ap_item["id"]
@ -1748,9 +1758,9 @@ class TestActivityPubGateway:
assert event_data["head-picture"] == {
"sources": [{"url": "https://example.org/head_picture.jpg"}]
}
assert event_data["descriptions"] == [{
"description": "<p>meeting for test</p>", "type": "xhtml"
}]
assert event_data["descriptions"] == [
{"description": "<p>meeting for test</p>", "type": "xhtml"}
]
assert event_data["categories"] == [{"term": "test"}]
assert event_data["locations"] == [{"description": "somewhere in the world"}]
assert event_data["rsvp"] == [

40
tests/unit/test_plugin_xep_0082.py
View file

@ -30,7 +30,7 @@ __all__ = [ # pylint: disable=unused-variable
"test_datetime_formatting",
"test_datetime_parsing",
"test_time_formatting",
"test_time_parsing"
"test_time_parsing",
]
@ -85,9 +85,10 @@ def test_datetime_formatting() -> None: # pylint: disable=missing-function-docs
# explicit input
# datetime
assert isinstance(XEP_0082.parse_datetime(XEP_0082.format_datetime()), datetime)
assert isinstance(XEP_0082.parse_datetime(XEP_0082.format_datetime(
include_microsecond=True
)), datetime)
assert isinstance(
XEP_0082.parse_datetime(XEP_0082.format_datetime(include_microsecond=True)),
datetime,
)
assert XEP_0082.parse_datetime(XEP_0082.format_datetime()).tzinfo == timezone.utc
@ -109,16 +110,19 @@ def test_datetime_parsing() -> None: # pylint: disable=missing-function-docstri
XEP_0082.parse_datetime("1969-07-21T02:56:15")
# With timezone 'Z', with a fraction of a second consisting of two digits
assert XEP_0082.parse_datetime("1969-07-21T02:56:15.12Z") == \
value.replace(microsecond=120000)
assert XEP_0082.parse_datetime("1969-07-21T02:56:15.12Z") == value.replace(
microsecond=120000
)
# With timezone 'Z', with a fraction of a second consisting of nine digits
assert XEP_0082.parse_datetime("1969-07-21T02:56:15.123456789Z") == \
value.replace(microsecond=123456)
assert XEP_0082.parse_datetime("1969-07-21T02:56:15.123456789Z") == value.replace(
microsecond=123456
)
# With timezone '+04:00', with a fraction of a second consisting of six digits
assert XEP_0082.parse_datetime("1969-07-21T06:56:15.123456+04:00") == \
value.replace(microsecond=123456)
assert XEP_0082.parse_datetime("1969-07-21T06:56:15.123456+04:00") == value.replace(
microsecond=123456
)
# With timezone '-05:00', with a fraction of a second consisting of zero digits
assert XEP_0082.parse_datetime("1969-07-20T21:56:15.-05:00") == value
@ -162,9 +166,9 @@ def test_time_formatting() -> None: # pylint: disable=missing-function-docstrin
# Check that a time instance is returned when format_time is called without an
# explicit input date
assert isinstance(XEP_0082.parse_time(XEP_0082.format_time()), time)
assert isinstance(XEP_0082.parse_time(XEP_0082.format_time(
include_microsecond=True
)), time)
assert isinstance(
XEP_0082.parse_time(XEP_0082.format_time(include_microsecond=True)), time
)
assert XEP_0082.parse_time(XEP_0082.format_time()).tzinfo == timezone.utc
@ -191,12 +195,14 @@ def test_time_parsing() -> None: # pylint: disable=missing-function-docstring
assert XEP_0082.parse_time("16:00:00.123456789Z") == value.replace(microsecond=123456)
# With timezone '+04:00', with a fraction of a second consisting of six digits
assert XEP_0082.parse_time("20:00:00.123456+04:00") == \
value.replace(microsecond=123456)
assert XEP_0082.parse_time("20:00:00.123456+04:00") == value.replace(
microsecond=123456
)
# With timezone '-05:00', with a fraction of a second consisting of zero digits
assert XEP_0082.parse_time("11:00:00.-05:00") == value
# Without a timezone, with a fraction of a second consisting of six digits
assert XEP_0082.parse_time("16:00:00.123456") == \
value.replace(microsecond=123456, tzinfo=None)
assert XEP_0082.parse_time("16:00:00.123456") == value.replace(
microsecond=123456, tzinfo=None
)

30
tests/unit/test_plugin_xep_0167.py
View file

@ -25,7 +25,10 @@ from twisted.words.protocols.jabber import jid
from libervia.backend.plugins.plugin_xep_0166 import XEP_0166
from libervia.backend.plugins.plugin_xep_0167 import XEP_0167, mapping
from libervia.backend.plugins.plugin_xep_0167.constants import NS_JINGLE_RTP, NS_JINGLE_RTP_INFO
from libervia.backend.plugins.plugin_xep_0167.constants import (
NS_JINGLE_RTP,
NS_JINGLE_RTP_INFO,
)
from libervia.backend.tools import xml_tools
from libervia.backend.tools.common import data_format
@ -45,8 +48,8 @@ class TestXEP0167Mapping:
"""Senders are mapped to SDP attribute"""
assert mapping.senders_to_sdp("both", {"role": "initiator"}) == "a=sendrecv"
assert mapping.senders_to_sdp("none", {"role": "initiator"}) == "a=inactive"
assert mapping.senders_to_sdp("initiator", {"role": "initiator"}) == "a=sendonly"
assert mapping.senders_to_sdp("responder", {"role": "initiator"}) == "a=recvonly"
assert mapping.senders_to_sdp("initiator", {"role": "initiator"}) == "a=recvonly"
assert mapping.senders_to_sdp("responder", {"role": "initiator"}) == "a=sendonly"
def test_generate_sdp_from_session(self):
"""SDP is correctly generated from session data"""
@ -134,35 +137,36 @@ class TestXEP0167Mapping:
"media": "audio",
"payload_types": {
96: {
"clockrate": 48000,
"id": 96,
"name": "opus",
"clockrate": 48000,
"parameters": {"sprop-stereo": "1"},
}
},
},
"id": "audio",
"senders": "both",
"transport_data": {
"port": 9999,
"pwd": "pwd",
"ufrag": "ufrag",
"candidates": [
{
"foundation": "1",
"component_id": 1,
"transport": "UDP",
"priority": 1,
"address": "10.0.0.1",
"component_id": 1,
"foundation": "1",
"port": 12345,
"priority": 1,
"transport": "UDP",
"type": "host",
}
],
"port": 9999,
"pwd": "pwd",
"ufrag": "ufrag",
},
"id": "audio",
},
"metadata": {},
}
assert mapping.parse_sdp(sdp) == expected_session
assert mapping.parse_sdp(sdp, "responder") == expected_session
def test_build_description(self):
"""<description> element is generated from media data"""

14
tests/unit/test_plugin_xep_0272.py
View file

@ -35,7 +35,6 @@ from libervia.backend.tools.common import data_format
TEST_ROOM_JID = jid.JID("room@example.com/user")
@fixture
def xep_0272(host) -> XEP_0272:
"""Fixture for initializing XEP-0272 plugin."""
@ -64,11 +63,8 @@ class TestXEP0272:
muji_elt = presence_elt.addElement((NS_MUJI, "muji"))
muji_elt.addElement("preparing")
muji_data = {
"preparing_jids": set()
}
with patch.object(
xep_0272._muc, "get_room_user_jid", return_value=TEST_ROOM_JID) :
muji_data = {"preparing_jids": set()}
with patch.object(xep_0272._muc, "get_room_user_jid", return_value=TEST_ROOM_JID):
with patch.object(xep_0272, "get_muji_data", return_value=muji_data):
with patch.object(
xep_0272, "try_to_finish_preparation"
@ -83,11 +79,7 @@ class TestXEP0272:
presence_elt["from"] = "room@example.com/user"
presence_elt.addElement((NS_MUJI, "muji"))
muji_data = {
"done_preparing": True,
"preparing_jids": set(),
"to_call": set()
}
muji_data = {"done_preparing": True, "preparing_jids": set(), "to_call": set()}
with patch.object(xep_0272, "get_muji_data", return_value=muji_data):
with patch.object(
xep_0272, "try_to_finish_preparation"

36
tests/unit/test_plugin_xep_0293.py
View file

@ -19,9 +19,14 @@
from twisted.words.xish import domish
from libervia.backend.plugins.plugin_xep_0167.constants import NS_JINGLE_RTP
from libervia.backend.plugins.plugin_xep_0293 import NS_JINGLE_RTP_RTCP_FB, RTCP_FB_KEY, XEP_0293
from libervia.backend.plugins.plugin_xep_0293 import (
NS_JINGLE_RTP_RTCP_FB,
RTCP_FB_KEY,
XEP_0293,
)
from libervia.backend.tools import xml_tools
class TestXEP0293:
def test_parse_sdp_rtcp_fb_general(self, host):
@ -49,9 +54,7 @@ class TestXEP0293:
"""Parsing of a payload-specific RTCP feedback SDP line."""
xep_0293 = XEP_0293(host)
application_data = {
"payload_types": {96: {}}
}
application_data = {"payload_types": {96: {}}}
transport_data = {}
# SDP line: a=rtcp-fb:96 nack
@ -93,9 +96,7 @@ class TestXEP0293:
"""Parsing of a payload-specific RTCP feedback with trr-int SDP line."""
xep_0293 = XEP_0293(host)
application_data = {
"payload_types": {96: {}}
}
application_data = {"payload_types": {96: {}}}
transport_data = {}
# SDP line: a=rtcp-fb-trr-int:96 100
@ -139,7 +140,9 @@ class TestXEP0293:
sdp_lines = []
application_data = {}
media_data = {
"payload_types": {96: {RTCP_FB_KEY: [("nack", None, {})], "rtcp-fb-trr-int": 100}}
"payload_types": {
96: {RTCP_FB_KEY: [("nack", None, {})], "rtcp-fb-trr-int": 100}
}
}
xep_0293._generate_sdp_content_trigger(
@ -204,10 +207,7 @@ class TestXEP0293:
def test_build_rtcp_fb_elements(self, host):
"""Building the <rtcp-fb> and <rtcp-fb-trr-int> elements."""
xep_0293 = XEP_0293(host)
data = {
RTCP_FB_KEY: [("nack", "pli", {})],
"rtcp-fb-trr-int": 100
}
data = {RTCP_FB_KEY: [("nack", "pli", {})], "rtcp-fb-trr-int": 100}
# Test _build_description_trigger
desc_elt = domish.Element((NS_JINGLE_RTP, "description"))
@ -219,7 +219,9 @@ class TestXEP0293:
assert rtcp_fb_elt["type"] == "nack"
assert rtcp_fb_elt["subtype"] == "pli"
rtcp_fb_trr_int_elts = list(desc_elt.elements(NS_JINGLE_RTP_RTCP_FB, "rtcp-fb-trr-int"))
rtcp_fb_trr_int_elts = list(
desc_elt.elements(NS_JINGLE_RTP_RTCP_FB, "rtcp-fb-trr-int")
)
assert len(rtcp_fb_trr_int_elts) == 1
rtcp_fb_trr_int_elt = rtcp_fb_trr_int_elts[0]
assert int(rtcp_fb_trr_int_elt["value"]) == 100
@ -227,7 +229,9 @@ class TestXEP0293:
# Test _build_description_payload_type_trigger
desc_elt = domish.Element((NS_JINGLE_RTP, "description"))
payload_type_elt = desc_elt.addElement((NS_JINGLE_RTP, "payload-type"))
xep_0293._build_description_payload_type_trigger(desc_elt, {}, data, payload_type_elt)
xep_0293._build_description_payload_type_trigger(
desc_elt, {}, data, payload_type_elt
)
rtcp_fb_elts = list(payload_type_elt.elements(NS_JINGLE_RTP_RTCP_FB, "rtcp-fb"))
assert len(rtcp_fb_elts) == 1
@ -235,7 +239,9 @@ class TestXEP0293:
assert rtcp_fb_elt["type"] == "nack"
assert rtcp_fb_elt["subtype"] == "pli"
rtcp_fb_trr_int_elts = list(payload_type_elt.elements(NS_JINGLE_RTP_RTCP_FB, "rtcp-fb-trr-int"))
rtcp_fb_trr_int_elts = list(
payload_type_elt.elements(NS_JINGLE_RTP_RTCP_FB, "rtcp-fb-trr-int")
)
assert len(rtcp_fb_trr_int_elts) == 1
rtcp_fb_trr_int_elt = rtcp_fb_trr_int_elts[0]
assert int(rtcp_fb_trr_int_elt["value"]) == 100

9
tests/unit/test_plugin_xep_0320.py
View file

@ -40,8 +40,7 @@ class TestXEP0320:
"fingerprint": {
"hash": "sha-256",
"setup": "active",
"fingerprint":
"6D:8F:6A:53:A3:7E:10:B2:58:16:AB:A3:92:6F:8A:5B:2D:55:1C:"
"fingerprint": "6D:8F:6A:53:A3:7E:10:B2:58:16:AB:A3:92:6F:8A:5B:2D:55:1C:"
"FB:2F:E3:6E:94:FE:4F:4E:FE:D4:77:49:B6",
},
}
@ -59,8 +58,7 @@ class TestXEP0320:
"fingerprint": {
"hash": "sha-256",
"setup": "active",
"fingerprint":
"6D:8F:6A:53:A3:7E:10:B2:58:16:AB:A3:92:6F:8A:5B:2D:55:1C:"
"fingerprint": "6D:8F:6A:53:A3:7E:10:B2:58:16:AB:A3:92:6F:8A:5B:2D:55:1C:"
"FB:2F:E3:6E:94:FE:4F:4E:FE:D4:77:49:B6",
},
}
@ -70,7 +68,8 @@ class TestXEP0320:
expected_transport_elt = xml_tools.parse(
f"<transport><fingerprint xmlns='{NS_JINGLE_DTLS}' hash='sha-256' "
"setup='active'>6D:8F:6A:53:A3:7E:10:B2:58:16:AB:A3:92:6F:8A:5B:2D:55:1C:FB:"
"2F:E3:6E:94:FE:4F:4E:FE:D4:77:49:B6</fingerprint></transport>"""
"2F:E3:6E:94:FE:4F:4E:FE:D4:77:49:B6</fingerprint></transport>"
""
)
assert transport_elt.toXml() == expected_transport_elt.toXml()

55
tests/unit/test_plugin_xep_0373.py
View file

@ -2,6 +2,7 @@ from datetime import datetime, timedelta, timezone
import pytest
from libervia.backend.core import exceptions
try:
from libervia.backend.plugins.plugin_xep_0373 import XEP_0373, NS_OX
except exceptions.MissingModule as e:
@ -20,7 +21,7 @@ def test_signcrypt_element_args() -> None:
def test_signcrypt_element() -> None:
signcrypt_elt, payload_elt = XEP_0373.build_signcrypt_element([ a, b ])
signcrypt_elt, payload_elt = XEP_0373.build_signcrypt_element([a, b])
payload_elt.addElement("signcrypt-test-content", content="signcrypt test content")
rpad_elt = next(signcrypt_elt.elements(NS_OX, "rpad"))
@ -36,11 +37,11 @@ def test_signcrypt_element() -> None:
assert (datetime.now(timezone.utc) - timestamp) < timedelta(seconds=10)
assert signcrypt_elt.toXml() == (
"<signcrypt xmlns='urn:xmpp:openpgp:0'>"
"<to jid='foo@example.com'/>"
"<to jid='bar@example.com'/>"
"<payload>"
"<signcrypt-test-content>signcrypt test content</signcrypt-test-content>"
"</payload>"
"<to jid='foo@example.com'/>"
"<to jid='bar@example.com'/>"
"<payload>"
"<signcrypt-test-content>signcrypt test content</signcrypt-test-content>"
"</payload>"
"</signcrypt>"
)
@ -51,7 +52,7 @@ def test_sign_element_args() -> None:
def test_sign_element_with_rpad() -> None:
sign_elt, payload_elt = XEP_0373.build_sign_element([ a, b ], True)
sign_elt, payload_elt = XEP_0373.build_sign_element([a, b], True)
payload_elt.addElement("sign-test-content", content="sign test content")
rpad_elt = next(sign_elt.elements(NS_OX, "rpad"))
@ -67,17 +68,17 @@ def test_sign_element_with_rpad() -> None:
assert (datetime.now(timezone.utc) - timestamp) < timedelta(seconds=10)
assert sign_elt.toXml() == (
"<sign xmlns='urn:xmpp:openpgp:0'>"
"<to jid='foo@example.com'/>"
"<to jid='bar@example.com'/>"
"<payload>"
"<sign-test-content>sign test content</sign-test-content>"
"</payload>"
"<to jid='foo@example.com'/>"
"<to jid='bar@example.com'/>"
"<payload>"
"<sign-test-content>sign test content</sign-test-content>"
"</payload>"
"</sign>"
)
def test_sign_element_without_rpad() -> None:
sign_elt, payload_elt = XEP_0373.build_sign_element([ a, b ], False)
sign_elt, payload_elt = XEP_0373.build_sign_element([a, b], False)
payload_elt.addElement("sign-test-content", content="sign test content")
rpad_elt = next(sign_elt.elements(NS_OX, "rpad"), None)
@ -91,17 +92,17 @@ def test_sign_element_without_rpad() -> None:
assert (datetime.now(timezone.utc) - timestamp) < timedelta(seconds=10)
assert sign_elt.toXml() == (
"<sign xmlns='urn:xmpp:openpgp:0'>"
"<to jid='foo@example.com'/>"
"<to jid='bar@example.com'/>"
"<payload>"
"<sign-test-content>sign test content</sign-test-content>"
"</payload>"
"<to jid='foo@example.com'/>"
"<to jid='bar@example.com'/>"
"<payload>"
"<sign-test-content>sign test content</sign-test-content>"
"</payload>"
"</sign>"
)
def test_crypt_element_with_recipients() -> None:
crypt_elt, payload_elt = XEP_0373.build_crypt_element([ a, b ])
crypt_elt, payload_elt = XEP_0373.build_crypt_element([a, b])
payload_elt.addElement("crypt-test-content", content="crypt test content")
rpad_elt = next(crypt_elt.elements(NS_OX, "rpad"))
@ -117,11 +118,11 @@ def test_crypt_element_with_recipients() -> None:
assert (datetime.now(timezone.utc) - timestamp) < timedelta(seconds=10)
assert crypt_elt.toXml() == (
"<crypt xmlns='urn:xmpp:openpgp:0'>"
"<to jid='foo@example.com'/>"
"<to jid='bar@example.com'/>"
"<payload>"
"<crypt-test-content>crypt test content</crypt-test-content>"
"</payload>"
"<to jid='foo@example.com'/>"
"<to jid='bar@example.com'/>"
"<payload>"
"<crypt-test-content>crypt test content</crypt-test-content>"
"</payload>"
"</crypt>"
)
@ -143,8 +144,8 @@ def test_crypt_element_without_recipients() -> None:
assert (datetime.now(timezone.utc) - timestamp) < timedelta(seconds=10)
assert crypt_elt.toXml() == (
"<crypt xmlns='urn:xmpp:openpgp:0'>"
"<payload>"
"<crypt-test-content>crypt test content</crypt-test-content>"
"</payload>"
"<payload>"
"<crypt-test-content>crypt test content</crypt-test-content>"
"</payload>"
"</crypt>"
)

167
tests/unit/test_plugin_xep_0420.py
View file

@ -24,8 +24,13 @@ from libervia.backend.core import exceptions
from libervia.backend.plugins.plugin_xep_0334 import NS_HINTS
from libervia.backend.plugins.plugin_xep_0420 import (
NS_SCE, XEP_0420, AffixVerificationFailed, ProfileRequirementsNotMet, SCEAffixPolicy,
SCECustomAffix, SCEProfile
NS_SCE,
XEP_0420,
AffixVerificationFailed,
ProfileRequirementsNotMet,
SCEAffixPolicy,
SCECustomAffix,
SCEProfile,
)
from libervia.backend.tools.xml_tools import ElementParser
from twisted.words.xish import domish
@ -43,7 +48,7 @@ __all__ = [ # pylint: disable=unused-variable
"test_custom_affixes",
"test_namespace_conversion",
"test_non_encryptable_elements",
"test_schema_validation"
"test_schema_validation",
]
@ -110,7 +115,7 @@ def test_unpack_matches_original() -> None: # pylint: disable=missing-function-
SCEAffixPolicy.OPTIONAL,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.OPTIONAL,
custom_policies={ CustomAffixImpl(): SCEAffixPolicy.NOT_NEEDED }
custom_policies={CustomAffixImpl(): SCEAffixPolicy.NOT_NEEDED},
)
stanza_string = (
@ -142,19 +147,19 @@ def test_affixes_included() -> None: # pylint: disable=missing-function-docstri
SCEAffixPolicy.OPTIONAL,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.OPTIONAL,
custom_policies={ custom_affix: SCEAffixPolicy.OPTIONAL }
custom_policies={custom_affix: SCEAffixPolicy.OPTIONAL},
)
stanza = string_to_domish("""<message from="foo@example.com" to="bar@example.com">
stanza = string_to_domish(
"""<message from="foo@example.com" to="bar@example.com">
<body>
Make sure that both the REQUIRED and the OPTIONAL affixes are included.
</body>
</message>""")
</message>"""
)
affix_values = XEP_0420.unpack_stanza(
profile,
stanza,
XEP_0420.pack_stanza(profile, stanza)
profile, stanza, XEP_0420.pack_stanza(profile, stanza)
)
assert affix_values.rpad is not None
@ -170,7 +175,7 @@ def test_all_affixes_verified() -> None: # pylint: disable=missing-function-doc
SCEAffixPolicy.REQUIRED,
SCEAffixPolicy.REQUIRED,
SCEAffixPolicy.REQUIRED,
custom_policies={}
custom_policies={},
)
unpacking_profile = SCEProfile(
@ -178,14 +183,16 @@ def test_all_affixes_verified() -> None: # pylint: disable=missing-function-doc
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
custom_policies={}
custom_policies={},
)
stanza = string_to_domish("""<message from="foo@example.com" to="bar@example.com">
stanza = string_to_domish(
"""<message from="foo@example.com" to="bar@example.com">
<body>
When unpacking, all affixes are loaded, even those marked as NOT_NEEDED.
</body>
</message>""")
</message>"""
)
envelope_serialized = XEP_0420.pack_stanza(packing_profile, stanza)
@ -211,7 +218,7 @@ def test_incomplete_affixes() -> None: # pylint: disable=missing-function-docst
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
custom_policies={}
custom_policies={},
)
unpacking_profile = SCEProfile(
@ -219,18 +226,18 @@ def test_incomplete_affixes() -> None: # pylint: disable=missing-function-docst
SCEAffixPolicy.REQUIRED,
SCEAffixPolicy.REQUIRED,
SCEAffixPolicy.REQUIRED,
custom_policies={}
custom_policies={},
)
stanza = string_to_domish("""<message from="foo@example.com" to="bar@example.com">
stanza = string_to_domish(
"""<message from="foo@example.com" to="bar@example.com">
<body>Check that all affixes REQUIRED by the profile are present.</body>
</message>""")
</message>"""
)
with pytest.raises(ProfileRequirementsNotMet):
XEP_0420.unpack_stanza(
unpacking_profile,
stanza,
XEP_0420.pack_stanza(packing_profile, stanza)
unpacking_profile, stanza, XEP_0420.pack_stanza(packing_profile, stanza)
)
# Do the same but with a custom affix missing
@ -241,7 +248,7 @@ def test_incomplete_affixes() -> None: # pylint: disable=missing-function-docst
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
custom_policies={ custom_affix: SCEAffixPolicy.NOT_NEEDED }
custom_policies={custom_affix: SCEAffixPolicy.NOT_NEEDED},
)
unpacking_profile = SCEProfile(
@ -249,21 +256,21 @@ def test_incomplete_affixes() -> None: # pylint: disable=missing-function-docst
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
custom_policies={ custom_affix: SCEAffixPolicy.REQUIRED }
custom_policies={custom_affix: SCEAffixPolicy.REQUIRED},
)
stanza = string_to_domish("""<message from="foo@example.com" to="bar@example.com">
stanza = string_to_domish(
"""<message from="foo@example.com" to="bar@example.com">
<body>
Check that all affixes REQUIRED by the profile are present, including custom
affixes.
</body>
</message>""")
</message>"""
)
with pytest.raises(ProfileRequirementsNotMet):
XEP_0420.unpack_stanza(
unpacking_profile,
stanza,
XEP_0420.pack_stanza(packing_profile, stanza)
unpacking_profile, stanza, XEP_0420.pack_stanza(packing_profile, stanza)
)
@ -273,18 +280,18 @@ def test_rpad_affix() -> None: # pylint: disable=missing-function-docstring
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
custom_policies={}
custom_policies={},
)
for _ in range(100):
stanza = string_to_domish("""<message from="foo@example.com" to="bar@example.com">
stanza = string_to_domish(
"""<message from="foo@example.com" to="bar@example.com">
<body>OK</body>
</message>""")
</message>"""
)
affix_values = XEP_0420.unpack_stanza(
profile,
stanza,
XEP_0420.pack_stanza(profile, stanza)
profile, stanza, XEP_0420.pack_stanza(profile, stanza)
)
# Test that the rpad exists and that the content elements are always padded to at
@ -299,7 +306,7 @@ def test_time_affix() -> None: # pylint: disable=missing-function-docstring
SCEAffixPolicy.REQUIRED,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
custom_policies={}
custom_policies={},
)
stanza = string_to_domish(
@ -314,7 +321,9 @@ def test_time_affix() -> None: # pylint: disable=missing-function-docstring
</body>
</content>
<time stamp="1969-07-21T02:56:15Z"/>
</envelope>""".encode("utf-8")
</envelope>""".encode(
"utf-8"
)
affix_values = XEP_0420.unpack_stanza(profile, stanza, envelope_serialized)
assert affix_values.timestamp == datetime(1969, 7, 21, 2, 56, 15, tzinfo=timezone.utc)
@ -326,12 +335,14 @@ def test_to_affix() -> None: # pylint: disable=missing-function-docstring
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.REQUIRED,
SCEAffixPolicy.NOT_NEEDED,
custom_policies={}
custom_policies={},
)
stanza = string_to_domish("""<message from="foo@example.com" to="bar@example.com">
stanza = string_to_domish(
"""<message from="foo@example.com" to="bar@example.com">
<body>Check that the ``to`` affix is correctly added.</body>
</message>""")
</message>"""
)
envelope_serialized = XEP_0420.pack_stanza(profile, stanza)
affix_values = XEP_0420.unpack_stanza(profile, stanza, envelope_serialized)
@ -355,11 +366,13 @@ def test_to_affix() -> None: # pylint: disable=missing-function-docstring
assert affix_values.recipient is not None
assert affix_values.recipient.userhost() == "bar@example.com"
stanza = string_to_domish("""<message from="foo@example.com">
stanza = string_to_domish(
"""<message from="foo@example.com">
<body>
Check that a missing "to" attribute on the stanza fails stanza packing.
</body>
</message>""")
</message>"""
)
with pytest.raises(ValueError):
XEP_0420.pack_stanza(profile, stanza)
@ -371,12 +384,14 @@ def test_from_affix() -> None: # pylint: disable=missing-function-docstring
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.REQUIRED,
custom_policies={}
custom_policies={},
)
stanza = string_to_domish("""<message from="foo@example.com" to="bar@example.com">
stanza = string_to_domish(
"""<message from="foo@example.com" to="bar@example.com">
<body>Check that the ``from`` affix is correctly added.</body>
</message>""")
</message>"""
)
envelope_serialized = XEP_0420.pack_stanza(profile, stanza)
affix_values = XEP_0420.unpack_stanza(profile, stanza, envelope_serialized)
@ -400,11 +415,13 @@ def test_from_affix() -> None: # pylint: disable=missing-function-docstring
assert affix_values.sender is not None
assert affix_values.sender.userhost() == "foo@example.com"
stanza = string_to_domish("""<message to="bar@example.com">
stanza = string_to_domish(
"""<message to="bar@example.com">
<body>
Check that a missing "from" attribute on the stanza fails stanza packing.
</body>
</message>""")
</message>"""
)
with pytest.raises(ValueError):
XEP_0420.pack_stanza(profile, stanza)
@ -418,7 +435,7 @@ def test_custom_affixes() -> None: # pylint: disable=missing-function-docstring
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
custom_policies={ custom_affix: SCEAffixPolicy.REQUIRED }
custom_policies={custom_affix: SCEAffixPolicy.REQUIRED},
)
unpacking_profile = SCEProfile(
@ -426,41 +443,45 @@ def test_custom_affixes() -> None: # pylint: disable=missing-function-docstring
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
custom_policies={}
custom_policies={},
)
stanza = string_to_domish("""<message from="foo@example.com" to="bar@example.com">
stanza = string_to_domish(
"""<message from="foo@example.com" to="bar@example.com">
<body>
If a custom affix is included in the envelope, but not excpected by the
recipient, the schema validation should fail.
</body>
</message>""")
</message>"""
)
with pytest.raises(exceptions.ParsingError):
XEP_0420.unpack_stanza(
unpacking_profile,
stanza,
XEP_0420.pack_stanza(packing_profile, stanza)
unpacking_profile, stanza, XEP_0420.pack_stanza(packing_profile, stanza)
)
profile = packing_profile
stanza = string_to_domish("""<message from="foo@example.com/device0"
stanza = string_to_domish(
"""<message from="foo@example.com/device0"
to="bar@example.com/Libervia.123">
<body>The affix element should be returned as part of the affix values.</body>
</message>""")
</message>"""
)
affix_values = XEP_0420.unpack_stanza(
profile,
stanza,
XEP_0420.pack_stanza(profile, stanza)
profile, stanza, XEP_0420.pack_stanza(profile, stanza)
)
assert custom_affix in affix_values.custom
assert affix_values.custom[custom_affix].getAttribute("recipient") == \
"bar@example.com/Libervia.123"
assert affix_values.custom[custom_affix].getAttribute("sender") == \
"foo@example.com/device0"
assert (
affix_values.custom[custom_affix].getAttribute("recipient")
== "bar@example.com/Libervia.123"
)
assert (
affix_values.custom[custom_affix].getAttribute("sender")
== "foo@example.com/device0"
)
def test_namespace_conversion() -> None: # pylint: disable=missing-function-docstring
@ -469,7 +490,7 @@ def test_namespace_conversion() -> None: # pylint: disable=missing-function-doc
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
custom_policies={}
custom_policies={},
)
stanza = domish.Element((None, "message"))
@ -480,7 +501,7 @@ def test_namespace_conversion() -> None: # pylint: disable=missing-function-doc
content=(
"This body element has namespace ``None``, which has to be replaced with"
" jabber:client."
)
),
)
envelope_serialized = XEP_0420.pack_stanza(profile, stanza)
@ -502,13 +523,15 @@ def test_non_encryptable_elements() -> None: # pylint: disable=missing-function
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
custom_policies={}
custom_policies={},
)
stanza = string_to_domish("""<message from="foo@example.com" to="bar@example.com">
stanza = string_to_domish(
"""<message from="foo@example.com" to="bar@example.com">
<body>This stanza includes a store hint which must not be encrypted.</body>
<store xmlns="urn:xmpp:hints"/>
</message>""")
</message>"""
)
envelope_serialized = XEP_0420.pack_stanza(profile, stanza)
envelope = string_to_domish(envelope_serialized.decode("utf-8"))
@ -529,7 +552,9 @@ def test_non_encryptable_elements() -> None: # pylint: disable=missing-function
</body>
<store xmlns="urn:xmpp:hints"/>
</content>
</envelope>""".encode("utf-8")
</envelope>""".encode(
"utf-8"
)
XEP_0420.unpack_stanza(profile, stanza, envelope_serialized)
@ -542,7 +567,7 @@ def test_schema_validation() -> None: # pylint: disable=missing-function-docstr
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
SCEAffixPolicy.NOT_NEEDED,
custom_policies={}
custom_policies={},
)
stanza = string_to_domish(
@ -557,7 +582,9 @@ def test_schema_validation() -> None: # pylint: disable=missing-function-docstr
<store xmlns="urn:xmpp:hints"/>
</content>
<unknown-affix unknown-attr="unknown"/>
</envelope>""".encode("utf-8")
</envelope>""".encode(
"utf-8"
)
with pytest.raises(exceptions.ParsingError):
XEP_0420.unpack_stanza(profile, stanza, envelope_serialized)

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

@ -33,7 +33,7 @@ class TestPubsubCache:
client.pubsub_client.items = MagicMock(return_value=items_ret)
host.memory.storage.get_pubsub_node.return_value = None
pubsub_node = host.memory.storage.set_pubsub_node.return_value = PubsubNode(
sync_state = None
sync_state=None
)
with patch.object(host.plugins["PUBSUB_CACHE"], "cache_node") as cache_node:
await host.plugins["XEP-0060"].get_items(
@ -51,15 +51,10 @@ class TestPubsubCache:
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,
None,
"urn:xmpp:microblog:0",
extra = {C.KEY_USE_CACHE: False}
client, None, "urn:xmpp:microblog:0", extra={C.KEY_USE_CACHE: False}
)
assert not cache_node.called
@ -71,9 +66,7 @@ class TestPubsubCache:
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,
@ -82,7 +75,6 @@ class TestPubsubCache:
)
assert not cache_node.called
@ed
async def test_no_pubsub_get_when_cache_completed(self, host, client):
"""No pubsub get is emitted when items are fully cached"""
@ -90,11 +82,10 @@ class TestPubsubCache:
items_ret.callback(([], {}))
client.pubsub_client.items = MagicMock(return_value=items_ret)
host.memory.storage.get_pubsub_node.return_value = PubsubNode(
sync_state = SyncState.COMPLETED
sync_state=SyncState.COMPLETED
)
with patch.object(
host.plugins["PUBSUB_CACHE"],
"get_items_from_cache"
host.plugins["PUBSUB_CACHE"], "get_items_from_cache"
) as get_items_from_cache:
get_items_from_cache.return_value = ([], {})
await host.plugins["XEP-0060"].get_items(
@ -112,13 +103,12 @@ class TestPubsubCache:
items_ret.callback(([], {}))
client.pubsub_client.items = MagicMock(return_value=items_ret)
host.memory.storage.get_pubsub_node.return_value = PubsubNode(
sync_state = SyncState.IN_PROGRESS
sync_state=SyncState.IN_PROGRESS
)
with patch.object(host.plugins["PUBSUB_CACHE"], "analyse_node") as analyse_node:
analyse_node.return_value = {"to_sync": True}
with patch.object(
host.plugins["PUBSUB_CACHE"],
"get_items_from_cache"
host.plugins["PUBSUB_CACHE"], "get_items_from_cache"
) as get_items_from_cache:
get_items_from_cache.return_value = ([], {})
assert client.pubsub_client.items.call_count == 0