Mercurial > libervia-backend
changeset 4285:f1d0cde61af7 default tip @
tests (unit): fix tests + black reformatting.
| author | Goffi <goffi@goffi.org> |
|---|---|
| date | Sun, 14 Jul 2024 17:42:53 +0200 |
| parents | 3a550e9a2b55 |
| children | |
| files | tests/unit/conftest.py tests/unit/frontends/test_webrtc.py tests/unit/test_ap-gateway.py tests/unit/test_plugin_xep_0082.py tests/unit/test_plugin_xep_0167.py tests/unit/test_plugin_xep_0272.py tests/unit/test_plugin_xep_0293.py tests/unit/test_plugin_xep_0320.py tests/unit/test_plugin_xep_0373.py tests/unit/test_plugin_xep_0420.py tests/unit/test_pubsub-cache.py |
| diffstat | 11 files changed, 233 insertions(+), 198 deletions(-) [+] |
line wrap: on
line diff
--- a/tests/unit/conftest.py Sun Jul 14 16:47:45 2024 +0200 +++ b/tests/unit/conftest.py Sun Jul 14 17:42:53 2024 +0200 @@ -48,7 +48,7 @@ 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,
--- a/tests/unit/frontends/test_webrtc.py Sun Jul 14 16:47:45 2024 +0200 +++ b/tests/unit/frontends/test_webrtc.py Sun Jul 14 17:42:53 2024 +0200 @@ -29,7 +29,6 @@ from libervia.frontends.tools import webrtc as webrtc_mod - @pytest.fixture def host(monkeypatch): host = MagicMock() @@ -37,6 +36,7 @@ host.app.expand = lambda s: s return host + @pytest.fixture(scope="function") def webrtc(host): """Fixture for WebRTC instantiation.""" @@ -185,7 +185,7 @@ 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 @@ 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
--- a/tests/unit/test_ap-gateway.py Sun Jul 14 16:47:45 2024 +0200 +++ b/tests/unit/test_ap-gateway.py Sun Jul 14 17:42:53 2024 +0200 @@ -718,8 +718,7 @@ 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 @@ ) 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 @@ ) 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 @@ 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 @@ 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 @@ 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 @@ 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 @@ 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"] == [
--- a/tests/unit/test_plugin_xep_0082.py Sun Jul 14 16:47:45 2024 +0200 +++ b/tests/unit/test_plugin_xep_0082.py Sun Jul 14 17:42:53 2024 +0200 @@ -30,7 +30,7 @@ "test_datetime_formatting", "test_datetime_parsing", "test_time_formatting", - "test_time_parsing" + "test_time_parsing", ] @@ -85,9 +85,10 @@ # 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 @@ 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 @@ # 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 @@ 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 + )
--- a/tests/unit/test_plugin_xep_0167.py Sun Jul 14 16:47:45 2024 +0200 +++ b/tests/unit/test_plugin_xep_0167.py Sun Jul 14 17:42:53 2024 +0200 @@ -25,7 +25,10 @@ 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 @@ """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 @@ "media": "audio", "payload_types": { 96: { + "clockrate": 48000, "id": 96, "name": "opus", - "clockrate": 48000, "parameters": {"sprop-stereo": "1"}, } }, }, + "id": "audio", + "senders": "both", "transport_data": { + "candidates": [ + { + "address": "10.0.0.1", + "component_id": 1, + "foundation": "1", + "port": 12345, + "priority": 1, + "transport": "UDP", + "type": "host", + } + ], "port": 9999, "pwd": "pwd", "ufrag": "ufrag", - "candidates": [ - { - "foundation": "1", - "component_id": 1, - "transport": "UDP", - "priority": 1, - "address": "10.0.0.1", - "port": 12345, - "type": "host", - } - ], }, - "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"""
--- a/tests/unit/test_plugin_xep_0272.py Sun Jul 14 16:47:45 2024 +0200 +++ b/tests/unit/test_plugin_xep_0272.py Sun Jul 14 17:42:53 2024 +0200 @@ -35,7 +35,6 @@ 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 @@ 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 @@ 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"
--- a/tests/unit/test_plugin_xep_0293.py Sun Jul 14 16:47:45 2024 +0200 +++ b/tests/unit/test_plugin_xep_0293.py Sun Jul 14 17:42:53 2024 +0200 @@ -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 @@ """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 @@ """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 @@ 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 @@ 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 @@ 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 @@ # 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 @@ 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
--- a/tests/unit/test_plugin_xep_0320.py Sun Jul 14 16:47:45 2024 +0200 +++ b/tests/unit/test_plugin_xep_0320.py Sun Jul 14 17:42:53 2024 +0200 @@ -40,8 +40,7 @@ "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 @@ "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 @@ 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()
--- a/tests/unit/test_plugin_xep_0373.py Sun Jul 14 16:47:45 2024 +0200 +++ b/tests/unit/test_plugin_xep_0373.py Sun Jul 14 17:42:53 2024 +0200 @@ -2,6 +2,7 @@ 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() -> 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 @@ 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_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 @@ 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 @@ 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 @@ 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 @@ 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>" )
--- a/tests/unit/test_plugin_xep_0420.py Sun Jul 14 16:47:45 2024 +0200 +++ b/tests/unit/test_plugin_xep_0420.py Sun Jul 14 17:42:53 2024 +0200 @@ -24,8 +24,13 @@ 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 @@ "test_custom_affixes", "test_namespace_conversion", "test_non_encryptable_elements", - "test_schema_validation" + "test_schema_validation", ] @@ -110,7 +115,7 @@ 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 @@ 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 @@ SCEAffixPolicy.REQUIRED, SCEAffixPolicy.REQUIRED, SCEAffixPolicy.REQUIRED, - custom_policies={} + custom_policies={}, ) unpacking_profile = SCEProfile( @@ -178,14 +183,16 @@ 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 @@ SCEAffixPolicy.NOT_NEEDED, SCEAffixPolicy.NOT_NEEDED, SCEAffixPolicy.NOT_NEEDED, - custom_policies={} + custom_policies={}, ) unpacking_profile = SCEProfile( @@ -219,18 +226,18 @@ 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 @@ 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 @@ 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 @@ 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 @@ SCEAffixPolicy.REQUIRED, SCEAffixPolicy.NOT_NEEDED, SCEAffixPolicy.NOT_NEEDED, - custom_policies={} + custom_policies={}, ) stanza = string_to_domish( @@ -314,7 +321,9 @@ </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 @@ 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 @@ 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 @@ 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 @@ 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 @@ 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 @@ 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 @@ SCEAffixPolicy.NOT_NEEDED, SCEAffixPolicy.NOT_NEEDED, SCEAffixPolicy.NOT_NEEDED, - custom_policies={} + custom_policies={}, ) stanza = domish.Element((None, "message")) @@ -480,7 +501,7 @@ 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 @@ 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 @@ </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 @@ SCEAffixPolicy.NOT_NEEDED, SCEAffixPolicy.NOT_NEEDED, SCEAffixPolicy.NOT_NEEDED, - custom_policies={} + custom_policies={}, ) stanza = string_to_domish( @@ -557,7 +582,9 @@ <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)
--- a/tests/unit/test_pubsub-cache.py Sun Jul 14 16:47:45 2024 +0200 +++ b/tests/unit/test_pubsub-cache.py Sun Jul 14 17:42:53 2024 +0200 @@ -33,7 +33,7 @@ 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 @@ 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 @@ 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 @@ ) 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 @@ 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 @@ 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