Phase 8: Tests & Misc

tests/test_community_mqtt.py

@@ -18,6 +18,7 @@ from app.fanout.community_mqtt import (
     _build_radio_info,
     _build_status_topic,
     _calculate_packet_hash,
+    _decode_packet_fields,
     _ed25519_sign_expanded,
     _format_raw_packet,
     _generate_jwt_token,
@@ -376,6 +377,156 @@ class TestCalculatePacketHash:
         raw = bytes([0x10, 0x01, 0x02])
         assert _calculate_packet_hash(raw) == "0" * 16
 
+    def test_multibyte_2byte_hops_skips_correct_path_length(self):
+        """Mode 1 (2-byte hops), 2 hops → 4 bytes of path data to skip."""
+        import hashlib
+
+        # FLOOD route, payload_type=2 (TXT_MSG): (2<<2)|1 = 0x09
+        # path_byte = 0x42 → mode=1, hop_count=2 → path_wire_len = 2*2 = 4
+        path_data = b"\xaa\xbb\xcc\xdd"
+        payload = b"\x48\x65\x6c\x6c\x6f"  # "Hello"
+        raw = bytes([0x09, 0x42]) + path_data + payload
+        result = _calculate_packet_hash(raw)
+
+        expected = hashlib.sha256(bytes([2]) + payload).hexdigest()[:16].upper()
+        assert result == expected
+
+    def test_multibyte_3byte_hops_skips_correct_path_length(self):
+        """Mode 2 (3-byte hops), 1 hop → 3 bytes of path data to skip."""
+        import hashlib
+
+        # FLOOD route, payload_type=4 (ADVERT): (4<<2)|1 = 0x11
+        # path_byte = 0x81 → mode=2, hop_count=1 → path_wire_len = 1*3 = 3
+        path_data = b"\xaa\xbb\xcc"
+        payload = b"\xde\xad"
+        raw = bytes([0x11, 0x81]) + path_data + payload
+        result = _calculate_packet_hash(raw)
+
+        expected = hashlib.sha256(bytes([4]) + payload).hexdigest()[:16].upper()
+        assert result == expected
+
+    def test_trace_multibyte_uses_raw_wire_byte_in_hash(self):
+        """TRACE with multi-byte hops uses raw path_byte (not decoded hop count) in hash."""
+        import hashlib
+
+        # TRACE with FLOOD route: (9<<2)|1 = 0x25
+        # path_byte = 0x42 → mode=1, hop_count=2 → path_wire_len = 4
+        path_byte = 0x42
+        path_data = b"\xaa\xbb\xcc\xdd"
+        payload = b"\x01\x02"
+        raw = bytes([0x25, path_byte]) + path_data + payload
+        result = _calculate_packet_hash(raw)
+
+        # TRACE hash includes raw path_byte as uint16_t LE (0x42, 0x00)
+        expected = (
+            hashlib.sha256(bytes([9]) + path_byte.to_bytes(2, byteorder="little") + payload)
+            .hexdigest()[:16]
+            .upper()
+        )
+        assert result == expected
+
+    def test_multibyte_truncated_path_returns_zeroes(self):
+        """Packet claiming multi-byte path but truncated before path ends → zero hash."""
+        # FLOOD route, payload_type=2: (2<<2)|1 = 0x09
+        # path_byte = 0x42 → mode=1, hop_count=2 → needs 4 bytes of path, only 2 present
+        raw = bytes([0x09, 0x42, 0xAA, 0xBB])
+        assert _calculate_packet_hash(raw) == "0" * 16
+
+    def test_multibyte_transport_flood_with_2byte_hops(self):
+        """TRANSPORT_FLOOD with 2-byte hops correctly skips transport codes + path."""
+        import hashlib
+
+        # TRANSPORT_FLOOD (0x00), payload_type=4: (4<<2)|0 = 0x10
+        transport_codes = b"\x01\x02\x03\x04"
+        # path_byte = 0x41 → mode=1, hop_count=1 → path_wire_len = 2
+        path_data = b"\xaa\xbb"
+        payload = b"\xca\xfe"
+        raw = bytes([0x10]) + transport_codes + bytes([0x41]) + path_data + payload
+        result = _calculate_packet_hash(raw)
+
+        expected = hashlib.sha256(bytes([4]) + payload).hexdigest()[:16].upper()
+        assert result == expected
+
+
+class TestDecodePacketFieldsMultibyte:
+    """Test _decode_packet_fields with multi-byte hop paths."""
+
+    def test_1byte_hops_legacy(self):
+        """Legacy packet with mode=0, 3 hops → 3 path bytes."""
+        # FLOOD route, payload_type=2 (TXT_MSG): (2<<2)|1 = 0x09
+        path_data = b"\xaa\xbb\xcc"
+        payload = b"\x48\x65\x6c\x6c\x6f"
+        raw = bytes([0x09, 0x03]) + path_data + payload
+
+        route, ptype, plen, path_values, payload_type = _decode_packet_fields(raw)
+        assert route == "F"
+        assert ptype == "2"
+        assert path_values == ["aa", "bb", "cc"]
+        assert payload_type == 2
+
+    def test_2byte_hops(self):
+        """Mode 1, 2 hops → 4 path bytes split into 2 two-byte identifiers."""
+        # FLOOD route, payload_type=2: (2<<2)|1 = 0x09
+        # path_byte = 0x42 → mode=1, hop_count=2
+        path_data = b"\xaa\xbb\xcc\xdd"
+        payload = b"\x48\x65\x6c\x6c\x6f"
+        raw = bytes([0x09, 0x42]) + path_data + payload
+
+        route, ptype, plen, path_values, payload_type = _decode_packet_fields(raw)
+        assert route == "F"
+        assert ptype == "2"
+        assert path_values == ["aabb", "ccdd"]
+        assert payload_type == 2
+        assert plen == str(len(payload))
+
+    def test_3byte_hops(self):
+        """Mode 2, 1 hop → 3 path bytes as a single three-byte identifier."""
+        # FLOOD route, payload_type=4 (ADVERT): (4<<2)|1 = 0x11
+        # path_byte = 0x81 → mode=2, hop_count=1
+        path_data = b"\xaa\xbb\xcc"
+        payload = b"\xde\xad"
+        raw = bytes([0x11, 0x81]) + path_data + payload
+
+        route, ptype, plen, path_values, payload_type = _decode_packet_fields(raw)
+        assert route == "F"
+        assert ptype == "4"
+        assert path_values == ["aabbcc"]
+        assert payload_type == 4
+
+    def test_direct_packet_no_path(self):
+        """Direct packet (0 hops) → empty path list."""
+        # FLOOD route, payload_type=2: (2<<2)|1 = 0x09
+        # path_byte = 0x00 → mode=0, hop_count=0
+        payload = b"\x48\x65\x6c\x6c\x6f"
+        raw = bytes([0x09, 0x00]) + payload
+
+        route, ptype, plen, path_values, payload_type = _decode_packet_fields(raw)
+        assert path_values == []
+        assert plen == str(len(payload))
+
+    def test_transport_flood_2byte_hops(self):
+        """TRANSPORT_FLOOD with 2-byte hops: skip 4 transport bytes, then decode path."""
+        # TRANSPORT_FLOOD (0x00), payload_type=2: (2<<2)|0 = 0x08
+        transport_codes = b"\x01\x02\x03\x04"
+        # path_byte = 0x42 → mode=1, hop_count=2
+        path_data = b"\xaa\xbb\xcc\xdd"
+        payload = b"\xca\xfe"
+        raw = bytes([0x08]) + transport_codes + bytes([0x42]) + path_data + payload
+
+        route, ptype, plen, path_values, payload_type = _decode_packet_fields(raw)
+        assert route == "F"
+        assert path_values == ["aabb", "ccdd"]
+
+    def test_truncated_multibyte_path_returns_defaults(self):
+        """Packet claiming 2-byte hops but truncated → defaults."""
+        # FLOOD route, payload_type=2: (2<<2)|1 = 0x09
+        # path_byte = 0x42 → mode=1, hop_count=2 → needs 4 bytes, only 1
+        raw = bytes([0x09, 0x42, 0xAA])
+
+        route, ptype, plen, path_values, payload_type = _decode_packet_fields(raw)
+        assert path_values == []
+        assert plen == "0"
+
 
 class TestCommunityMqttPublisher:
     def test_initial_state(self):

tests/test_fanout.py

@@ -661,6 +661,75 @@ class TestAppriseFormatBody:
         )
         assert "`direct`" in body
 
+    def test_dm_with_2byte_hop_path(self):
+        """Multi-byte (2-byte) hops are correctly split using path_len metadata."""
+        from app.fanout.apprise_mod import _format_body
+
+        body = _format_body(
+            {
+                "type": "PRIV",
+                "text": "hi",
+                "sender_name": "Alice",
+                "paths": [{"path": "aabbccdd", "path_len": 2}],
+            },
+            include_path=True,
+        )
+        assert "**via:**" in body
+        assert "`aabb`" in body
+        assert "`ccdd`" in body
+
+    def test_dm_with_3byte_hop_path(self):
+        """Multi-byte (3-byte) hops are correctly split using path_len metadata."""
+        from app.fanout.apprise_mod import _format_body
+
+        body = _format_body(
+            {
+                "type": "PRIV",
+                "text": "hi",
+                "sender_name": "Alice",
+                "paths": [{"path": "aabbccddeeff", "path_len": 2}],
+            },
+            include_path=True,
+        )
+        assert "**via:**" in body
+        assert "`aabbcc`" in body
+        assert "`ddeeff`" in body
+
+    def test_channel_with_multibyte_path(self):
+        """Channel message with 2-byte hop path_len metadata."""
+        from app.fanout.apprise_mod import _format_body
+
+        body = _format_body(
+            {
+                "type": "CHAN",
+                "text": "hi",
+                "sender_name": "Bob",
+                "channel_name": "#general",
+                "paths": [{"path": "aabbccdd", "path_len": 2}],
+            },
+            include_path=True,
+        )
+        assert "**#general:**" in body
+        assert "`aabb`" in body
+        assert "`ccdd`" in body
+
+    def test_legacy_path_without_path_len(self):
+        """Legacy path (no path_len) falls back to 2-char chunks."""
+        from app.fanout.apprise_mod import _format_body
+
+        body = _format_body(
+            {
+                "type": "PRIV",
+                "text": "hi",
+                "sender_name": "Alice",
+                "paths": [{"path": "aabb"}],
+            },
+            include_path=True,
+        )
+        assert "**via:**" in body
+        assert "`aa`" in body
+        assert "`bb`" in body
+
 
 class TestAppriseNormalizeDiscordUrl:
     def test_discord_scheme(self):

tests/test_path_utils.py

@@ -143,3 +143,42 @@ class TestInferHashSize:
 
     def test_zero_hop_count_defaults_to_1(self):
         assert infer_hash_size("1a2b", 0) == 1
+
+
+class TestContactToRadioDictHashMode:
+    """Test that Contact.to_radio_dict() correctly derives out_path_hash_mode."""
+
+    def test_1byte_hops(self):
+        from app.models import Contact
+
+        c = Contact(public_key="aa" * 32, last_path="1a2b3c", last_path_len=3)
+        d = c.to_radio_dict()
+        assert d["out_path_hash_mode"] == 0  # infer_hash_size=1, mode=0
+
+    def test_2byte_hops(self):
+        from app.models import Contact
+
+        c = Contact(public_key="bb" * 32, last_path="1a2b3c4d", last_path_len=2)
+        d = c.to_radio_dict()
+        assert d["out_path_hash_mode"] == 1  # infer_hash_size=2, mode=1
+
+    def test_3byte_hops(self):
+        from app.models import Contact
+
+        c = Contact(public_key="cc" * 32, last_path="1a2b3c4d5e6f", last_path_len=2)
+        d = c.to_radio_dict()
+        assert d["out_path_hash_mode"] == 2  # infer_hash_size=3, mode=2
+
+    def test_no_path_defaults_to_mode0(self):
+        from app.models import Contact
+
+        c = Contact(public_key="dd" * 32, last_path=None, last_path_len=-1)
+        d = c.to_radio_dict()
+        assert d["out_path_hash_mode"] == 0
+
+    def test_empty_path_defaults_to_mode0(self):
+        from app.models import Contact
+
+        c = Contact(public_key="ee" * 32, last_path="", last_path_len=0)
+        d = c.to_radio_dict()
+        assert d["out_path_hash_mode"] == 0