Mercurial > libervia-backend
view tests/unit/test_plugin_xep_0373.py @ 4306:94e0968987cd
plugin XEP-0033: code modernisation, improve delivery, data validation:
- Code has been rewritten using Pydantic models and `async` coroutines for data validation
and cleaner element parsing/generation.
- Delivery has been completely rewritten. It now works even if server doesn't support
multicast, and send to local multicast service first. Delivering to local multicast
service first is due to bad support of XEP-0033 in server (notably Prosody which has an
incomplete implementation), and the current impossibility to detect if a sub-domain
service handles fully multicast or only for local domains. This is a workaround to have
a good balance between backward compatilibity and use of bandwith, and to make it work
with the incoming email gateway implementation (the gateway will only deliver to
entities of its own domain).
- disco feature checking now uses `async` corountines. `host` implementation still use
Deferred return values for compatibility with legacy code.
rel 450
author | Goffi <goffi@goffi.org> |
---|---|
date | Thu, 26 Sep 2024 16:12:01 +0200 |
parents | f1d0cde61af7 |
children |
line wrap: on
line source
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: pytest.skip(f"Can't test XEP-0373: {e}", allow_module_level=True) from libervia.backend.tools.xmpp_datetime import parse_datetime from twisted.words.protocols.jabber import jid a = jid.JID("foo@example.com") b = jid.JID("bar@example.com") def test_signcrypt_element_args() -> None: with pytest.raises(ValueError): XEP_0373.build_signcrypt_element([]) def test_signcrypt_element() -> None: 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")) time_elt = next(signcrypt_elt.elements(NS_OX, "time")) rpad = str(rpad_elt) timestamp = parse_datetime(time_elt["stamp"]) signcrypt_elt.children.remove(rpad_elt) signcrypt_elt.children.remove(time_elt) assert rpad 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>" "</signcrypt>" ) def test_sign_element_args() -> None: with pytest.raises(ValueError): XEP_0373.build_sign_element([], True) def test_sign_element_with_rpad() -> None: 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")) time_elt = next(sign_elt.elements(NS_OX, "time")) rpad = str(rpad_elt) timestamp = parse_datetime(time_elt["stamp"]) sign_elt.children.remove(rpad_elt) sign_elt.children.remove(time_elt) assert rpad 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>" "</sign>" ) def test_sign_element_without_rpad() -> None: 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) time_elt = next(sign_elt.elements(NS_OX, "time")) timestamp = parse_datetime(time_elt["stamp"]) sign_elt.children.remove(time_elt) assert rpad_elt is 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>" "</sign>" ) def test_crypt_element_with_recipients() -> None: 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")) time_elt = next(crypt_elt.elements(NS_OX, "time")) rpad = str(rpad_elt) timestamp = parse_datetime(time_elt["stamp"]) crypt_elt.children.remove(rpad_elt) crypt_elt.children.remove(time_elt) assert rpad 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>" "</crypt>" ) def test_crypt_element_without_recipients() -> None: crypt_elt, payload_elt = XEP_0373.build_crypt_element([]) payload_elt.addElement("crypt-test-content", content="crypt test content") rpad_elt = next(crypt_elt.elements(NS_OX, "rpad")) time_elt = next(crypt_elt.elements(NS_OX, "time")) rpad = str(rpad_elt) timestamp = parse_datetime(time_elt["stamp"]) crypt_elt.children.remove(rpad_elt) crypt_elt.children.remove(time_elt) assert rpad 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>" "</crypt>" )