mirror of
https://github.com/jkingsman/Remote-Terminal-for-MeshCore.git
synced 2026-03-28 17:43:05 +01:00
249 lines
7.5 KiB
Python
249 lines
7.5 KiB
Python
"""Tests for the centralized path encoding/decoding helpers."""
|
|
|
|
import pytest
|
|
|
|
from app.path_utils import (
|
|
decode_path_byte,
|
|
first_hop_hex,
|
|
parse_packet_envelope,
|
|
path_wire_len,
|
|
split_path_hex,
|
|
validate_path_byte,
|
|
)
|
|
|
|
|
|
class TestDecodePathByte:
|
|
"""Test decoding the packed [hash_mode:2][hop_count:6] byte."""
|
|
|
|
def test_mode0_single_hop(self):
|
|
"""Mode 0 (1-byte hops), 1 hop → path_byte = 0x01."""
|
|
hop_count, hash_size = decode_path_byte(0x01)
|
|
assert hop_count == 1
|
|
assert hash_size == 1
|
|
|
|
def test_mode0_three_hops(self):
|
|
"""Mode 0, 3 hops → path_byte = 0x03."""
|
|
hop_count, hash_size = decode_path_byte(0x03)
|
|
assert hop_count == 3
|
|
assert hash_size == 1
|
|
|
|
def test_mode0_zero_hops(self):
|
|
"""Mode 0, 0 hops (direct) → path_byte = 0x00."""
|
|
hop_count, hash_size = decode_path_byte(0x00)
|
|
assert hop_count == 0
|
|
assert hash_size == 1
|
|
|
|
def test_mode1_two_byte_hops(self):
|
|
"""Mode 1 (2-byte hops), 2 hops → path_byte = 0x42."""
|
|
hop_count, hash_size = decode_path_byte(0x42)
|
|
assert hop_count == 2
|
|
assert hash_size == 2
|
|
|
|
def test_mode1_single_hop(self):
|
|
"""Mode 1 (2-byte hops), 1 hop → path_byte = 0x41."""
|
|
hop_count, hash_size = decode_path_byte(0x41)
|
|
assert hop_count == 1
|
|
assert hash_size == 2
|
|
|
|
def test_mode2_three_byte_hops(self):
|
|
"""Mode 2 (3-byte hops), 1 hop → path_byte = 0x81."""
|
|
hop_count, hash_size = decode_path_byte(0x81)
|
|
assert hop_count == 1
|
|
assert hash_size == 3
|
|
|
|
def test_mode2_max_hops(self):
|
|
"""Mode 2, 63 hops (maximum) → path_byte = 0xBF."""
|
|
hop_count, hash_size = decode_path_byte(0xBF)
|
|
assert hop_count == 63
|
|
assert hash_size == 3
|
|
|
|
def test_mode3_reserved_raises(self):
|
|
"""Mode 3 is reserved and should raise ValueError."""
|
|
with pytest.raises(ValueError, match="Reserved path hash mode 3"):
|
|
decode_path_byte(0xC0)
|
|
|
|
def test_mode3_with_hops_raises(self):
|
|
"""Mode 3 with hop count should also raise."""
|
|
with pytest.raises(ValueError, match="Reserved"):
|
|
decode_path_byte(0xC5)
|
|
|
|
def test_backward_compat_old_firmware(self):
|
|
"""Old firmware packets have upper bits = 0, so mode=0 and path_byte = hop count."""
|
|
for n in range(0, 64):
|
|
hop_count, hash_size = decode_path_byte(n)
|
|
assert hop_count == n
|
|
assert hash_size == 1
|
|
|
|
|
|
class TestPathWireLen:
|
|
def test_basic(self):
|
|
assert path_wire_len(3, 1) == 3
|
|
assert path_wire_len(2, 2) == 4
|
|
assert path_wire_len(1, 3) == 3
|
|
assert path_wire_len(0, 1) == 0
|
|
|
|
|
|
class TestValidatePathByte:
|
|
def test_accepts_valid_multibyte_path_len(self):
|
|
hop_count, hash_size, byte_len = validate_path_byte(0x42)
|
|
assert (hop_count, hash_size, byte_len) == (2, 2, 4)
|
|
|
|
def test_rejects_oversize_path(self):
|
|
with pytest.raises(ValueError, match="MAX_PATH_SIZE"):
|
|
validate_path_byte(0xBF)
|
|
|
|
|
|
class TestParsePacketEnvelope:
|
|
def test_parses_valid_packet(self):
|
|
envelope = parse_packet_envelope(bytes([0x15, 0x42, 0xAA, 0xBB, 0xCC, 0xDD]) + b"hi")
|
|
assert envelope is not None
|
|
assert envelope.hop_count == 2
|
|
assert envelope.hash_size == 2
|
|
assert envelope.path == bytes([0xAA, 0xBB, 0xCC, 0xDD])
|
|
assert envelope.payload == b"hi"
|
|
|
|
def test_rejects_packet_with_no_payload(self):
|
|
assert parse_packet_envelope(bytes([0x15, 0x02, 0xAA, 0xBB])) is None
|
|
|
|
def test_rejects_oversize_path_encoding(self):
|
|
packet = bytes([0x15, 0xBF]) + bytes(189) + b"x"
|
|
assert parse_packet_envelope(packet) is None
|
|
|
|
|
|
class TestSplitPathHex:
|
|
def test_one_byte_hops(self):
|
|
assert split_path_hex("1a2b3c", 3) == ["1a", "2b", "3c"]
|
|
|
|
def test_two_byte_hops(self):
|
|
assert split_path_hex("1a2b3c4d", 2) == ["1a2b", "3c4d"]
|
|
|
|
def test_three_byte_hops(self):
|
|
assert split_path_hex("1a2b3c4d5e6f", 2) == ["1a2b3c", "4d5e6f"]
|
|
|
|
def test_empty_path(self):
|
|
assert split_path_hex("", 0) == []
|
|
assert split_path_hex("", 3) == []
|
|
|
|
def test_zero_hop_count(self):
|
|
assert split_path_hex("1a2b", 0) == []
|
|
|
|
def test_inconsistent_length_falls_back(self):
|
|
"""If hex length doesn't divide evenly by hop_count, fall back to 2-char chunks."""
|
|
assert split_path_hex("1a2b3c", 2) == ["1a", "2b", "3c"]
|
|
|
|
def test_single_hop_one_byte(self):
|
|
assert split_path_hex("ab", 1) == ["ab"]
|
|
|
|
def test_single_hop_two_bytes(self):
|
|
assert split_path_hex("abcd", 1) == ["abcd"]
|
|
|
|
|
|
class TestFirstHopHex:
|
|
def test_one_byte_hops(self):
|
|
assert first_hop_hex("1a2b3c", 3) == "1a"
|
|
|
|
def test_two_byte_hops(self):
|
|
assert first_hop_hex("1a2b3c4d", 2) == "1a2b"
|
|
|
|
def test_empty(self):
|
|
assert first_hop_hex("", 0) is None
|
|
assert first_hop_hex("", 1) is None
|
|
|
|
def test_direct_path(self):
|
|
assert first_hop_hex("", 0) is None
|
|
|
|
|
|
class TestContactToRadioDictHashMode:
|
|
"""Test that Contact.to_radio_dict() preserves the stored out_path_hash_mode."""
|
|
|
|
def test_preserves_1byte_mode(self):
|
|
from app.models import Contact
|
|
|
|
c = Contact(
|
|
public_key="aa" * 32,
|
|
last_path="1a2b3c",
|
|
last_path_len=3,
|
|
out_path_hash_mode=0,
|
|
)
|
|
d = c.to_radio_dict()
|
|
assert d["out_path_hash_mode"] == 0
|
|
|
|
def test_preserves_2byte_mode(self):
|
|
from app.models import Contact
|
|
|
|
c = Contact(
|
|
public_key="bb" * 32,
|
|
last_path="1a2b3c4d",
|
|
last_path_len=2,
|
|
out_path_hash_mode=1,
|
|
)
|
|
d = c.to_radio_dict()
|
|
assert d["out_path_hash_mode"] == 1
|
|
|
|
def test_preserves_3byte_mode(self):
|
|
from app.models import Contact
|
|
|
|
c = Contact(
|
|
public_key="cc" * 32,
|
|
last_path="1a2b3c4d5e6f",
|
|
last_path_len=2,
|
|
out_path_hash_mode=2,
|
|
)
|
|
d = c.to_radio_dict()
|
|
assert d["out_path_hash_mode"] == 2
|
|
|
|
def test_preserves_flood_mode(self):
|
|
from app.models import Contact
|
|
|
|
c = Contact(
|
|
public_key="dd" * 32,
|
|
last_path=None,
|
|
last_path_len=-1,
|
|
out_path_hash_mode=-1,
|
|
)
|
|
d = c.to_radio_dict()
|
|
assert d["out_path_hash_mode"] == -1
|
|
|
|
def test_preserves_mode_with_zero_bytes_in_path(self):
|
|
from app.models import Contact
|
|
|
|
c = Contact(
|
|
public_key="ee" * 32,
|
|
last_path="aa00bb00",
|
|
last_path_len=2,
|
|
out_path_hash_mode=1,
|
|
)
|
|
d = c.to_radio_dict()
|
|
assert d["out_path_hash_mode"] == 1
|
|
|
|
|
|
class TestContactFromRadioDictHashMode:
|
|
"""Test that Contact.from_radio_dict() preserves explicit path hash mode."""
|
|
|
|
def test_preserves_mode_from_radio_payload(self):
|
|
from app.models import Contact
|
|
|
|
d = Contact.from_radio_dict(
|
|
"aa" * 32,
|
|
{
|
|
"adv_name": "Alice",
|
|
"out_path": "aa00bb00",
|
|
"out_path_len": 2,
|
|
"out_path_hash_mode": 1,
|
|
},
|
|
)
|
|
assert d["out_path_hash_mode"] == 1
|
|
|
|
def test_flood_falls_back_to_minus_one(self):
|
|
from app.models import Contact
|
|
|
|
d = Contact.from_radio_dict(
|
|
"bb" * 32,
|
|
{
|
|
"adv_name": "Bob",
|
|
"out_path": "",
|
|
"out_path_len": -1,
|
|
},
|
|
)
|
|
assert d["out_path_hash_mode"] == -1
|