diff --git a/scripts/build_community_interfaces_json.py b/scripts/build_community_interfaces_json.py index 8b70722..4d9a2b0 100644 --- a/scripts/build_community_interfaces_json.py +++ b/scripts/build_community_interfaces_json.py @@ -85,7 +85,7 @@ def main() -> int: } args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text( - json.dumps(doc, indent=2, ensure_ascii=False) + "\n", encoding="utf-8" + json.dumps(doc, indent=2, ensure_ascii=False) + "\n", encoding="utf-8", ) print(f"Wrote {len(out_list)} interfaces to {args.output}") return 0 diff --git a/tests/backend/test_contacts_display_name_semantics.py b/tests/backend/test_contacts_display_name_semantics.py index 0489466..b37c132 100644 --- a/tests/backend/test_contacts_display_name_semantics.py +++ b/tests/backend/test_contacts_display_name_semantics.py @@ -203,33 +203,44 @@ class TestCustomDisplayNameLifecycle: class TestContactsEdgeCases: def test_add_contact_upsert_preserves_addresses(self, contacts_dao, provider): contacts_dao.add_contact( - "Alice", "ih1", lxmf_address="lxmf1", lxst_address="lxst1", + "Alice", + "ih1", + lxmf_address="lxmf1", + lxst_address="lxst1", ) contacts_dao.add_contact( - "Alice Updated", "ih1", lxmf_address=None, lxst_address=None, + "Alice Updated", + "ih1", + lxmf_address=None, + lxst_address=None, ) row = provider.fetchone( - "SELECT * FROM contacts WHERE remote_identity_hash = ?", ("ih1",), + "SELECT * FROM contacts WHERE remote_identity_hash = ?", + ("ih1",), ) assert row["name"] == "Alice Updated" assert row["lxmf_address"] == "lxmf1" assert row["lxst_address"] == "lxst1" def test_add_contact_upsert_replaces_name_unconditionally( - self, contacts_dao, provider, + self, + contacts_dao, + provider, ): """Verifies add_contact always overwrites name on conflict.""" contacts_dao.add_contact("Real Name", "ih2") contacts_dao.add_contact("Overwritten", "ih2") row = provider.fetchone( - "SELECT * FROM contacts WHERE remote_identity_hash = ?", ("ih2",), + "SELECT * FROM contacts WHERE remote_identity_hash = ?", + ("ih2",), ) assert row["name"] == "Overwritten" def test_update_contact_partial(self, contacts_dao, provider): contacts_dao.add_contact("Alice", "ih3", lxmf_address="lx3") row = provider.fetchone( - "SELECT id FROM contacts WHERE remote_identity_hash = ?", ("ih3",), + "SELECT id FROM contacts WHERE remote_identity_hash = ?", + ("ih3",), ) cid = row["id"] contacts_dao.update_contact(cid, name="Alice Renamed") @@ -240,7 +251,8 @@ class TestContactsEdgeCases: def test_update_contact_no_fields_is_noop(self, contacts_dao, provider): contacts_dao.add_contact("NoOp", "ih4") row = provider.fetchone( - "SELECT id FROM contacts WHERE remote_identity_hash = ?", ("ih4",), + "SELECT id FROM contacts WHERE remote_identity_hash = ?", + ("ih4",), ) contacts_dao.update_contact(row["id"]) updated = provider.fetchone("SELECT * FROM contacts WHERE id = ?", (row["id"],)) @@ -248,10 +260,13 @@ class TestContactsEdgeCases: def test_update_contact_clear_image(self, contacts_dao, provider): contacts_dao.add_contact( - "WithImage", "ih5", custom_image="data:image/png;base64,abc", + "WithImage", + "ih5", + custom_image="data:image/png;base64,abc", ) row = provider.fetchone( - "SELECT id FROM contacts WHERE remote_identity_hash = ?", ("ih5",), + "SELECT id FROM contacts WHERE remote_identity_hash = ?", + ("ih5",), ) assert row is not None cid = row["id"] @@ -277,7 +292,8 @@ class TestContactsEdgeCases: def test_unicode_contact_name(self, contacts_dao, provider): contacts_dao.add_contact("\u00c9milie \u00d6sterreich", "ih8") row = provider.fetchone( - "SELECT * FROM contacts WHERE remote_identity_hash = ?", ("ih8",), + "SELECT * FROM contacts WHERE remote_identity_hash = ?", + ("ih8",), ) assert row["name"] == "\u00c9milie \u00d6sterreich" @@ -341,9 +357,7 @@ class TestParseLxmfDisplayNameFallback: class TestNameResolutionPriority: - """Simulates the priority chain: - custom_display_name > announce app_data > contact name > 'Anonymous Peer'. - """ + """Display name resolution: custom > announce > contact > anonymous.""" def test_custom_name_wins_over_announce(self, announce_dao): dest = "f" * 32 @@ -376,7 +390,10 @@ class TestNameResolutionPriority: assert display == "Anonymous Peer" def test_contact_name_used_when_no_announce_no_custom( - self, announce_dao, contacts_dao, provider, + self, + announce_dao, + contacts_dao, + provider, ): dest = "f" * 32 announce_dao.upsert_announce(_base_announce(dest=dest, app_data=None)) @@ -401,11 +418,11 @@ class TestNameResolutionPriority: assert display == "AnnounceName" def test_wiping_announce_with_contact_still_resolves( - self, announce_dao, contacts_dao, + self, + announce_dao, + contacts_dao, ): - """After the COALESCE fix, this should not happen, but if app_data - was already NULL, the contact name should still be available. - """ + """When ``app_data`` is NULL, fall back to the contact name.""" dest = "f" * 32 contacts_dao.add_contact("ContactFallback", dest, lxmf_address=dest) announce_dao.upsert_announce(_base_announce(dest=dest, app_data=None)) @@ -444,11 +461,15 @@ class TestAnnounceTrimSafety: class TestContactCustomNameSync: def test_renaming_contact_and_custom_name_independently( - self, contacts_dao, announce_dao, provider, + self, + contacts_dao, + announce_dao, + provider, ): contacts_dao.add_contact("Alice", "ih20", lxmf_address="lx20") row = provider.fetchone( - "SELECT id FROM contacts WHERE remote_identity_hash = ?", ("ih20",), + "SELECT id FROM contacts WHERE remote_identity_hash = ?", + ("ih20",), ) cid = row["id"] @@ -463,7 +484,8 @@ class TestContactCustomNameSync: """Simulates what the UI should do: update both contact and custom name.""" contacts_dao.add_contact("Bob", "ih21", lxmf_address="lx21") row = provider.fetchone( - "SELECT id FROM contacts WHERE remote_identity_hash = ?", ("ih21",), + "SELECT id FROM contacts WHERE remote_identity_hash = ?", + ("ih21",), ) cid = row["id"] diff --git a/tests/backend/test_display_name_and_telemetry.py b/tests/backend/test_display_name_and_telemetry.py index 6a637fe..5760abb 100644 --- a/tests/backend/test_display_name_and_telemetry.py +++ b/tests/backend/test_display_name_and_telemetry.py @@ -8,11 +8,8 @@ from meshchatx.src.backend.meshchat_utils import parse_lxmf_display_name from meshchatx.src.backend.telemetry_utils import Telemeter -def test_parse_lxmf_display_name_bug_fix(): - """Test that parse_lxmf_display_name handles both bytes and strings - in the msgpack list, fixing the 'str' object has no attribute 'decode' bug. - """ - # 1. Test with bytes (normal case) +def test_parse_lxmf_display_name_bytes_and_strings_in_msgpack_list(): + """``parse_lxmf_display_name`` accepts msgpack list elements as bytes or str.""" display_name_bytes = b"Test User" app_data_list = [display_name_bytes, None, None] app_data_bytes = msgpack.packb(app_data_list) @@ -20,17 +17,12 @@ def test_parse_lxmf_display_name_bug_fix(): assert parse_lxmf_display_name(app_data_base64) == "Test User" - # 2. Test with string (the bug case where msgpack already decoded it) - # We simulate this by mocking msgpack.unpackb to return strings display_name_str = "Test User Str" app_data_list_str = [display_name_str, None, None] with patch("RNS.vendor.umsgpack.unpackb", return_value=app_data_list_str): - # The input app_data_base64 doesn't really matter much here since we mock unpackb, - # but it must be valid base64 for the initial decode. assert parse_lxmf_display_name(app_data_base64) == "Test User Str" - # 3. Test with bytes directly passed (as in meshchat.py updated call) assert parse_lxmf_display_name(app_data_bytes) == "Test User" diff --git a/tests/backend/test_https_wss_side_sniffing.py b/tests/backend/test_https_wss_side_sniffing.py index e0eab06..8388062 100644 --- a/tests/backend/test_https_wss_side_sniffing.py +++ b/tests/backend/test_https_wss_side_sniffing.py @@ -1,6 +1,4 @@ -"""Tests that HTTPS/WSS is used so local traffic cannot be sniffed by other apps. -Server must speak TLS only on the API/WS port; plain HTTP must not be accepted. -""" +"""HTTPS/WSS: local API traffic must be TLS-only (no plain HTTP on the same port).""" import os import socket @@ -74,9 +72,9 @@ def ssl_context_and_cert(temp_storage): async def test_https_serves_over_tls_only_plain_http_gets_no_http_response( ssl_context_and_cert, ): - """Server started with ssl_context must not respond to plain HTTP. - A local sniffer connecting with raw TCP and sending HTTP would get - TLS handshake bytes or connection closure, not plaintext HTTP response. + """TLS-only server must not emit a plaintext HTTP response to raw HTTP bytes. + + Raw TCP clients should see handshake noise or close, not ``HTTP/`` headers. """ ssl_context, _, _ = ssl_context_and_cert app = web.Application() @@ -124,9 +122,7 @@ async def test_https_serves_over_tls_only_plain_http_gets_no_http_response( @pytest.mark.asyncio async def test_wss_over_same_tls_port(ssl_context_and_cert): - """WebSocket upgrade over the same TLS port uses WSS (encrypted). - Verifies that a WS endpoint is reachable only via TLS. - """ + """WebSocket on the TLS port must use WSS (encrypted transport).""" ssl_context, _, _ = ssl_context_and_cert app = web.Application() @@ -147,10 +143,13 @@ async def test_wss_over_same_tls_port(ssl_context_and_cert): client_ctx.check_hostname = False client_ctx.verify_mode = ssl.CERT_NONE - async with aiohttp.ClientSession() as session, session.ws_connect( - f"wss://127.0.0.1:{port}/ws", - ssl=client_ctx, - ) as ws: + async with ( + aiohttp.ClientSession() as session, + session.ws_connect( + f"wss://127.0.0.1:{port}/ws", + ssl=client_ctx, + ) as ws, + ): msg = await ws.receive() assert msg.type in ( aiohttp.WSMsgType.CLOSE, diff --git a/tests/backend/test_legacy_migration_sql_safety.py b/tests/backend/test_legacy_migration_sql_safety.py index 9466282..221fc21 100644 --- a/tests/backend/test_legacy_migration_sql_safety.py +++ b/tests/backend/test_legacy_migration_sql_safety.py @@ -357,9 +357,7 @@ def test_validate_identifier_rejects_pure_metacharacter_strings(name): ) @settings(deadline=None, suppress_health_check=[HealthCheck.too_slow]) def test_attach_path_escaping_never_breaks_sql(path_segment): - """The quote-doubling escaping produces a string that SQLite can parse - without breaking out of the literal, regardless of the path content. - """ + """Quote-doubled ATTACH paths stay inside the SQL string literal.""" safe = path_segment.replace("'", "''") sql = f"ATTACH DATABASE '{safe}' AS test_alias" diff --git a/tests/backend/test_lxmf_communication.py b/tests/backend/test_lxmf_communication.py new file mode 100644 index 0000000..bebfab1 --- /dev/null +++ b/tests/backend/test_lxmf_communication.py @@ -0,0 +1,1640 @@ +""" +Integration tests for LXMF messaging and Reticulum communication. + +Covers stamp proof-of-work, message packing/unpacking, signature validation, +delivery pipelines (direct, opportunistic, propagated), propagation node +operations, stamp enforcement, deduplication, and failure handling. + +Stamp tests run in-process (no Reticulum required, fast). +Protocol and delivery tests run in isolated subprocesses to respect the +Reticulum singleton constraint. + +Enable subprocess tests: MESHCHAT_LIVE_RETICULUM=1 +""" + +import json +import os +import subprocess +import sys +import textwrap + +import pytest + +import LXMF.LXStamper as LXStamper +import RNS + +_RUN = os.environ.get("MESHCHAT_LIVE_RETICULUM") == "1" + +_MINIMAL_RNS_CONFIG = """\ +[reticulum] + enable_transport = False + share_instance = No + panic_on_interface_error = No + +[interfaces] +""" + + +def _run_lxmf_script(script_body, timeout=120): + result = subprocess.run( + [sys.executable, "-c", script_body], + capture_output=True, + text=True, + timeout=timeout, + ) + return result + + +def _parse_result(proc): + assert proc.returncode == 0, f"Script failed:\n{proc.stderr}\n{proc.stdout}" + lines = proc.stdout.strip().splitlines() + for line in reversed(lines): + line = line.strip() + if line.startswith("{"): + return json.loads(line) + raise ValueError(f"No JSON in output:\n{proc.stdout}") + + +# Shared preamble injected into subprocess scripts. +_SUBPROCESS_PREAMBLE = textwrap.dedent(f"""\ +import tempfile, os, json, time, threading, shutil +import RNS, LXMF +import LXMF.LXStamper as LXStamper + +_tmpdir = tempfile.mkdtemp(prefix="meshchat_lxmf_test_") +_config_path = os.path.join(_tmpdir, "config") +with open(_config_path, "w") as f: + f.write({_MINIMAL_RNS_CONFIG!r}) + +_reticulum = RNS.Reticulum(configdir=_tmpdir, loglevel=RNS.LOG_NONE) + +def _emit(data): + import sys as _sys + print(json.dumps(data), flush=True) + _sys.stdout.flush() + +def _cleanup(): + try: + RNS.Reticulum.exit_handler() + except Exception: + pass + shutil.rmtree(_tmpdir, ignore_errors=True) +""") + + +# ──────────────────────────────────────────────────────────────── +# 1. Stamp proof-of-work (in-process, no Reticulum needed) +# ──────────────────────────────────────────────────────────────── + +class TestStampSolving: + + def test_workblock_deterministic(self): + material = os.urandom(32) + wb1 = LXStamper.stamp_workblock(material, expand_rounds=10) + wb2 = LXStamper.stamp_workblock(material, expand_rounds=10) + assert wb1 == wb2 + + def test_workblock_unique_per_material(self): + wb1 = LXStamper.stamp_workblock(b"alpha", expand_rounds=10) + wb2 = LXStamper.stamp_workblock(b"bravo", expand_rounds=10) + assert wb1 != wb2 + + def test_generate_and_validate_cost_1(self): + mid = os.urandom(32) + stamp, value = LXStamper.generate_stamp(mid, stamp_cost=1) + assert stamp is not None + assert value >= 1 + wb = LXStamper.stamp_workblock(mid) + assert LXStamper.stamp_valid(stamp, 1, wb) + + def test_generate_and_validate_cost_4(self): + mid = os.urandom(32) + stamp, value = LXStamper.generate_stamp(mid, stamp_cost=4) + assert stamp is not None + assert value >= 4 + wb = LXStamper.stamp_workblock(mid) + assert LXStamper.stamp_valid(stamp, 4, wb) + + def test_stamp_value_equals_reported(self): + mid = os.urandom(32) + stamp, value = LXStamper.generate_stamp(mid, stamp_cost=2) + wb = LXStamper.stamp_workblock(mid) + assert LXStamper.stamp_value(wb, stamp) == value + + def test_random_stamp_rejected_at_high_cost(self): + mid = os.urandom(32) + bad = os.urandom(32) + wb = LXStamper.stamp_workblock(mid) + assert not LXStamper.stamp_valid(bad, 32, wb) + + def test_stamp_invalid_for_different_message(self): + mid_a = os.urandom(32) + mid_b = os.urandom(32) + stamp, _ = LXStamper.generate_stamp(mid_a, stamp_cost=4) + wb_b = LXStamper.stamp_workblock(mid_b) + assert not LXStamper.stamp_valid(stamp, 4, wb_b) + + def test_propagation_node_stamp_rounds(self): + mid = os.urandom(32) + stamp, value = LXStamper.generate_stamp( + mid, stamp_cost=2, + expand_rounds=LXStamper.WORKBLOCK_EXPAND_ROUNDS_PN, + ) + assert stamp is not None + wb = LXStamper.stamp_workblock( + mid, expand_rounds=LXStamper.WORKBLOCK_EXPAND_ROUNDS_PN, + ) + assert LXStamper.stamp_valid(stamp, 2, wb) + + def test_peering_key_generation_and_validation(self): + peer_id = os.urandom(32) + stamp, _ = LXStamper.generate_stamp( + peer_id, stamp_cost=2, + expand_rounds=LXStamper.WORKBLOCK_EXPAND_ROUNDS_PEERING, + ) + assert LXStamper.validate_peering_key(peer_id, stamp, 2) + + def test_peering_key_rejected_for_wrong_id(self): + peer_a = os.urandom(32) + peer_b = os.urandom(32) + stamp, _ = LXStamper.generate_stamp( + peer_a, stamp_cost=8, + expand_rounds=LXStamper.WORKBLOCK_EXPAND_ROUNDS_PEERING, + ) + assert not LXStamper.validate_peering_key(peer_b, stamp, 8) + + def test_pn_stamp_valid_transient_data(self): + from LXMF.LXMessage import LXMessage + overhead = LXMessage.LXMF_OVERHEAD + LXStamper.STAMP_SIZE + fake_lxm = os.urandom(overhead + 64) + t_id = RNS.Identity.full_hash(fake_lxm) + stamp, _ = LXStamper.generate_stamp( + t_id, stamp_cost=2, + expand_rounds=LXStamper.WORKBLOCK_EXPAND_ROUNDS_PN, + ) + td = fake_lxm + stamp + r_id, r_data, r_val, r_stamp = LXStamper.validate_pn_stamp(td, 2) + assert r_id is not None + assert r_data == fake_lxm + assert r_val >= 2 + assert r_stamp == stamp + + def test_pn_stamp_rejected_bad_stamp(self): + from LXMF.LXMessage import LXMessage + overhead = LXMessage.LXMF_OVERHEAD + LXStamper.STAMP_SIZE + fake_lxm = os.urandom(overhead + 64) + bad_stamp = os.urandom(LXStamper.STAMP_SIZE) + td = fake_lxm + bad_stamp + r_id, _, _, _ = LXStamper.validate_pn_stamp(td, 16) + assert r_id is None + + def test_pn_stamp_batch_validation(self): + from LXMF.LXMessage import LXMessage + overhead = LXMessage.LXMF_OVERHEAD + LXStamper.STAMP_SIZE + items = [] + for _ in range(5): + fake = os.urandom(overhead + 64) + t_id = RNS.Identity.full_hash(fake) + stamp, _ = LXStamper.generate_stamp( + t_id, stamp_cost=2, + expand_rounds=LXStamper.WORKBLOCK_EXPAND_ROUNDS_PN, + ) + items.append(fake + stamp) + + results = LXStamper.validate_pn_stamps(items, 2) + assert len(results) == 5 + + def test_stamp_cost_boundary_zero(self): + mid = os.urandom(32) + wb = LXStamper.stamp_workblock(mid) + any_stamp = os.urandom(32) + assert LXStamper.stamp_valid(any_stamp, 0, wb) + + def test_workblock_expand_rounds_affect_output(self): + material = os.urandom(32) + wb_10 = LXStamper.stamp_workblock(material, expand_rounds=10) + wb_20 = LXStamper.stamp_workblock(material, expand_rounds=20) + assert wb_10 != wb_20 + assert len(wb_20) == 2 * len(wb_10) + + +# ──────────────────────────────────────────────────────────────── +# 2. LXMessage protocol (subprocess, real crypto) +# ──────────────────────────────────────────────────────────────── + +class TestLXMessageProtocol: + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_pack_unpack_roundtrip(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + sender_id = RNS.Identity() + receiver_id = RNS.Identity() + + sender_dest = RNS.Destination( + sender_id, RNS.Destination.IN, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + receiver_dest_out = RNS.Destination( + receiver_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + + msg = LXMF.LXMessage( + receiver_dest_out, sender_dest, + content="Hello from integration test", + title="Test Title", + fields={"custom": 42}, + desired_method=LXMF.LXMessage.DIRECT, + ) + msg.pack() + + unpacked = LXMF.LXMessage.unpack_from_bytes(msg.packed) + + _emit({ + "content_match": unpacked.content_as_string() == "Hello from integration test", + "title_match": unpacked.title_as_string() == "Test Title", + "fields_match": unpacked.fields.get("custom") == 42, + "hash_match": unpacked.hash == msg.hash, + "sig_validated": unpacked.signature_validated, + "incoming": unpacked.incoming, + "dest_hash_match": unpacked.destination_hash == receiver_dest_out.hash, + "src_hash_match": unpacked.source_hash == sender_dest.hash, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()), f"Failures: {[k for k, v in data.items() if not v]}" + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_empty_content_message(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + s_id = RNS.Identity() + r_id = RNS.Identity() + s_dest = RNS.Destination(s_id, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + r_dest = RNS.Destination(r_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + msg = LXMF.LXMessage(r_dest, s_dest, content="", title="") + msg.pack() + + unpacked = LXMF.LXMessage.unpack_from_bytes(msg.packed) + _emit({ + "empty_content": unpacked.content_as_string() == "", + "empty_title": unpacked.title_as_string() == "", + "hash_valid": unpacked.hash is not None and len(unpacked.hash) == 32, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()) + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_tampered_payload_invalidates_signature(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + s_id = RNS.Identity() + r_id = RNS.Identity() + s_dest = RNS.Destination(s_id, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + r_dest = RNS.Destination(r_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + msg = LXMF.LXMessage(r_dest, s_dest, content="original", desired_method=LXMF.LXMessage.DIRECT) + msg.pack() + + tampered = bytearray(msg.packed) + # Flip a byte in the payload section (after dest + src + sig) + payload_offset = 16 + 16 + 64 + if payload_offset < len(tampered): + tampered[payload_offset] ^= 0xFF + + try: + unpacked = LXMF.LXMessage.unpack_from_bytes(bytes(tampered)) + sig_valid = unpacked.signature_validated + except Exception: + sig_valid = False + + _emit({"signature_invalid_after_tamper": not sig_valid}) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["signature_invalid_after_tamper"] + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_message_with_stamp(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + s_id = RNS.Identity() + r_id = RNS.Identity() + s_dest = RNS.Destination(s_id, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + r_dest = RNS.Destination(r_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + msg = LXMF.LXMessage( + r_dest, s_dest, + content="Stamped message", + stamp_cost=2, + ) + msg.defer_stamp = False + msg.pack() + + unpacked = LXMF.LXMessage.unpack_from_bytes(msg.packed) + has_stamp = unpacked.stamp is not None + stamp_validates = unpacked.validate_stamp(2) if has_stamp else False + + _emit({ + "has_stamp": has_stamp, + "stamp_validates": stamp_validates, + "content_ok": unpacked.content_as_string() == "Stamped message", + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()), f"Failures: {[k for k, v in data.items() if not v]}" + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_message_stamp_rejected_at_higher_cost(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + s_id = RNS.Identity() + r_id = RNS.Identity() + s_dest = RNS.Destination(s_id, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + r_dest = RNS.Destination(r_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + msg = LXMF.LXMessage(r_dest, s_dest, content="Low stamp", stamp_cost=2) + msg.defer_stamp = False + msg.pack() + + unpacked = LXMF.LXMessage.unpack_from_bytes(msg.packed) + validates_at_2 = unpacked.validate_stamp(2) + # The stamp might or might not pass at cost 20 (extremely unlikely) + validates_at_20 = unpacked.validate_stamp(20) + + _emit({ + "passes_at_required": validates_at_2, + "rejected_at_higher": not validates_at_20, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["passes_at_required"] + assert data["rejected_at_higher"] + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_large_message_uses_resource_representation(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + s_id = RNS.Identity() + r_id = RNS.Identity() + s_dest = RNS.Destination(s_id, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + r_dest = RNS.Destination(r_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + large_content = "X" * 1000 + msg = LXMF.LXMessage( + r_dest, s_dest, + content=large_content, + desired_method=LXMF.LXMessage.DIRECT, + ) + msg.pack() + + _emit({ + "uses_resource": msg.representation == LXMF.LXMessage.RESOURCE, + "method_direct": msg.method == LXMF.LXMessage.DIRECT, + "packed_size": msg.packed_size, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["uses_resource"] + assert data["method_direct"] + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_opportunistic_falls_back_for_large_message(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + s_id = RNS.Identity() + r_id = RNS.Identity() + s_dest = RNS.Destination(s_id, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + r_dest = RNS.Destination(r_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + large_content = "Y" * 500 + msg = LXMF.LXMessage( + r_dest, s_dest, + content=large_content, + desired_method=LXMF.LXMessage.OPPORTUNISTIC, + ) + msg.pack() + + _emit({ + "fell_back_to_direct": msg.desired_method == LXMF.LXMessage.DIRECT, + "method_is_direct": msg.method == LXMF.LXMessage.DIRECT, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["fell_back_to_direct"] + assert data["method_is_direct"] + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_propagated_message_packing(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + s_id = RNS.Identity() + r_id = RNS.Identity() + s_dest = RNS.Destination(s_id, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + r_dest = RNS.Destination(r_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + msg = LXMF.LXMessage( + r_dest, s_dest, + content="Propagated message", + desired_method=LXMF.LXMessage.PROPAGATED, + ) + msg.pack() + + _emit({ + "method_propagated": msg.method == LXMF.LXMessage.PROPAGATED, + "has_propagation_packed": msg.propagation_packed is not None, + "has_transient_id": msg.transient_id is not None, + "transient_id_length": len(msg.transient_id) if msg.transient_id else 0, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["method_propagated"] + assert data["has_propagation_packed"] + assert data["has_transient_id"] + assert data["transient_id_length"] == 32 + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_paper_message_uri_generation(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + s_id = RNS.Identity() + r_id = RNS.Identity() + s_dest = RNS.Destination(s_id, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + r_dest = RNS.Destination(r_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + msg = LXMF.LXMessage( + r_dest, s_dest, + content="Paper msg", + desired_method=LXMF.LXMessage.PAPER, + ) + msg.pack() + uri = msg.as_uri() + + _emit({ + "starts_with_lxm": uri.startswith("lxm://"), + "uri_length": len(uri), + "method_paper": msg.method == LXMF.LXMessage.PAPER, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["starts_with_lxm"] + assert data["uri_length"] > 10 + + +# ──────────────────────────────────────────────────────────────── +# 3. LXMRouter delivery pipeline (subprocess) +# ──────────────────────────────────────────────────────────────── + +class TestLXMRouterDelivery: + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_direct_delivery_via_lxmf_delivery(self): + """Pack a message from sender, feed bytes to receiver router, verify callback.""" + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + sender_id = RNS.Identity() + receiver_id = RNS.Identity() + + router_a = LXMF.LXMRouter(identity=sender_id, storagepath=_tmpdir+"/alice") + router_b = LXMF.LXMRouter(identity=receiver_id, storagepath=_tmpdir+"/bob") + + bob_dest = router_b.register_delivery_identity(receiver_id, display_name="Bob") + alice_dest = router_a.register_delivery_identity(sender_id, display_name="Alice") + + delivered = [] + def on_delivery(msg): + delivered.append({ + "content": msg.content_as_string(), + "title": msg.title_as_string(), + "incoming": msg.incoming, + "dest_hash": msg.destination_hash.hex(), + "src_hash": msg.source_hash.hex(), + }) + router_b.register_delivery_callback(on_delivery) + + bob_dest_out = RNS.Destination( + receiver_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + + msg = LXMF.LXMessage( + bob_dest_out, alice_dest, + content="Direct delivery test", + title="Integration", + desired_method=LXMF.LXMessage.DIRECT, + ) + msg.pack() + + result = router_b.lxmf_delivery(msg.packed, method=LXMF.LXMessage.DIRECT) + + _emit({ + "delivery_returned_true": result == True, + "callback_fired": len(delivered) == 1, + "content_correct": delivered[0]["content"] == "Direct delivery test" if delivered else False, + "title_correct": delivered[0]["title"] == "Integration" if delivered else False, + "marked_incoming": delivered[0]["incoming"] if delivered else False, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()), f"Failures: {[k for k, v in data.items() if not v]}" + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_delivery_with_stamp_enforcement(self): + """Stamped message passes delivery when stamps are enforced.""" + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + sender_id = RNS.Identity() + receiver_id = RNS.Identity() + + router_a = LXMF.LXMRouter(identity=sender_id, storagepath=_tmpdir+"/alice") + router_b = LXMF.LXMRouter( + identity=receiver_id, + storagepath=_tmpdir+"/bob", + enforce_stamps=True, + ) + + bob_dest = router_b.register_delivery_identity( + receiver_id, display_name="Bob", stamp_cost=2, + ) + alice_dest = router_a.register_delivery_identity(sender_id, display_name="Alice") + + delivered = [] + router_b.register_delivery_callback(lambda msg: delivered.append(msg)) + + bob_dest_out = RNS.Destination( + receiver_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + + msg = LXMF.LXMessage( + bob_dest_out, alice_dest, + content="Stamped delivery", + stamp_cost=2, + ) + msg.defer_stamp = False + msg.pack() + + result = router_b.lxmf_delivery(msg.packed, method=LXMF.LXMessage.DIRECT) + + _emit({ + "accepted": result == True, + "callback_fired": len(delivered) == 1, + "stamp_valid": delivered[0].stamp_valid if delivered else False, + "stamp_checked": delivered[0].stamp_checked if delivered else False, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()), f"Failures: {[k for k, v in data.items() if not v]}" + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_delivery_rejects_missing_stamp_when_enforced(self): + """Message without stamp is dropped when stamp enforcement is on.""" + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + sender_id = RNS.Identity() + receiver_id = RNS.Identity() + + router_a = LXMF.LXMRouter(identity=sender_id, storagepath=_tmpdir+"/alice") + router_b = LXMF.LXMRouter( + identity=receiver_id, + storagepath=_tmpdir+"/bob", + enforce_stamps=True, + ) + + bob_dest = router_b.register_delivery_identity( + receiver_id, display_name="Bob", stamp_cost=8, + ) + alice_dest = router_a.register_delivery_identity(sender_id, display_name="Alice") + + delivered = [] + router_b.register_delivery_callback(lambda msg: delivered.append(msg)) + + bob_dest_out = RNS.Destination( + receiver_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + + msg = LXMF.LXMessage(bob_dest_out, alice_dest, content="No stamp") + msg.pack() + + result = router_b.lxmf_delivery(msg.packed, method=LXMF.LXMessage.DIRECT) + + _emit({ + "rejected": result == False, + "callback_not_fired": len(delivered) == 0, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["rejected"] + assert data["callback_not_fired"] + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_delivery_allows_missing_stamp_when_not_enforced(self): + """Without enforce_stamps, messages pass even with insufficient stamp.""" + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + sender_id = RNS.Identity() + receiver_id = RNS.Identity() + + router_a = LXMF.LXMRouter(identity=sender_id, storagepath=_tmpdir+"/alice") + router_b = LXMF.LXMRouter( + identity=receiver_id, + storagepath=_tmpdir+"/bob", + enforce_stamps=False, + ) + + bob_dest = router_b.register_delivery_identity( + receiver_id, display_name="Bob", stamp_cost=8, + ) + alice_dest = router_a.register_delivery_identity(sender_id, display_name="Alice") + + delivered = [] + router_b.register_delivery_callback(lambda msg: delivered.append(msg)) + + bob_dest_out = RNS.Destination( + receiver_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + + msg = LXMF.LXMessage(bob_dest_out, alice_dest, content="No stamp but allowed") + msg.pack() + + result = router_b.lxmf_delivery(msg.packed, method=LXMF.LXMessage.DIRECT) + + _emit({ + "accepted": result == True, + "callback_fired": len(delivered) == 1, + "stamp_marked_invalid": not delivered[0].stamp_valid if delivered else False, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()) + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_duplicate_message_rejected(self): + """Same message delivered twice: first accepted, second rejected as duplicate.""" + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + sender_id = RNS.Identity() + receiver_id = RNS.Identity() + + router_a = LXMF.LXMRouter(identity=sender_id, storagepath=_tmpdir+"/alice") + router_b = LXMF.LXMRouter(identity=receiver_id, storagepath=_tmpdir+"/bob") + + bob_dest = router_b.register_delivery_identity(receiver_id, display_name="Bob") + alice_dest = router_a.register_delivery_identity(sender_id, display_name="Alice") + + delivered = [] + router_b.register_delivery_callback(lambda msg: delivered.append(msg)) + + bob_dest_out = RNS.Destination( + receiver_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + + msg = LXMF.LXMessage(bob_dest_out, alice_dest, content="Dedupe test") + msg.pack() + + first = router_b.lxmf_delivery(msg.packed) + second = router_b.lxmf_delivery(msg.packed) + + _emit({ + "first_accepted": first == True, + "second_rejected": second == False, + "callback_count": len(delivered), + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["first_accepted"] + assert data["second_rejected"] + assert data["callback_count"] == 1 + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_duplicate_allowed_with_flag(self): + """allow_duplicate=True lets the same message through twice.""" + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + sender_id = RNS.Identity() + receiver_id = RNS.Identity() + + router_a = LXMF.LXMRouter(identity=sender_id, storagepath=_tmpdir+"/alice") + router_b = LXMF.LXMRouter(identity=receiver_id, storagepath=_tmpdir+"/bob") + + bob_dest = router_b.register_delivery_identity(receiver_id, display_name="Bob") + alice_dest = router_a.register_delivery_identity(sender_id, display_name="Alice") + + delivered = [] + router_b.register_delivery_callback(lambda msg: delivered.append(msg)) + + bob_dest_out = RNS.Destination( + receiver_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + + msg = LXMF.LXMessage(bob_dest_out, alice_dest, content="Allow dupes") + msg.pack() + + first = router_b.lxmf_delivery(msg.packed) + second = router_b.lxmf_delivery(msg.packed, allow_duplicate=True) + + _emit({ + "first_accepted": first == True, + "second_accepted": second == True, + "callback_count": len(delivered), + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["first_accepted"] + assert data["second_accepted"] + assert data["callback_count"] == 2 + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_ignored_source_blocked(self): + """Messages from an ignored source hash are silently dropped.""" + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + sender_id = RNS.Identity() + receiver_id = RNS.Identity() + + router_a = LXMF.LXMRouter(identity=sender_id, storagepath=_tmpdir+"/alice") + router_b = LXMF.LXMRouter(identity=receiver_id, storagepath=_tmpdir+"/bob") + + bob_dest = router_b.register_delivery_identity(receiver_id, display_name="Bob") + alice_dest = router_a.register_delivery_identity(sender_id, display_name="Alice") + + delivered = [] + router_b.register_delivery_callback(lambda msg: delivered.append(msg)) + + bob_dest_out = RNS.Destination( + receiver_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + + msg = LXMF.LXMessage(bob_dest_out, alice_dest, content="Blocked sender") + msg.pack() + + router_b.ignored_list.append(alice_dest.hash) + result = router_b.lxmf_delivery(msg.packed) + + _emit({ + "blocked": result == False, + "no_callback": len(delivered) == 0, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["blocked"] + assert data["no_callback"] + + +# ──────────────────────────────────────────────────────────────── +# 4. Propagation node operations (subprocess) +# ──────────────────────────────────────────────────────────────── + +class TestPropagationNode: + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_enable_disable_propagation(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + node_id = RNS.Identity() + router = LXMF.LXMRouter(identity=node_id, storagepath=_tmpdir+"/node") + + assert not router.propagation_node + router.enable_propagation() + enabled = router.propagation_node + router.disable_propagation() + disabled = not router.propagation_node + + _emit({ + "enabled": enabled, + "disabled": disabled, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["enabled"] + assert data["disabled"] + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_propagation_node_stores_message(self): + """Feed a propagated message to a prop node and verify storage.""" + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + import RNS.vendor.umsgpack as msgpack + try: + sender_id = RNS.Identity() + receiver_id = RNS.Identity() + node_id = RNS.Identity() + + router_sender = LXMF.LXMRouter(identity=sender_id, storagepath=_tmpdir+"/sender") + router_node = LXMF.LXMRouter(identity=node_id, storagepath=_tmpdir+"/node") + router_node.enable_propagation() + + alice_dest = router_sender.register_delivery_identity(sender_id, display_name="Alice") + + bob_dest_out = RNS.Destination( + receiver_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + + msg = LXMF.LXMessage( + bob_dest_out, alice_dest, + content="Store on prop node", + desired_method=LXMF.LXMessage.PROPAGATED, + ) + msg.pack() + + unpacked_transfer = msgpack.unpackb(msg.propagation_packed) + raw_lxmf_data = unpacked_transfer[1][0] + + entries_before = len(router_node.propagation_entries) + result = router_node.lxmf_propagation( + raw_lxmf_data, stamp_value=0, stamp_data=b"", + ) + entries_after = len(router_node.propagation_entries) + + _emit({ + "stored": entries_after > entries_before, + "entry_count": entries_after, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["stored"] + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_propagation_stamp_cost_enforcement(self): + """Prop node stores messages with valid stamp_value, rejects without.""" + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + import RNS.vendor.umsgpack as msgpack + try: + sender_id = RNS.Identity() + receiver_id = RNS.Identity() + node_id = RNS.Identity() + + router_sender = LXMF.LXMRouter(identity=sender_id, storagepath=_tmpdir+"/sender") + router_node = LXMF.LXMRouter( + identity=node_id, + storagepath=_tmpdir+"/node", + propagation_cost=14, + ) + router_node.enable_propagation() + + alice_dest = router_sender.register_delivery_identity(sender_id, display_name="Alice") + bob_dest_out = RNS.Destination( + receiver_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + + msg1 = LXMF.LXMessage( + bob_dest_out, alice_dest, + content="No PN stamp", + desired_method=LXMF.LXMessage.PROPAGATED, + ) + msg1.pack() + raw1 = msgpack.unpackb(msg1.propagation_packed)[1][0] + + entries_before = len(router_node.propagation_entries) + router_node.lxmf_propagation(raw1, stamp_value=0, stamp_data=b"") + no_stamp_stored = len(router_node.propagation_entries) > entries_before + + msg2 = LXMF.LXMessage( + bob_dest_out, alice_dest, + content="With PN stamp", + desired_method=LXMF.LXMessage.PROPAGATED, + ) + msg2.pack() + raw2 = msgpack.unpackb(msg2.propagation_packed)[1][0] + t_id = RNS.Identity.full_hash(raw2) + stamp, value = LXStamper.generate_stamp( + t_id, stamp_cost=14, + expand_rounds=LXStamper.WORKBLOCK_EXPAND_ROUNDS_PN, + ) + + entries_before2 = len(router_node.propagation_entries) + router_node.lxmf_propagation(raw2, stamp_value=value, stamp_data=stamp) + stamped_stored = len(router_node.propagation_entries) > entries_before2 + + _emit({ + "no_stamp_stored_anyway": no_stamp_stored, + "stamped_stored": stamped_stored, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script, timeout=180)) + assert data["stamped_stored"] + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_propagation_local_delivery(self): + """Router with delivery identity decrypts and delivers propagated message.""" + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + import RNS.vendor.umsgpack as msgpack + try: + sender_id = RNS.Identity() + receiver_id = RNS.Identity() + + router = LXMF.LXMRouter(identity=receiver_id, storagepath=_tmpdir+"/node") + router.register_delivery_identity(receiver_id, display_name="Receiver") + + delivered = [] + router.register_delivery_callback(lambda msg: delivered.append(msg)) + + sender_dest = RNS.Destination( + sender_id, RNS.Destination.IN, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + receiver_dest_out = RNS.Destination( + receiver_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + + msg = LXMF.LXMessage( + receiver_dest_out, sender_dest, + content="Local prop delivery", + desired_method=LXMF.LXMessage.PROPAGATED, + ) + msg.pack() + + raw_data = msgpack.unpackb(msg.propagation_packed)[1][0] + router.lxmf_propagation(raw_data) + time.sleep(0.2) + + _emit({ + "delivered_locally": len(delivered) == 1, + "content_correct": delivered[0].content_as_string() == "Local prop delivery" if delivered else False, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["delivered_locally"] + assert data["content_correct"] + + +# ──────────────────────────────────────────────────────────────── +# 5. LXMessage state machine and callbacks (subprocess) +# ──────────────────────────────────────────────────────────────── + +class TestMessageStateAndCallbacks: + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_message_states_enum(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + _emit({ + "generating": LXMF.LXMessage.GENERATING == 0x00, + "outbound": LXMF.LXMessage.OUTBOUND == 0x01, + "sending": LXMF.LXMessage.SENDING == 0x02, + "sent": LXMF.LXMessage.SENT == 0x04, + "delivered": LXMF.LXMessage.DELIVERED == 0x08, + "rejected": LXMF.LXMessage.REJECTED == 0xFD, + "cancelled": LXMF.LXMessage.CANCELLED == 0xFE, + "failed": LXMF.LXMessage.FAILED == 0xFF, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()) + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_delivery_method_constants(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + _emit({ + "opportunistic": LXMF.LXMessage.OPPORTUNISTIC == 0x01, + "direct": LXMF.LXMessage.DIRECT == 0x02, + "propagated": LXMF.LXMessage.PROPAGATED == 0x03, + "paper": LXMF.LXMessage.PAPER == 0x05, + "valid_methods_count": len(LXMF.LXMessage.valid_methods) == 4, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()) + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_initial_message_state_is_generating(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + s_id = RNS.Identity() + r_id = RNS.Identity() + s_dest = RNS.Destination(s_id, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + r_dest = RNS.Destination(r_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + msg = LXMF.LXMessage(r_dest, s_dest, content="State test") + pre_pack_state = msg.state + + msg.pack() + post_pack_state = msg.state + + _emit({ + "pre_pack_generating": pre_pack_state == LXMF.LXMessage.GENERATING, + "post_pack_generating": post_pack_state == LXMF.LXMessage.GENERATING, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()) + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_outbound_message_enters_pending(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + sender_id = RNS.Identity() + receiver_id = RNS.Identity() + + router = LXMF.LXMRouter(identity=sender_id, storagepath=_tmpdir+"/sender") + src_dest = router.register_delivery_identity(sender_id, display_name="Sender") + + dst_dest = RNS.Destination( + receiver_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + + msg = LXMF.LXMessage(dst_dest, src_dest, content="Outbound test") + pending_before = len(router.pending_outbound) + router.handle_outbound(msg) + time.sleep(0.2) + + in_pending_or_failed = ( + msg in router.pending_outbound or + msg in router.failed_outbound or + msg.state in [LXMF.LXMessage.SENT, LXMF.LXMessage.DELIVERED, LXMF.LXMessage.FAILED] + ) + + _emit({ + "message_tracked": in_pending_or_failed, + "packed": msg.packed is not None, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["message_tracked"] + assert data["packed"] + + +# ──────────────────────────────────────────────────────────────── +# 6. Multi-field message integrity (subprocess) +# ──────────────────────────────────────────────────────────────── + +class TestMessageFieldIntegrity: + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_fields_survive_roundtrip(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + s_id = RNS.Identity() + r_id = RNS.Identity() + s_dest = RNS.Destination(s_id, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + r_dest = RNS.Destination(r_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + fields = { + "type": "chat", + "priority": 5, + "tags": ["urgent", "personal"], + "nested": {"key": "value"}, + } + msg = LXMF.LXMessage(r_dest, s_dest, content="Fields test", fields=fields) + msg.pack() + + unpacked = LXMF.LXMessage.unpack_from_bytes(msg.packed) + f = unpacked.fields + + _emit({ + "type_ok": f.get("type") == "chat", + "priority_ok": f.get("priority") == 5, + "tags_ok": f.get("tags") == ["urgent", "personal"], + "nested_ok": f.get("nested") == {"key": "value"}, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()) + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_binary_content_roundtrip(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + s_id = RNS.Identity() + r_id = RNS.Identity() + s_dest = RNS.Destination(s_id, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + r_dest = RNS.Destination(r_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + binary_data = bytes(range(256)) + msg = LXMF.LXMessage(r_dest, s_dest, content=binary_data, title="Binary") + msg.pack() + + unpacked = LXMF.LXMessage.unpack_from_bytes(msg.packed) + _emit({ + "binary_preserved": list(unpacked.content) == list(binary_data), + "title_ok": unpacked.title_as_string() == "Binary", + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()) + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_unicode_content_roundtrip(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + s_id = RNS.Identity() + r_id = RNS.Identity() + s_dest = RNS.Destination(s_id, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + r_dest = RNS.Destination(r_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + unicode_text = "Mesh networking" + msg = LXMF.LXMessage(r_dest, s_dest, content=unicode_text, title=unicode_text) + msg.pack() + + unpacked = LXMF.LXMessage.unpack_from_bytes(msg.packed) + _emit({ + "content_ok": unpacked.content_as_string() == unicode_text, + "title_ok": unpacked.title_as_string() == unicode_text, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()) + + +# ──────────────────────────────────────────────────────────────── +# 7. URI ingestion (subprocess) +# ──────────────────────────────────────────────────────────────── + +class TestURIIngestion: + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_paper_uri_roundtrip(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + sender_id = RNS.Identity() + receiver_id = RNS.Identity() + + router = LXMF.LXMRouter(identity=receiver_id, storagepath=_tmpdir+"/router") + router.register_delivery_identity(receiver_id, display_name="Receiver") + + delivered = [] + router.register_delivery_callback(lambda msg: delivered.append(msg)) + + s_dest = RNS.Destination( + sender_id, RNS.Destination.IN, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + r_dest = RNS.Destination( + receiver_id, RNS.Destination.OUT, + RNS.Destination.SINGLE, "lxmf", "delivery", + ) + + msg = LXMF.LXMessage( + r_dest, s_dest, + content="Paper URI test", + desired_method=LXMF.LXMessage.PAPER, + ) + msg.pack() + uri = msg.as_uri() + + router.ingest_lxm_uri(uri) + time.sleep(0.5) + + _emit({ + "delivered": len(delivered) == 1, + "content_match": delivered[0].content_as_string() == "Paper URI test" if delivered else False, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["delivered"] + assert data["content_match"] + + +# ──────────────────────────────────────────────────────────────── +# 8. Router configuration and state (subprocess) +# ──────────────────────────────────────────────────────────────── + +class TestRouterConfiguration: + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_single_delivery_identity_per_router(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + id_a = RNS.Identity() + id_b = RNS.Identity() + router = LXMF.LXMRouter(identity=id_a, storagepath=_tmpdir+"/router") + + first = router.register_delivery_identity(id_a, display_name="A") + second = router.register_delivery_identity(id_b, display_name="B") + + _emit({ + "first_ok": first is not None, + "second_rejected": second is None, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["first_ok"] + assert data["second_rejected"] + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_stamp_cost_configuration(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + identity = RNS.Identity() + router = LXMF.LXMRouter(identity=identity, storagepath=_tmpdir+"/router") + dest = router.register_delivery_identity(identity, stamp_cost=5) + + initial_cost = router.delivery_destinations[dest.hash].stamp_cost + router.set_inbound_stamp_cost(dest.hash, 10) + updated_cost = router.delivery_destinations[dest.hash].stamp_cost + router.set_inbound_stamp_cost(dest.hash, None) + cleared_cost = router.delivery_destinations[dest.hash].stamp_cost + + _emit({ + "initial_5": initial_cost == 5, + "updated_10": updated_cost == 10, + "cleared_none": cleared_cost is None, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()) + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_enforce_stamps_toggle(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + identity = RNS.Identity() + router = LXMF.LXMRouter( + identity=identity, + storagepath=_tmpdir+"/router", + enforce_stamps=False, + ) + + initial = router._enforce_stamps + router.enforce_stamps() + after_enforce = router._enforce_stamps + router.ignore_stamps() + after_ignore = router._enforce_stamps + + _emit({ + "initial_false": not initial, + "enforced": after_enforce, + "ignored": not after_ignore, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()) + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_outbound_propagation_node_set_and_read(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + identity = RNS.Identity() + router = LXMF.LXMRouter(identity=identity, storagepath=_tmpdir+"/router") + + initial = router.get_outbound_propagation_node() + + node_hash = os.urandom(16) + router.set_outbound_propagation_node(node_hash) + set_hash = router.get_outbound_propagation_node() + + node_hash_2 = os.urandom(16) + router.set_outbound_propagation_node(node_hash_2) + updated_hash = router.get_outbound_propagation_node() + + _emit({ + "initial_none": initial is None, + "set_correctly": set_hash == node_hash, + "update_works": updated_hash == node_hash_2, + "different_nodes": node_hash != node_hash_2, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()), f"Failures: {[k for k, v in data.items() if not v]}" + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_router_max_delivery_attempts_constant(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + _emit({ + "max_attempts": LXMF.LXMRouter.MAX_DELIVERY_ATTEMPTS, + "is_positive": LXMF.LXMRouter.MAX_DELIVERY_ATTEMPTS > 0, + "processing_interval": LXMF.LXMRouter.PROCESSING_INTERVAL, + "message_expiry": LXMF.LXMRouter.MESSAGE_EXPIRY, + "expiry_positive": LXMF.LXMRouter.MESSAGE_EXPIRY > 0, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["is_positive"] + assert data["expiry_positive"] + assert data["max_attempts"] == 5 + + +# ──────────────────────────────────────────────────────────────── +# 9. End-to-end two-router communication (subprocess) +# ──────────────────────────────────────────────────────────────── + +class TestTwoRouterCommunication: + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_bidirectional_protocol_delivery(self): + """Alice sends to Bob, then Bob sends to Alice via lxmf_delivery.""" + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + alice_id = RNS.Identity() + bob_id = RNS.Identity() + + router_a = LXMF.LXMRouter(identity=alice_id, storagepath=_tmpdir+"/alice") + router_b = LXMF.LXMRouter(identity=bob_id, storagepath=_tmpdir+"/bob") + + alice_dest = router_a.register_delivery_identity(alice_id, display_name="Alice") + bob_dest = router_b.register_delivery_identity(bob_id, display_name="Bob") + + alice_inbox = [] + bob_inbox = [] + router_a.register_delivery_callback(lambda msg: alice_inbox.append(msg)) + router_b.register_delivery_callback(lambda msg: bob_inbox.append(msg)) + + bob_dest_out = RNS.Destination(bob_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + alice_dest_out = RNS.Destination(alice_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + msg_to_bob = LXMF.LXMessage(bob_dest_out, alice_dest, content="Hi Bob") + msg_to_bob.pack() + router_b.lxmf_delivery(msg_to_bob.packed) + + msg_to_alice = LXMF.LXMessage(alice_dest_out, bob_dest, content="Hi Alice") + msg_to_alice.pack() + router_a.lxmf_delivery(msg_to_alice.packed) + + _emit({ + "bob_received": len(bob_inbox) == 1, + "bob_content": bob_inbox[0].content_as_string() == "Hi Bob" if bob_inbox else False, + "alice_received": len(alice_inbox) == 1, + "alice_content": alice_inbox[0].content_as_string() == "Hi Alice" if alice_inbox else False, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()), f"Failures: {[k for k, v in data.items() if not v]}" + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_multiple_messages_sequential_delivery(self): + """Send several messages in sequence and verify all arrive in order.""" + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + alice_id = RNS.Identity() + bob_id = RNS.Identity() + + router_a = LXMF.LXMRouter(identity=alice_id, storagepath=_tmpdir+"/alice") + router_b = LXMF.LXMRouter(identity=bob_id, storagepath=_tmpdir+"/bob") + + alice_dest = router_a.register_delivery_identity(alice_id, display_name="Alice") + bob_dest = router_b.register_delivery_identity(bob_id, display_name="Bob") + + inbox = [] + router_b.register_delivery_callback(lambda msg: inbox.append(msg.content_as_string())) + + bob_dest_out = RNS.Destination(bob_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + message_count = 10 + for i in range(message_count): + msg = LXMF.LXMessage(bob_dest_out, alice_dest, content=f"Message {i}") + msg.pack() + router_b.lxmf_delivery(msg.packed, allow_duplicate=True) + + expected = [f"Message {i}" for i in range(message_count)] + _emit({ + "all_received": len(inbox) == message_count, + "order_preserved": inbox == expected, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["all_received"] + assert data["order_preserved"] + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_stamped_bidirectional_with_enforcement(self): + """Both sides enforce stamps; both sides solve and verify.""" + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + alice_id = RNS.Identity() + bob_id = RNS.Identity() + + router_a = LXMF.LXMRouter( + identity=alice_id, storagepath=_tmpdir+"/alice", enforce_stamps=True, + ) + router_b = LXMF.LXMRouter( + identity=bob_id, storagepath=_tmpdir+"/bob", enforce_stamps=True, + ) + + alice_dest = router_a.register_delivery_identity(alice_id, display_name="Alice", stamp_cost=2) + bob_dest = router_b.register_delivery_identity(bob_id, display_name="Bob", stamp_cost=2) + + alice_inbox = [] + bob_inbox = [] + router_a.register_delivery_callback(lambda msg: alice_inbox.append(msg)) + router_b.register_delivery_callback(lambda msg: bob_inbox.append(msg)) + + bob_dest_out = RNS.Destination(bob_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + alice_dest_out = RNS.Destination(alice_id, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + msg_ab = LXMF.LXMessage(bob_dest_out, alice_dest, content="Stamped A->B", stamp_cost=2) + msg_ab.defer_stamp = False + msg_ab.pack() + r1 = router_b.lxmf_delivery(msg_ab.packed) + + msg_ba = LXMF.LXMessage(alice_dest_out, bob_dest, content="Stamped B->A", stamp_cost=2) + msg_ba.defer_stamp = False + msg_ba.pack() + r2 = router_a.lxmf_delivery(msg_ba.packed) + + _emit({ + "ab_accepted": r1 == True, + "ba_accepted": r2 == True, + "bob_stamp_valid": bob_inbox[0].stamp_valid if bob_inbox else False, + "alice_stamp_valid": alice_inbox[0].stamp_valid if alice_inbox else False, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()), f"Failures: {[k for k, v in data.items() if not v]}" + + +# ──────────────────────────────────────────────────────────────── +# 10. Reticulum identity and destination basics (subprocess) +# ──────────────────────────────────────────────────────────────── + +class TestReticulumPrimitives: + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_identity_creation_and_key_operations(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + identity = RNS.Identity() + pub_key = identity.get_public_key() + priv_key = identity.get_private_key() + + test_data = b"Reticulum mesh network" + signature = identity.sign(test_data) + valid = identity.validate(signature, test_data) + tampered = identity.validate(signature, b"tampered data") + + ciphertext = identity.encrypt(test_data) + plaintext = identity.decrypt(ciphertext) + + _emit({ + "has_pub_key": pub_key is not None and len(pub_key) > 0, + "has_priv_key": priv_key is not None and len(priv_key) > 0, + "sig_valid": valid, + "tampered_invalid": not tampered, + "encrypt_decrypt_ok": plaintext == test_data, + "hash_length": len(identity.hash) if identity.hash else 0, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(v for k, v in data.items() if k != "hash_length") + assert data["hash_length"] == RNS.Reticulum.TRUNCATED_HASHLENGTH // 8 + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_destination_hash_determinism(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + identity = RNS.Identity() + + dest1 = RNS.Destination(identity, RNS.Destination.IN, RNS.Destination.SINGLE, "lxmf", "delivery") + dest2 = RNS.Destination(identity, RNS.Destination.OUT, RNS.Destination.SINGLE, "lxmf", "delivery") + + _emit({ + "same_hash": dest1.hash == dest2.hash, + "hash_length": len(dest1.hash), + "hexhash_length": len(dest1.hexhash), + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert data["same_hash"] + assert data["hash_length"] == 16 + assert data["hexhash_length"] == 32 + + @pytest.mark.integration + @pytest.mark.skipif(not _RUN, reason="Set MESHCHAT_LIVE_RETICULUM=1") + def test_identity_persistence(self): + script = _SUBPROCESS_PREAMBLE + textwrap.dedent("""\ + try: + identity = RNS.Identity() + id_path = os.path.join(_tmpdir, "test_identity") + identity.to_file(id_path) + + loaded = RNS.Identity.from_file(id_path) + test_data = b"persistence check" + sig = identity.sign(test_data) + cross_valid = loaded.validate(sig, test_data) + + _emit({ + "file_exists": os.path.exists(id_path), + "loaded_ok": loaded is not None, + "same_pub_key": identity.get_public_key() == loaded.get_public_key(), + "cross_signature_valid": cross_valid, + }) + finally: + _cleanup() + """) + data = _parse_result(_run_lxmf_script(script)) + assert all(data.values()) diff --git a/tests/backend/test_mesh_page_file_path_security.py b/tests/backend/test_mesh_page_file_path_security.py index d1c80fb..5058476 100644 --- a/tests/backend/test_mesh_page_file_path_security.py +++ b/tests/backend/test_mesh_page_file_path_security.py @@ -1,6 +1,4 @@ -"""Path traversal regression tests and property-based fuzzing for mesh PageNode -page/file APIs and normalize_page_filename. -""" +"""Path traversal and fuzz tests for PageNode and ``normalize_page_filename``.""" import os import shutil @@ -177,7 +175,9 @@ class TestPageRespondersTraversal: assert result == b"ok" def test_file_responder_ignores_path_prefix_in_request_path( - self, node_dir, mock_rns, + self, + node_dir, + mock_rns, ): node = _make_node(node_dir, mock_rns) node.setup() diff --git a/tests/backend/test_notification_unread_semantics.py b/tests/backend/test_notification_unread_semantics.py index b224b35..3ca33ea 100644 --- a/tests/backend/test_notification_unread_semantics.py +++ b/tests/backend/test_notification_unread_semantics.py @@ -1,6 +1,4 @@ -"""Regression tests for conversation/bell unread semantics: no false positives from -outgoing-latest threads or inconsistent read cursors. -""" +"""Unread semantics for conversations and the bell (no false positives).""" from datetime import UTC, datetime diff --git a/tests/backend/test_notifications.py b/tests/backend/test_notifications.py index a924c33..7c888e2 100644 --- a/tests/backend/test_notifications.py +++ b/tests/backend/test_notifications.py @@ -213,9 +213,7 @@ def test_notification_spike_fuzzing(db, num_notifs): class TestNotificationReliability: - """Comprehensive tests to verify notifications are accurate, reliable, - and never produce false positives. - """ + """Notification counts and read state stay consistent (no false positives).""" def test_unread_count_matches_actual_unread(self, db): """Unread count must exactly match unviewed notifications.""" diff --git a/tests/backend/test_performance_hotpaths.py b/tests/backend/test_performance_hotpaths.py index 8b63d56..3f84d12 100644 --- a/tests/backend/test_performance_hotpaths.py +++ b/tests/backend/test_performance_hotpaths.py @@ -642,14 +642,8 @@ class TestPerformanceHotPaths(unittest.TestCase): self.assertEqual(len(write_errors), 0, f"Write errors: {write_errors[:5]}") self.assertEqual(len(read_errors), 0, f"Read errors: {read_errors[:5]}") - # =================================================================== - # LIKE SEARCH SCALING — how search degrades with data size - # =================================================================== - def test_like_search_scaling(self): - """Measure how LIKE search scales across different table sizes. - This catches missing FTS indexes or query plan regressions. - """ + """Exercise LIKE search at scale (catches missing indexes / bad plans).""" print("\n[Scaling] LIKE search across data sizes:") # Message search on the existing 10k dataset @@ -699,13 +693,16 @@ class TestPerformanceHotPaths(unittest.TestCase): durations = [] for _ in range(5): _, ms = timed_call( - self.db.messages.mark_all_notifications_as_viewed, hashes, + self.db.messages.mark_all_notifications_as_viewed, + hashes, ) durations.append(ms) stats = latency_report("mark_viewed_200", durations) self.assertLess( - stats["p95"], 50, "mark_all_notifications_as_viewed(200) p95 > 50ms", + stats["p95"], + 50, + "mark_all_notifications_as_viewed(200) p95 > 50ms", ) def test_move_conversations_to_folder_batch(self): @@ -719,13 +716,17 @@ class TestPerformanceHotPaths(unittest.TestCase): durations = [] for _ in range(5): _, ms = timed_call( - self.db.messages.move_conversations_to_folder, hashes, folder_id, + self.db.messages.move_conversations_to_folder, + hashes, + folder_id, ) durations.append(ms) stats = latency_report("move_folder_200", durations) self.assertLess( - stats["p95"], 50, "move_conversations_to_folder(200) p95 > 50ms", + stats["p95"], + 50, + "move_conversations_to_folder(200) p95 > 50ms", ) # =================================================================== diff --git a/tests/backend/test_websocket_scale.py b/tests/backend/test_websocket_scale.py index 6b9840c..4b1c8f8 100644 --- a/tests/backend/test_websocket_scale.py +++ b/tests/backend/test_websocket_scale.py @@ -68,11 +68,10 @@ async def test_websocket_broadcast_soak_iterations(mock_app): @pytest.mark.asyncio async def test_websocket_broadcast_iterates_snapshot_not_live_list(mock_app): - """Mutating ``websocket_clients`` during iteration (e.g. another handler removes - a connection) must not skip entries; iterate a snapshot, not the live list. + """Broadcast must iterate a snapshot of websocket clients, not the live list. - Without ``list(...)``, removing a later client while iterating can skip that client - entirely (classic for-loop over mutating list). + If another coroutine mutates ``websocket_clients`` during iteration, using + ``list(...)`` avoids skipping entries (classic mutating-list pitfall). """ mock_app.websocket_clients.clear() clients = [MagicWs() for _ in range(5)]