apply black to the whole codebase.

This commit is contained in:
Goffi 2026-03-09 09:22:04 +01:00
commit 82e2c6efac
163 changed files with 175 additions and 448 deletions

View file

@ -80,8 +80,7 @@ class Editor:
self.editor_path = self.tmp_dir_path / "editor.py"
self.ori_content_path = self.tmp_dir_path / "original_content"
self.new_content_path = self.tmp_dir_path / "new_content"
self.base_script = dedent(
f"""\
self.base_script = dedent(f"""\
#!{sys.executable}
import sys
@ -100,8 +99,7 @@ class Editor:
with open("{self.new_content_path}", "w") as f:
f.write(new_content)
"""
)
""")
self._env = os.environ.copy()
self._env["EDITOR"] = str(self.editor_path)

View file

@ -30,7 +30,6 @@ from libervia.backend.plugins.plugin_sec_pte import NS_PTE
from libervia.backend.plugins.plugin_xep_0277 import NS_ATOM
from libervia.backend.tools.common import uri
if os.getenv("LIBERVIA_TEST_ENV_E2E") is None:
pytest.skip(
"skipping end-to-end tests, we are not in a test environment",

View file

@ -25,7 +25,6 @@ from datetime import datetime
from pathlib import Path
import helium
WEB_HOST = "web"
WEB_PORT_HTTPS = 8443
BASE_URL = f"https://{WEB_HOST}:{WEB_PORT_HTTPS}"

View file

@ -40,7 +40,6 @@ from helium import (
)
from selenium.webdriver.common.by import By
if os.getenv("LIBERVIA_TEST_ENV_E2E_WEB") is None:
pytest.skip(
"skipping end-to-end tests, we are not in a test environment for Libervia",

View file

@ -44,8 +44,7 @@ OPT_KEEP_BROWSER = "--keep-browser"
OPT_VISUAL = "--visual"
OPT_DEV_MODE = "--dev-mode"
dev_mode_inst = dedent(
"""\
dev_mode_inst = dedent("""\
Here is a short script to start working with a logged account:
from helium import *
@ -54,8 +53,7 @@ dev_mode_inst = dedent(
write("account1", "login")
write("test", "password")
click("log in")
"""
)
""")
report_buffer = io.StringIO()
live_out_buf = []
live_err_buf = []
@ -186,8 +184,7 @@ def use_e2e_env():
with tempfile.TemporaryDirectory(prefix="libervia_test_e2e_") as temp_dir:
override_path = Path(temp_dir) / "test_override.yml"
override = yaml.load(
dedent(
f"""\
dedent(f"""\
version: "3.6"
services:
backend:
@ -210,8 +207,7 @@ def use_e2e_env():
source: {libervia_templates_source}
target: {libervia_templates_target}
read_only: true
"""
),
"""),
Loader=Loader,
)

View file

@ -1,4 +1,3 @@
from pytest_twisted import init_asyncio_reactor
init_asyncio_reactor()

View file

@ -28,12 +28,12 @@ from pytest_twisted import ensureDeferred as ed
from libervia.backend.core import exceptions
from libervia.backend.tools.common import data_format
try:
from libervia.frontends.tools import webrtc as webrtc_mod
except exceptions.MissingModule as e:
pytest.skip(
"Can't import webrtc due to missing modules: {e}",
allow_module_level=True
"Can't import webrtc due to missing modules: {e}", allow_module_level=True
)

View file

@ -49,7 +49,6 @@ from libervia.backend.tools import xml_tools
from libervia.backend.tools.common import uri as xmpp_uri
from libervia.backend.tools.utils import xmpp_date
TEST_BASE_URL = "https://example.org"
TEST_USER = "test_user"
TEST_AP_ACCOUNT = f"{TEST_USER}@example.org"
@ -1360,8 +1359,7 @@ class TestActivityPubGateway:
node=ap_gateway._m.namespace,
item=ap_item["id"],
)
item_elt = xml_tools.parse(
f"""
item_elt = xml_tools.parse(f"""
<item id="123" publisher="{TEST_JID}/res.123">
<entry xmlns="http://www.w3.org/2005/Atom">
<title type="text">test message 1</title>
@ -1380,8 +1378,7 @@ class TestActivityPubGateway:
<link href="{ap_item_url}" rel="via"/>
</entry>
</item>
"""
)
""")
item_elt.uri = pubsub.NS_PUBSUB_EVENT
with patch.object(ap_gateway, "ap_post") as ap_post:
@ -1473,15 +1470,13 @@ class TestActivityPubGateway:
attachment_node = ap_gateway._pa.get_attachment_node_name(
recipient_jid, ap_gateway._m.namespace, ap_item["id"]
)
item_elt = xml_tools.parse(
f"""
item_elt = xml_tools.parse(f"""
<item id="{TEST_JID.userhost()}" published="{TEST_JID.userhostJID()}">
<attachments xmlns="urn:xmpp:pubsub-attachments:1">
<noticed timestamp="2022-07-22T12:29:45Z"/>
</attachments>
</item>
"""
)
""")
item_elt.uri = pubsub.NS_PUBSUB_EVENT
items_event = pubsub.ItemsEvent(
TEST_JID, recipient_jid, attachment_node, [item_elt], {}
@ -1565,8 +1560,7 @@ class TestActivityPubGateway:
recipient_jid, ap_gateway._m.namespace, ap_item["id"]
)
reactions = ["🦁", "🥜", "🎻"]
item_elt = xml_tools.parse(
f"""
item_elt = xml_tools.parse(f"""
<item id="{TEST_JID.userhost()}" published="{TEST_JID.userhostJID()}">
<attachments xmlns="urn:xmpp:pubsub-attachments:1">
<reactions timestamp="2022-08-31T12:17:23Z">
@ -1576,8 +1570,7 @@ class TestActivityPubGateway:
</reactions>
</attachments>
</item>
"""
)
""")
item_elt.uri = pubsub.NS_PUBSUB_EVENT
items_event = pubsub.ItemsEvent(
TEST_JID, recipient_jid, attachment_node, [item_elt], {}

View file

@ -255,6 +255,7 @@ class TestContact:
"name"
), "name attribute should not be present in group with id_only"
class TestContacts:
"""Tests for Contacts root model."""

View file

@ -23,7 +23,6 @@ from libervia.backend.core import exceptions
from libervia.backend.plugins.plugin_xep_0082 import XEP_0082
__all__ = [ # pylint: disable=unused-variable
"test_date_formatting",
"test_date_parsing",

View file

@ -247,8 +247,7 @@ class TestXEP0167Mapping:
def test_parse_description(self):
"""Parsing <description> to a dict is successful"""
description_element = xml_tools.parse(
"""
description_element = xml_tools.parse("""
<description xmlns="urn:xmpp:jingle:apps:rtp:1" media="audio">
<payload-type id="96" channels="2" clockrate="48000" maxptime="60" name="opus" ptime="20">
<parameter name="sprop-stereo" value="1" />
@ -259,8 +258,7 @@ class TestXEP0167Mapping:
<crypto tag="1" crypto-suite="AES_CM_128_HMAC_SHA1_80" key-params="inline:DPSKYRle84Ua2MjbScjadpCvFQH5Tutuls2N4/xx" session-params="" />
</encryption>
</description>
"""
)
""")
parsed_data = mapping.parse_description(description_element)
@ -296,16 +294,14 @@ class TestXEP0167:
mock_call_info = MagicMock()
host.bridge.call_info = mock_call_info
jingle_elt = xml_tools.parse(
"""
jingle_elt = xml_tools.parse("""
<jingle xmlns='urn:xmpp:jingle:1'
action='session-info'
initiator='client1@example.org'
sid='a73sjjvkla37jfea'>
<mute xmlns="urn:xmpp:jingle:apps:rtp:info:1" name="mute_name"/>
</jingle>
"""
)
""")
xep_0167.jingle_session_info(client, "mute", session, "content_name", jingle_elt)
@ -323,16 +319,14 @@ class TestXEP0167:
mock_call_info = MagicMock()
host.bridge.call_info = mock_call_info
jingle_elt = xml_tools.parse(
"""
jingle_elt = xml_tools.parse("""
<jingle xmlns='urn:xmpp:jingle:1'
action='session-info'
initiator='client1@example.org'
sid='a73sjjvkla37jfea'>
<invalid xmlns="urn:xmpp:jingle:apps:rtp:info:1" name="invalid_name"/>
</jingle>
"""
)
""")
xep_0167.jingle_session_info(
client, "invalid", session, "content_name", jingle_elt
@ -345,8 +339,7 @@ class TestXEP0167:
session_id = "123"
extra = {"name": "test"}
iq_elt = xml_tools.parse(
"""
iq_elt = xml_tools.parse("""
<iq from='client1@example.org'
id='yh3gr714'
to='client2@example.net'
@ -358,8 +351,7 @@ class TestXEP0167:
<active xmlns='urn:xmpp:jingle:apps:rtp:info:1'/>
</jingle>
</iq>
"""
)
""")
jingle_elt = iq_elt.firstChildElement()
mock_send = MagicMock()
iq_elt.send = mock_send

View file

@ -144,8 +144,7 @@ class TestXEP0176:
"""Transport element is correctly parsed into ICE data"""
xep_0176 = XEP_0176(host)
transport_elt = xml_tools.parse(
"""
transport_elt = xml_tools.parse("""
<transport xmlns="urn:xmpp:jingle:transports:ice-udp:1"
pwd="password1"
ufrag="user1">
@ -172,8 +171,7 @@ class TestXEP0176:
rel-addr="10.0.0.1"
rel-port="9012" />
</transport>
"""
)
""")
ice_data = xep_0176.parse_transport(transport_elt)

View file

@ -31,8 +31,7 @@ class TestXEP0215:
host.memory.disco.has_feature = AsyncMock()
host.memory.disco.has_feature.return_value = True
services_elt = xml_tools.parse(
"""
services_elt = xml_tools.parse("""
<services xmlns="urn:xmpp:extdisco:2">
<service host="test.example" type="stun" port="1234" />
<service host="example.org" type="turn" port="5678" restricted="true">
@ -53,8 +52,7 @@ class TestXEP0215:
</x>
</service>
</services>"
"""
)
""")
services = xep_0215.parse_services(services_elt)
@ -109,8 +107,7 @@ class TestXEP0215:
client.IQ.return_value = iq_result
iq_result_elt = xml_tools.parse(
"""
iq_result_elt = xml_tools.parse("""
<iq type="result">
<services xmlns="urn:xmpp:extdisco:2">
<service host="test.example" type="stun" port="1234" />
@ -118,8 +115,7 @@ class TestXEP0215:
restricted="true" />
</services>
</iq>
"""
)
""")
iq_result.send.return_value = defer.succeed(iq_result_elt)
services = await xep_0215.get_external_services(client)
@ -149,16 +145,14 @@ class TestXEP0215:
client.IQ.return_value = iq_result
iq_result_elt = xml_tools.parse(
"""
iq_result_elt = xml_tools.parse("""
<iq type="result">
<credentials xmlns="urn:xmpp:extdisco:2">
<service host="test.example" type="stun" port="1234" username="user1"
password="pass1" restricted="true"/>
</credentials>
</iq>
"""
)
""")
iq_result.send.return_value = defer.succeed(iq_result_elt)
credentials = await xep_0215.request_credentials(client, "test.example", "stun")
@ -194,8 +188,7 @@ class TestXEP0215:
],
}
iq_elt = xml_tools.parse(
"""
iq_elt = xml_tools.parse("""
<iq type="set" from="test.example">
<services xmlns="urn:xmpp:extdisco:2">
<service action="add" host="example.net" type="stun" port="2345" />
@ -204,8 +197,7 @@ class TestXEP0215:
<service action="delete" host="example.org" type="turn" port="5678" />
</services>
</iq>
"""
)
""")
xep_0215.on_services_push(iq_elt, client)

View file

@ -41,16 +41,14 @@ class TestXEP0234:
async def test_parse_file_element(self, host, client):
"""``parse_file_element`` updates file_data dictionary correctly."""
xep_0234 = XEP_0234(host)
file_elt = xml_tools.parse(
"""
file_elt = xml_tools.parse("""
<file xmlns='urn:xmpp:jingle:apps:file-transfer:5'>
<media-type>text/plain</media-type>
<name>test.txt</name>
<date>2015-07-26T21:46:00+01:00</date>
<size>6144</size>
</file>
"""
)
""")
file_data = {}
result = await xep_0234.parse_file_element(client, file_elt, file_data)
@ -75,13 +73,11 @@ class TestXEP0234:
async def test_parse_file_element_invalid_file_element(self, host, client):
"""Raise DataError exception if <file> element is invalid."""
xep_0234 = XEP_0234(host)
file_elt = xml_tools.parse(
"""
file_elt = xml_tools.parse("""
<file xmlns='invalid_namespace'>
<name>example.txt</name>
</file>
"""
)
""")
with pytest.raises(exceptions.DataError):
await xep_0234.parse_file_element(client, file_elt)
@ -111,16 +107,14 @@ class TestXEP0234:
content_data = {"application_data": {}, "transport_data": {"webrtc": False}}
content_name = "dummy_content"
file_data = {"progress_id": "123"}
file_elt = xml_tools.parse(
"""
file_elt = xml_tools.parse("""
<file xmlns='urn:xmpp:jingle:apps:file-transfer:5'>
<media-type>text/plain</media-type>
<name>test.txt</name>
<date>2015-07-26T21:46:00+01:00</date>
<size>6144</size>
</file>
"""
)
""")
confirmed = await xep_0234._file_receiving_request_conf(
client, session, content_data, content_name, file_data, file_elt
@ -157,16 +151,14 @@ class TestXEP0234:
content_data = {"application_data": {}, "transport_data": {"webrtc": True}}
content_name = "dummy_content"
file_data = {"progress_id": "123"}
file_elt = xml_tools.parse(
"""
file_elt = xml_tools.parse("""
<file xmlns='urn:xmpp:jingle:apps:file-transfer:5'>
<media-type>text/plain</media-type>
<name>test.txt</name>
<date>2015-07-26T21:46:00+01:00</date>
<size>6144</size>
</file>
"""
)
""")
confirmed = await xep_0234._file_receiving_request_conf(
client, session, content_data, content_name, file_data, file_elt

View file

@ -164,14 +164,12 @@ class TestXEP0293:
"""Parsing of <rtcp-fb> and <rtcp-fb-trr-int> elements from a description."""
xep_0293 = XEP_0293(host)
desc_element = xml_tools.parse(
f"""
desc_element = xml_tools.parse(f"""
<description xmlns="urn:xmpp:jingle:apps:rtp:1" media="audio">
<rtcp-fb xmlns="{NS_JINGLE_RTP_RTCP_FB}" type="nack" subtype="pli"/>
<rtcp-fb-trr-int xmlns="{NS_JINGLE_RTP_RTCP_FB}" value="100"/>
</description>
"""
)
""")
media_data = {}
xep_0293._parse_description_trigger(desc_element, media_data)
@ -183,16 +181,14 @@ class TestXEP0293:
"""Parsing of <rtcp-fb> and <rtcp-fb-trr-int> elements from a payload type."""
xep_0293 = XEP_0293(host)
desc_element = xml_tools.parse(
f"""
desc_element = xml_tools.parse(f"""
<description xmlns="urn:xmpp:jingle:apps:rtp:1" media="audio">
<payload-type id="96">
<rtcp-fb xmlns="{NS_JINGLE_RTP_RTCP_FB}" type="nack" subtype="pli"/>
<rtcp-fb-trr-int xmlns="{NS_JINGLE_RTP_RTCP_FB}" value="100"/>
</payload-type>
</description>
"""
)
""")
media_data = {}
payload_type_elt = desc_element.firstChildElement()

View file

@ -10,7 +10,6 @@ except exceptions.MissingModule as e:
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")

View file

@ -35,7 +35,6 @@ from libervia.backend.plugins.plugin_xep_0420 import (
from libervia.backend.tools.xml_tools import ElementParser
from twisted.words.xish import domish
__all__ = [ # pylint: disable=unused-variable
"test_unpack_matches_original",
"test_affixes_included",
@ -150,13 +149,11 @@ def test_affixes_included() -> None: # pylint: disable=missing-function-docstri
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)
@ -186,13 +183,11 @@ def test_all_affixes_verified() -> None: # pylint: disable=missing-function-doc
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)
@ -229,11 +224,9 @@ def test_incomplete_affixes() -> None: # pylint: disable=missing-function-docst
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(
@ -259,14 +252,12 @@ def test_incomplete_affixes() -> None: # pylint: disable=missing-function-docst
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(
@ -284,11 +275,9 @@ def test_rpad_affix() -> None: # pylint: disable=missing-function-docstring
)
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)
@ -321,9 +310,7 @@ 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)
@ -338,11 +325,9 @@ def test_to_affix() -> None: # pylint: disable=missing-function-docstring
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)
@ -366,13 +351,11 @@ 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)
@ -387,11 +370,9 @@ def test_from_affix() -> None: # pylint: disable=missing-function-docstring
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)
@ -415,13 +396,11 @@ 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)
@ -446,14 +425,12 @@ def test_custom_affixes() -> None: # pylint: disable=missing-function-docstring
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(
@ -462,12 +439,10 @@ def test_custom_affixes() -> None: # pylint: disable=missing-function-docstring
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)
@ -526,12 +501,10 @@ def test_non_encryptable_elements() -> None: # pylint: disable=missing-function
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"))
@ -552,9 +525,7 @@ 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)
@ -582,9 +553,7 @@ 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)