Add isolated Ethernet and SSH throughput test harnesses

This commit is contained in:
mikecarper committed 2026-10-15 11:05:46 -07:00
1 parent 2c4a51c9ba
commit ecbca6a84c
35 files changed
+7211

No files matched your search

+140
View File
@@ -0,0 +1,140 @@
"""Actual HIL constructors must not pulse CH340 reset on POSIX serial open."""
import contextlib
import importlib.util
import io
import os
from pathlib import Path
import sys
from types import ModuleType, SimpleNamespace
import unittest
from unittest.mock import Mock, patch
BASE = Path(__file__).resolve().parents[1] / "tools/hil"
sys.path.insert(0, str(BASE))
import ethernet_kiss_probe as probe
import ethernet_tap_relay as tap
class FakeSerial:
def __init__(self, **kwargs):
self.dsrdtr = False
self._dtr = True
self._rts = True
self.driver_rts = True
self.driver_dtr = True
self.is_open = False
self.resets = 0
self.writes = []
self.trace = []
@property
def dtr(self):
return self._dtr
@dtr.setter
def dtr(self, value):
self._dtr = value
if self.is_open:
self._apply_dtr(value)
@property
def rts(self):
return self._rts
@rts.setter
def rts(self, value):
self._rts = value
if self.is_open:
self.driver_rts = value
self.trace.append(("rts", value))
def _apply_dtr(self, value):
if not value and self.driver_rts:
self.resets += 1
self.driver_dtr = value
self.trace.append(("dtr", value))
def open(self):
self.is_open = True
# POSIX pySerial's real open order: DTR before RTS unless deferred.
if not self.dsrdtr:
self._apply_dtr(self._dtr)
self.rts = self._rts
def close(self):
self.is_open = False
def write(self, data):
self.writes.append(data)
return len(data)
def flush(self):
pass
def fileno(self):
return 77
def serial_modules(port):
serial = ModuleType("serial")
serial.Serial = Mock(return_value=port)
tools = ModuleType("serial.tools")
list_ports = ModuleType("serial.tools.list_ports")
info = SimpleNamespace(device="/dev/ttyUSB-bench", serial_number="bench-only",
vid=0x1A86, pid=0x7523)
list_ports.comports = Mock(return_value=[info])
tools.list_ports = list_ports
serial.tools = tools
return {"serial": serial, "serial.tools": tools, "serial.tools.list_ports": list_ports}
@unittest.skipUnless(os.name == "posix", "POSIX CH340 pin-order behavior")
class EthernetHilSerialTests(unittest.TestCase):
def assert_safe(self, port):
self.assertEqual(port.resets, 0)
self.assertEqual(port.trace[0], ("rts", False))
self.assertFalse(port.dsrdtr)
self.assertFalse(port.driver_rts)
self.assertFalse(port.driver_dtr)
def test_reset_simulator_detects_original_plain_open_bug(self):
port = FakeSerial()
port.dtr = port.rts = False
port.open()
self.assertEqual(port.resets, 1)
def test_raw_kiss_probe_uses_shared_safe_opener(self):
port = FakeSerial()
with patch.dict(sys.modules, serial_modules(port)), patch.object(probe.threading, "Thread"):
endpoint = probe.KissEndpoint("serial:///dev/ttyUSB-bench")
endpoint.close()
self.assert_safe(port)
def test_tap_serial_transport_uses_shared_safe_opener(self):
port = FakeSerial()
with patch.dict(sys.modules, serial_modules(port)):
endpoint = tap.KissStream("serial:///dev/ttyUSB-bench", 115200)
endpoint.close()
self.assert_safe(port)
def test_companion_cli_uses_shared_safe_opener(self):
port = FakeSerial()
spec = importlib.util.spec_from_file_location("ethernet_bench_serial_under_test", BASE / "ethernet_bench_serial.py")
module = importlib.util.module_from_spec(spec)
with patch.dict(sys.modules, serial_modules(port)):
spec.loader.exec_module(module)
with patch.object(sys, "argv", ["ethernet_bench_serial.py", "--port", "/dev/ttyUSB-bench",
"--command", "get ethernet"]), \
patch.object(module, "read_quiet", return_value="OK"), \
patch.object(module, "read_command_reply", return_value="get ethernet\r\n -> on\r\n> "), \
contextlib.redirect_stdout(io.StringIO()):
module.main()
self.assert_safe(port)
self.assertIn(b"get ethernet\n", port.writes)
self.assertFalse(port.is_open)
if __name__ == "__main__":
unittest.main()
+289
View File
@@ -0,0 +1,289 @@
"""Host-only full KISS configuration tests; never open a serial device."""
from collections import deque
from pathlib import Path
import struct
import sys
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools/hil"))
import ethernet_kiss_modem_probe as probe
def fake_serial_modules():
serial = SimpleNamespace(Serial=Mock())
list_ports = SimpleNamespace(comports=Mock())
modules = {"serial": serial, "serial.tools": SimpleNamespace(list_ports=list_ports)}
return serial, list_ports, modules
def response(code, body=b""):
return probe.encode_frame(6, bytes((code,)) + body)
class Clock:
def __init__(self):
self.now = 0
def __call__(self):
self.now += 0.01
return self.now
class FakePort:
def __init__(self):
self.decoder = probe.Decoder()
self.input = deque()
self.writes = []
self.override = {}
self.tx = 0
def write(self, data):
for frame in self.decoder.feed(data):
self.writes.append(frame)
if frame[0] != 6:
continue
command = frame[1]
body = {0x21: struct.pack("<I", probe.GUARD_HZ), 0x11: b"\2\0",
0x01: bytes(range(32)), 0x17: b"",
0x12: struct.pack("<III", 10, self.tx, 0),
0x0B: probe.PRIMARY_STRUCT.pack(*probe.PRIMARY),
0x20: probe.SECONDARY_STRUCT.pack(*probe.SECONDARY), 0x0C: b"\0"}
result = self.override.get(command, response(command | 0x80, body[command])
if command in body else response(0xF0))
self.input.extend(result[i:i + 1] for i in range(len(result)))
return len(data)
def flush(self):
pass
def read(self, size):
return self.input.popleft() if self.input else b""
class ProbeTest(unittest.TestCase):
def link(self, port=None):
return probe.ControlLink(port or FakePort(), timeout=1, clock=Clock())
def test_full_configuration_exact_tuple_no_data_or_crypto(self):
port = FakePort()
result = probe.configure_modem(self.link(port))
self.assertTrue(result["success"])
self.assertEqual(result["radio2"]["preamble"], 512)
self.assertEqual(result["radio2"]["protocol"], 2)
self.assertEqual(port.writes[0], b"\6\x21")
self.assertIn(b"\6\x1f" + probe.SECONDARY_STRUCT.pack(*probe.SECONDARY), port.writes)
self.assertIn(b"\1\0", port.writes)
self.assertIn(b"\2\xff", port.writes)
self.assertTrue(all(frame[0] in (1, 2, 6) for frame in port.writes))
self.assertEqual(result["stats_before"], result["stats_after"])
self.assertNotIn(bytes(range(32)).hex(), str(result))
def test_bandwidth125_preamble128_exact_setter_and_readback_without_global_changes(self):
port = FakePort()
expected = (probe.GUARD_HZ, 125000, 5, 5, 2, 128, 2)
port.override[0x20] = response(0xA0, probe.SECONDARY_STRUCT.pack(*expected))
report = probe.configure_modem(self.link(port), bandwidth=125, preamble=128)
self.assertTrue(report["success"])
self.assertEqual(report["radio2"], dict(zip(("frequency_hz", "bandwidth_hz", "sf", "cr",
"mode", "preamble", "protocol"), expected)))
self.assertIn(b"\6\x1f" + probe.SECONDARY_STRUCT.pack(*expected), port.writes)
self.assertEqual(probe.SECONDARY, (probe.GUARD_HZ, 500000, 5, 5, 2, 512, 2))
self.assertEqual(probe.secondary_tuple(), probe.SECONDARY)
self.assertTrue(all(frame[0] in (1, 2, 6) for frame in port.writes))
def test_selected_secondary_readback_cannot_silently_use_default_width_or_preamble(self):
for observed in (probe.SECONDARY, (probe.GUARD_HZ, 125000, 5, 5, 2, 512, 2),
(probe.GUARD_HZ, 500000, 5, 5, 2, 128, 2)):
port = FakePort()
port.override[0x20] = response(0xA0, probe.SECONDARY_STRUCT.pack(*observed))
with self.assertRaisesRegex(probe.ProtocolError, "Secondary"):
probe.configure_modem(self.link(port), bandwidth=125, preamble=128)
def test_invalid_bandwidth_or_preamble_fails_before_any_hardware_request(self):
for options in ([dict(bandwidth=value) for value in (250, 125.0, True, "125", None)] +
[dict(preamble=value) for value in (31, 65, 1025, 128.0, True, "128", None)]):
port = FakePort()
with self.assertRaises(ValueError): probe.configure_modem(self.link(port), **options)
self.assertEqual(port.writes, [])
for preamble in probe.PREAMBLES:
self.assertEqual(probe.secondary_tuple(bandwidth=125, preamble=preamble),
(probe.GUARD_HZ, 125000, 5, 5, 2, preamble, 2))
def test_cli_passes_explicit_width_and_preamble_and_rejects_invalid_before_open(self):
from contextlib import redirect_stdout, redirect_stderr
from io import StringIO
port = FakePort(); port.close = Mock()
info = SimpleNamespace(serial_number="known", device="/dev/example", vid=0x239A, pid=0x8029)
with patch.object(probe, "open_verified_port", return_value=(port, info)), \
patch.object(probe, "configure_modem", return_value={"success": True}) as configure, \
redirect_stdout(StringIO()):
result = probe.main(["--port", "/dev/example", "--expect-serial", "known",
"--bandwidth", "125", "--preamble", "128"])
self.assertEqual(result, 0)
self.assertEqual(configure.call_args.kwargs, dict(bandwidth=125, preamble=128))
port.close.assert_called_once()
for option, value in (("--bandwidth", "250"), ("--preamble", "65")):
with patch.object(probe, "open_verified_port") as opened, redirect_stderr(StringIO()), \
self.assertRaises(SystemExit):
probe.main(["--port", "/dev/example", "--expect-serial", "known", option, value])
opened.assert_not_called()
def test_guard_unknown_fails_before_any_setter(self):
port = FakePort()
port.override[0x21] = response(0xF1, b"\5")
with self.assertRaisesRegex(probe.ProtocolError, "failed: 0x05"):
probe.configure_modem(self.link(port))
self.assertEqual(port.writes, [b"\6\x21"])
def test_wrong_guard_never_sets_anything(self):
port = FakePort()
port.override[0x21] = response(0xA1, struct.pack("<I", 910_525_000))
with self.assertRaisesRegex(probe.ProtocolError, "TX lock"):
probe.configure_modem(self.link(port))
self.assertEqual(len(port.writes), 1)
def test_bad_version_fails_before_setters(self):
port = FakePort()
port.override[0x11] = response(0x91, b"\1\0")
with self.assertRaisesRegex(probe.ProtocolError, "version"):
probe.configure_modem(self.link(port))
self.assertEqual(len(port.writes), 2)
def test_primary_float_rounding_only_frequency(self):
port = FakePort()
port.override[0x0B] = response(0x8B, probe.PRIMARY_STRUCT.pack(910_525_024, 62500, 7, 5))
self.assertEqual(probe.configure_modem(self.link(port))["primary_rx_only"]["frequency_hz"],
910_525_024)
for bad in ((910_526_000, 62500, 7, 5), (910_525_000, 125000, 7, 5)):
port = FakePort()
port.override[0x0B] = response(0x8B, probe.PRIMARY_STRUCT.pack(*bad))
with self.assertRaisesRegex(probe.ProtocolError, "Primary"):
probe.configure_modem(self.link(port))
def test_secondary_legacy_truncated_wrong_protocol_or_preamble_rejected(self):
for body in (probe.SECONDARY_STRUCT.pack(*probe.SECONDARY)[:-1],
probe.SECONDARY_STRUCT.pack(*probe.SECONDARY[:-1], 0),
probe.SECONDARY_STRUCT.pack(*probe.SECONDARY[:5], 32, 2)):
port = FakePort()
port.override[0x20] = response(0xA0, body)
with self.assertRaises(probe.ProtocolError):
probe.configure_modem(self.link(port))
def test_unknown_or_malformed_error_is_immediate(self):
for frame in (response(0xF1, b"\5"), response(0xF1), response(0xF1, b"\5\0")):
port = FakePort()
port.override[0x21] = frame
with self.assertRaises(probe.ProtocolError):
self.link(port).request(0x21, size=4)
def test_timeout_bounded_and_no_retries(self):
port = FakePort()
port.override[0x21] = b""
with self.assertRaisesRegex(probe.ProtocolError, "Timed out"):
self.link(port).request(0x21, size=4)
self.assertEqual(len(port.writes), 1)
def test_metadata_and_rx_data_allowed_but_tx_done_fails(self):
port = FakePort()
port.override[0x21] = (response(0xF9, b"\0\0") + b"\xc0\x10rx\xc0" +
response(0xA1, struct.pack("<I", probe.GUARD_HZ)))
link = self.link(port)
self.assertEqual(len(link.request(0x21, size=4)), 4)
self.assertEqual((link.rx_metadata_frames, link.ignored_data_frames), (1, 1))
port = FakePort()
port.override[0x21] = response(0xF8, b"\1")
with self.assertRaisesRegex(probe.ProtocolError, "TX_DONE"):
self.link(port).request(0x21, size=4)
def test_escape_split_and_overflow(self):
decoder = probe.Decoder()
wire = response(0x81, b"\xc0\xdb")
frames = []
for value in wire:
frames.extend(decoder.feed(bytes((value,))))
self.assertEqual(frames, [b"\6\x81\xc0\xdb"])
for wire in (b"\xc0\6\xdb\0", b"\xc0\6\xdb\xc0",
b"\xc0" + b"x" * 515):
with self.assertRaises(probe.ProtocolError):
probe.Decoder().feed(wire)
def test_encoder_refuses_data_and_nonzero_port(self):
for command in (0, 0x10, -1):
with self.assertRaises(ValueError):
probe.encode_frame(command, b"test")
def test_tx_counter_change_is_not_reported_as_success(self):
with patch.object(probe, "_stats", side_effect=[{"rx": 1, "tx": 0, "errors": 0},
{"rx": 2, "tx": 1, "errors": 0}]):
with self.assertRaisesRegex(probe.ProtocolError, "TX counter changed"):
probe.configure_modem(self.link())
def test_busy_or_wrong_usb_identity_is_never_opened(self):
serial, list_ports, modules = fake_serial_modules()
info = SimpleNamespace(device="/dev/example", serial_number="wrong", vid=0x239A)
with patch.dict(sys.modules, modules), \
patch.object(list_ports, "comports", return_value=[info]), \
patch.object(serial, "Serial") as constructor:
with self.assertRaisesRegex(probe.ProtocolError, "USB serial"):
probe.open_verified_port("/dev/example", "9AB3B64C641BA927")
constructor.assert_not_called()
info.serial_number = "9AB3B64C641BA927"
with patch.dict(sys.modules, modules), \
patch.object(list_ports, "comports", return_value=[info]), \
patch.object(probe, "assert_port_idle", side_effect=RuntimeError("busy")), \
patch.object(serial, "Serial") as constructor:
with self.assertRaisesRegex(RuntimeError, "busy"):
probe.open_verified_port("/dev/example", info.serial_number)
constructor.assert_not_called()
def test_native_nrf_usb_open_is_exclusive_dtr_on_rts_off(self):
serial, list_ports, modules = fake_serial_modules()
info = SimpleNamespace(device="/dev/example", serial_number="9AB3B64C641BA927", vid=0x239A)
port = SimpleNamespace(close=Mock())
with patch.dict(sys.modules, modules), \
patch.object(list_ports, "comports", return_value=[info]), \
patch.object(probe, "assert_port_idle"), \
patch.object(serial, "Serial", return_value=port) as constructor, \
patch.object(probe, "open_configured_session") as opener:
selected, observed = probe.open_verified_port("/dev/example", info.serial_number)
self.assertIs(selected, port)
self.assertIs(observed, info)
self.assertTrue(port.dtr)
self.assertFalse(port.rts)
self.assertTrue(constructor.call_args.kwargs["exclusive"])
self.assertEqual(constructor.call_args.kwargs["baudrate"], 115200)
opener.assert_called_once_with(port)
def test_post_open_usb_identity_change_closes_before_commands(self):
serial, list_ports, modules = fake_serial_modules()
info = SimpleNamespace(device="/dev/example", serial_number="9AB3B64C641BA927", vid=0x239A)
wrong = SimpleNamespace(device="/dev/example", serial_number="different", vid=0x239A)
port = SimpleNamespace(close=Mock())
with patch.dict(sys.modules, modules), \
patch.object(list_ports, "comports", side_effect=[[info], [wrong]]), \
patch.object(probe, "assert_port_idle"), \
patch.object(serial, "Serial", return_value=port), \
patch.object(probe, "open_configured_session"):
with self.assertRaisesRegex(probe.ProtocolError, "identity changed"):
probe.open_verified_port("/dev/example", info.serial_number)
port.close.assert_called_once()
def test_cli_releases_port_on_protocol_failure(self):
from contextlib import redirect_stdout
from io import StringIO
port = FakePort()
port.close = Mock()
info = SimpleNamespace(serial_number="9AB3B64C641BA927", device="/dev/example", vid=0x239A, pid=0x8029)
with patch.object(probe, "open_verified_port", return_value=(port, info)), \
patch.object(probe, "configure_modem", side_effect=probe.ProtocolError("rejected")), \
redirect_stdout(StringIO()):
result = probe.main(["--port", "/dev/example", "--expect-serial", info.serial_number])
self.assertEqual(result, 1)
port.close.assert_called_once()
if __name__ == "__main__":
unittest.main()
+126
View File
@@ -0,0 +1,126 @@
"""Host-only regression tests; no radio, network connection or TAP is opened."""
import contextlib
import importlib.util
import io
from pathlib import Path
import queue
import struct
import sys
import time
import unittest
from unittest.mock import patch
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "tools/hil"))
SPEC = importlib.util.spec_from_file_location("ethernet_kiss_probe", ROOT / "tools/hil/ethernet_kiss_probe.py")
probe = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(probe)
class FakeEndpoint:
def __init__(self):
self.incoming = queue.Queue()
self.peer = None
self.drop = False
self.frames = []
def send(self, frame, fragment=0):
self.frames.append(frame)
if not self.drop:
self.peer.incoming.put((frame, time.monotonic()))
def receive(self, timeout):
try:
return self.incoming.get(timeout=timeout)
except queue.Empty:
return None
class EthernetKissProbeTests(unittest.TestCase):
def test_tcp_endpoint_defaults_to_dedicated_kiss_port(self):
with patch.object(probe.socket, "create_connection") as connect, \
patch.object(probe.threading, "Thread"):
endpoint = probe.KissEndpoint("tcp://bench-radio")
connect.assert_called_once_with(("bench-radio", 5003), timeout=5)
endpoint.close()
def test_streaming_kiss_preserves_all_bytes(self):
frame = bytes(range(256))[:255]
encoded = probe.kiss_encode(frame)
decoder = probe.KissDecoder()
frames = []
for byte in encoded:
frames.extend(decoder.feed(bytes((byte,))))
self.assertEqual(frames, [frame])
self.assertEqual(decoder.errors, 0)
def test_decoder_ignores_controls_and_rejects_invalid_escapes(self):
decoder = probe.KissDecoder()
encoded = b"noise\xc0\x01control\xc0\x00\xdbbad\xc0"
self.assertEqual(decoder.feed(encoded), [])
self.assertEqual(decoder.errors, 1)
self.assertEqual(decoder.feed(probe.kiss_encode(b"valid")), [b"valid"])
self.assertEqual(decoder.feed(b"\xc0\x00" + b"X" * 300 + b"\xc0"), [])
self.assertEqual(decoder.errors, 2)
def test_ipv4_and_transport_checksums_are_valid_both_directions(self):
for kind in ("udp", "tcp", "icmp"):
for reverse in (False, True):
frame = probe.make_frame(kind, 255, 0x12345678, reverse)
self.assertEqual(len(frame), 255)
self.assertEqual(probe.checksum(frame[14:34]), 0)
self.assertEqual(struct.unpack("!H", frame[16:18])[0], 241)
transport = frame[34:]
protocol = frame[23]
pseudo = frame[26:34] + struct.pack("!BBH", 0, protocol, len(transport))
self.assertEqual(probe.checksum(transport if kind == "icmp" else pseudo + transport), 0)
self.assertIn(b"\xc0\xdb", transport)
def test_arp_and_boundaries_are_distinct(self):
arp = probe.make_frame("arp", 60, 1)
self.assertEqual(len(arp), 60)
self.assertEqual(arp[:6], b"\xff" * 6)
self.assertEqual(arp[12:14], b"\x08\x06")
self.assertNotEqual(arp, probe.make_frame("arp", 60, 2))
self.assertNotEqual(probe.make_frame("raw", 14, 1), probe.make_frame("raw", 14, 2))
for kind, size in probe.CASES:
self.assertEqual(len(probe.make_frame(kind, size, 1)), size)
for kind, size in (("raw", 13), ("udp", 41), ("tcp", 53), ("icmp", 256)):
with self.assertRaises(ValueError):
probe.make_frame(kind, size, 1)
def test_probe_exact_integrity_and_both_directions(self):
for duplex in (False, True):
a, b = FakeEndpoint(), FakeEndpoint()
a.peer, b.peer = b, a
result = probe.run_probe(a, b, count=2, timeout=0.1, interval=0, duplex=duplex)
self.assertTrue(result["passed"])
self.assertEqual(result["counts"], {"a_to_b_attempted": 16, "a_to_b_received": 16,
"b_to_a_attempted": 16, "b_to_a_received": 16})
for transfer in result["transfers"]:
self.assertEqual(transfer["sent_hex"], transfer["received_hex"])
self.assertGreaterEqual(transfer["latency_ms"], 0)
def test_lost_frames_are_not_retried_or_hidden(self):
a, b = FakeEndpoint(), FakeEndpoint()
a.peer, b.peer = b, a
a.drop = True
b.incoming.put((b"foreign traffic", time.monotonic()))
result = probe.run_probe(a, b, count=1, timeout=0.001, interval=0)
self.assertFalse(result["passed"])
self.assertEqual(result["counts"]["a_to_b_received"], 0)
self.assertEqual(result["counts"]["b_to_a_received"], 8)
self.assertEqual(len(a.frames), 8)
self.assertEqual(len(result["unexpected"]), 1)
self.assertEqual(result["retries"], 0)
def test_plan_does_not_open_endpoints(self):
with patch.object(probe, "KissEndpoint") as endpoint, contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(probe.main(["--plan"]), 0)
endpoint.assert_not_called()
if __name__ == "__main__":
unittest.main()
+281
View File
@@ -0,0 +1,281 @@
"""Preamble benchmark policy, accounting and subprocess protocol, no hardware."""
import contextlib
import io
import json
from pathlib import Path
import subprocess
import sys
import tempfile
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools/hil"))
import ethernet_preamble_bench as bench
class FakeNode:
def __init__(self):
self.identity = {"port": "fake-only"}
self.commands = []
self.guard = "> " + bench.GUARD
self.radio = "> 910.5250244,62.5,7,5"
self.tuple = "909.5,500,5,5,128"
self.primary = self.secondary = self.rx = self.fail = 0
self.enabled = True
self.bad_set = self.bad_readback = False
def command(self, text, timeout=10):
self.commands.append(text)
if text == "get lab.tx": return self.guard
if text == "get radio": return self.radio
if text == "get radio2.status": return f"> rxtx; RX=0,{self.rx} TX={self.primary},{self.secondary} switches=24"
if text == "get radio2.timing": return "> need=32,128; switch=2000us"
if text == "stats-radio-diag": return "> state=1(RX), recv=7, sent=9, errors=0"
if text == "get ethernet":
return f"> ethernet={'on; active (RX time-shared)' if self.enabled else 'off; off'}; TCP 5003 listening; RX={self.rx} TX={self.secondary} fail={self.fail}; radio2 only"
if text == "get ethernet.radio": return "> " + ("909.5,500,6,5,128" if self.bad_readback else self.tuple) + "; sync=0x16; saved"
if self.bad_set: return "Error: could not save"
if text == "set ethernet off": self.enabled = False
if text == "set ethernet on": self.enabled = True
if text.startswith("set ethernet.radio "): self.tuple = text.removeprefix("set ethernet.radio ")
return "OK - saved"
def result(seconds=2, **extra):
return dict(verified=True, handshake_ms=7000, total_ms=11000,
downlink=dict(bytes=4096, elapsed_ms=seconds * 1000, hash_verified=True),
uplink=dict(bytes=4096, elapsed_seconds=seconds, hash_verified=True), **extra)
class EthernetPreambleBenchTests(unittest.TestCase):
def run_fake(self, preambles=(128, 192), runner=None, nodes=None, **kwargs):
nodes = nodes or dict(a=FakeNode(), b=FakeNode())
return bench.run_cases(nodes, preambles, runner or (lambda request: result()), **kwargs), nodes
def test_default_only_requested_preambles_and_explicit_controls(self):
self.assertEqual(bench.DEFAULT_PREAMBLES, (128, 192, 256, 384, 512, 768))
self.assertEqual(bench.validate_preambles([32, 64, 2048, 32]), [32, 64, 2048, 32])
for values in ([], [31], [2049], [True], [128.0], [128] * 121):
with self.assertRaises(ValueError): bench.validate_preambles(values)
def test_sf5_default_unchanged_and_allowed_fixed_spreading_factors(self):
default, _ = self.run_fake(preambles=(128,))
self.assertEqual(default["sf"], 5)
self.assertEqual(default["cases"][0]["preamble_ms"], 8.192)
for sf in range(5, 10):
calls = []
report, nodes = self.run_fake(preambles=(64,), sf=sf,
runner=lambda request: calls.append(request) or result())
self.assertTrue(report["passed"], report.get("error"))
self.assertEqual(calls[0]["sf"], sf)
self.assertEqual(report["ranking"][0]["sf"], sf)
self.assertEqual(report["cases"][0]["preamble_ms"], 64 * (2 ** sf) / 500)
self.assertIn(f"SF{sf}", report["note"])
for node in nodes.values():
self.assertIn(f"set ethernet.radio 909.5,500,{sf},5,64", node.commands)
def test_bad_sf_fails_before_commands_runner_or_snapshot(self):
for sf in (4, 10, True, 7.0, "7", None):
nodes = dict(a=FakeNode(), b=FakeNode()); run = Mock()
with self.assertRaises(ValueError): bench.run_cases(nodes, [128], run, sf=sf)
run.assert_not_called()
self.assertTrue(all(not node.commands for node in nodes.values()))
with self.assertRaises(ValueError): bench.snapshot(nodes, 12, sf=sf)
self.assertTrue(all(not node.commands for node in nodes.values()))
def test_bandwidth_defaults_to500_and125_propagates_to_fixed_tuple_and_timing(self):
default, _ = self.run_fake(preambles=(128,))
self.assertEqual(default["bandwidth_khz"], 500)
for bandwidth in (125, 500):
calls = []
report, nodes = self.run_fake(preambles=(64,), bandwidth=bandwidth,
runner=lambda request: calls.append(request) or result())
self.assertTrue(report["passed"], report.get("error"))
self.assertEqual(calls[0]["bandwidth_khz"], bandwidth)
self.assertEqual(report["ranking"][0]["bandwidth_khz"], bandwidth)
self.assertEqual(report["cases"][0]["preamble_ms"], 64 * 32 / bandwidth)
self.assertEqual(report["cases"][0]["case_id"], "case-001-p64")
for node in nodes.values():
self.assertIn(f"set ethernet.radio 909.5,{bandwidth},5,5,64", node.commands)
self.assertIn(f"> 909.5,{bandwidth},5,5,64", report["cases"][0]["before"]["a"]["ethernet_radio"])
def test_invalid_bandwidth_fails_before_all_commands_and_runner(self):
for bandwidth in (62.5, 250, 126, True, 125.0, "125", None):
nodes = dict(a=FakeNode(), b=FakeNode()); run = Mock()
with self.assertRaises(ValueError): bench.run_cases(nodes, [64], run, bandwidth=bandwidth)
with self.assertRaises(ValueError): bench.snapshot(nodes, 12, bandwidth=bandwidth)
run.assert_not_called(); self.assertTrue(all(not node.commands for node in nodes.values()))
def test_cli_bandwidth125_plan_has_no_io_and_wrong_bandwidth_readback_aborts(self):
output = io.StringIO()
with patch.object(bench, "SerialNode") as serial, contextlib.redirect_stdout(output):
self.assertEqual(bench.main(["--plan", "--bandwidth", "125", "--sf", "5"]), 0)
serial.assert_not_called()
plan = json.loads(output.getvalue())
self.assertEqual(plan["radio"], "909.5,125,5,5")
self.assertEqual(plan["bandwidth_khz"], 125)
for value in ("250", "125.0"):
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
bench.main(["--plan", "--bandwidth", value])
nodes = dict(a=FakeNode(), b=FakeNode())
original = nodes["b"].command
nodes["b"].command = lambda text, **kwargs: "> 909.5,500,5,5,64" if text == "get ethernet.radio" else original(text, **kwargs)
run = Mock(return_value=result())
report, _ = self.run_fake(nodes=nodes, bandwidth=125, preambles=(64,), runner=run)
self.assertTrue(report["aborted"]); run.assert_not_called()
def test_sf7_readback_requires_sf7_not_previous_sf5_or_other_sf(self):
nodes = dict(a=FakeNode(), b=FakeNode())
run = Mock(return_value=result())
report, _ = self.run_fake(nodes=nodes, sf=7, preambles=(128,), runner=run)
self.assertTrue(report["passed"])
self.assertIn("> 909.5,500,7,5,128", report["cases"][0]["before"]["a"]["ethernet_radio"])
nodes = dict(a=FakeNode(), b=FakeNode()); nodes["b"].bad_readback = True
run = Mock(return_value=result())
report, _ = self.run_fake(nodes=nodes, sf=7, preambles=(128,), runner=run)
self.assertTrue(report["aborted"]); run.assert_not_called()
def test_cli_plan_sf_choices_and_no_io(self):
for sf in range(5, 10):
output = io.StringIO()
with patch.object(bench, "SerialNode") as serial, patch.object(bench, "python_runner") as runner, \
contextlib.redirect_stdout(output):
self.assertEqual(bench.main(["--plan", "--sf", str(sf), "--preambles", "32,64"]), 0)
serial.assert_not_called(); runner.assert_not_called()
self.assertEqual(json.loads(output.getvalue())["radio"], f"909.5,500,{sf},5")
for sf in ("4", "10", "7.0"):
with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit):
bench.main(["--plan", "--sf", sf])
def test_bad_initial_guard_radio_or_counter_prevents_all_writes_and_runner(self):
for field, value in (("guard", "> Error: unavailable"), ("radio", "> 909.5,500,5,5"), ("primary", 1)):
nodes = dict(a=FakeNode(), b=FakeNode()); setattr(nodes["b"], field, value)
runner = Mock()
report, _ = self.run_fake(nodes=nodes, runner=runner)
self.assertTrue(report["aborted"])
runner.assert_not_called()
self.assertFalse(any(text.startswith("set ") for node in nodes.values() for text in node.commands))
def test_exact_fixed_lab_tuple_only_and_complete_rf_counters(self):
nodes = dict(a=FakeNode(), b=FakeNode())
calls = []
def run(request):
calls.append(request)
for node in nodes.values():
node.secondary += 20; node.rx += 19; node.fail += 1
return result(tcp_client=dict(OutSegs=80, RetransSegs=3))
report, _ = self.run_fake(nodes=nodes, runner=run)
self.assertTrue(report["passed"], report.get("error"))
self.assertEqual([call["preamble"] for call in calls], [128, 192])
for call in calls:
self.assertEqual((call["frequency_mhz"], call["bandwidth_khz"], call["sf"], call["cr"]), (909.5, 500, 5, 5))
for node in nodes.values():
self.assertEqual([c for c in node.commands if c.startswith("set ")], [
"set ethernet off", "set ethernet.radio 909.5,500,5,5,128", "set ethernet on",
"set ethernet off", "set ethernet.radio 909.5,500,5,5,192", "set ethernet on"])
for case in report["cases"]:
self.assertEqual(case["counter_deltas"]["a"]["profile_tx"], [0, 20])
self.assertEqual(case["counter_deltas"]["b"]["ethernet"], dict(rx=19, tx=20, fail=1))
self.assertIn("sent=9", case["after"]["a"]["radio_diag"])
self.assertEqual(case["metrics"]["tcp_client"]["RetransSegs"], 3)
def test_failed_config_or_readback_aborts_before_runner(self):
for field in ("bad_set", "bad_readback"):
nodes = dict(a=FakeNode(), b=FakeNode()); setattr(nodes["b"], field, True)
runner = Mock()
report, _ = self.run_fake(nodes=nodes, runner=runner)
self.assertTrue(report["aborted"])
runner.assert_not_called()
def test_runner_timeout_records_failure_disables_both_then_continues_no_retry(self):
runner = Mock(side_effect=[subprocess.TimeoutExpired("redacted", 240), result()])
report, nodes = self.run_fake(runner=runner)
self.assertFalse(report["aborted"], report.get("error"))
self.assertFalse(report["passed"])
self.assertEqual(runner.call_count, 2)
self.assertEqual(report["cases"][0]["error"], "Runner timed out")
self.assertEqual([case["verified"] for case in report["cases"]], [False, True])
for node in nodes.values():
self.assertEqual(node.commands.count("set ethernet off"), 3)
def test_runner_false_hash_or_bad_protocol_does_not_rank_and_continues(self):
for failed in (dict(verified=False, error="SSH deadline"), "not a result", dict(verified=1)):
report, _ = self.run_fake(runner=Mock(side_effect=[failed, result()]))
self.assertFalse(report["aborted"])
self.assertEqual([item["index"] for item in report["ranking"]], [2])
bad_hash = result(); bad_hash["uplink"]["hash_verified"] = False
report, _ = self.run_fake(runner=lambda request: bad_hash)
self.assertFalse(report["passed"])
self.assertEqual(report["ranking"], [])
def test_safety_change_after_runner_aborts_remaining_cases(self):
for field, value in (("primary", 1), ("guard", "> unlocked"), ("radio", "> 909.5,500,5,5")):
nodes = dict(a=FakeNode(), b=FakeNode())
def changed(request):
setattr(nodes["b"], field, value)
return result()
report, _ = self.run_fake(nodes=nodes, runner=changed)
self.assertTrue(report["aborted"])
self.assertEqual(len(report["cases"]), 1)
self.assertEqual(report["ranking"], [])
def test_actual_verified_goodput_ranks_even_with_retransmissions_and_excludes_handshake(self):
first = result(1, tcp_client=dict(RetransSegs=6)); second = result(2, tcp_client=dict(RetransSegs=0))
first["handshake_ms"] = 9000
report, _ = self.run_fake(runner=Mock(side_effect=[first, second]))
self.assertEqual([item["preamble"] for item in report["ranking"]], [128, 192])
self.assertEqual(report["ranking"][0]["combined_payload_kbps"], 32.768)
self.assertAlmostEqual(report["ranking"][0]["end_to_end_kbps"], 8192 * 8 / 11000)
self.assertEqual(report["ranking"][0]["downlink_kbps"], 32.768)
def test_unknown_tcp_metrics_and_missing_or_reset_counters_are_not_zero(self):
metrics = bench.payload_metrics(result())
self.assertIsNone(metrics["tcp_client"]); self.assertIsNone(metrics["tcp_server"])
self.assertIsNone(bench.ethernet_counters("> unavailable"))
self.assertIsNone(bench.counter_delta(10, 2))
self.assertIsNone(bench.counter_delta(None, None))
self.assertIsNone(bench.payload_metrics(dict(verified=True))["combined_payload_kbps"])
self.assertFalse(bench.payload_metrics(dict(verified=True))["verified"])
unknown_hash = result(); del unknown_hash["uplink"]["hash_verified"]
self.assertFalse(bench.payload_metrics(unknown_hash)["verified"])
def test_progress_persists_running_before_runner_and_complete_after_checks(self):
events, saves = [], []
report, _ = self.run_fake(preambles=(128,), progress=events.append,
persist=lambda value: saves.append(json.loads(json.dumps(value))))
self.assertEqual([event["phase"] for event in events], ["checking", "running", "complete"])
running = next(save for save in saves if save["cases"] and save["cases"][0]["phase"] == "running")
self.assertNotIn("runner", running["cases"][0])
self.assertIn("commands", saves[-1])
def test_plan_has_no_io_and_seeded_repeats_are_explicit(self):
output = io.StringIO()
with patch.object(bench, "SerialNode") as serial, patch.object(bench, "SshSerialNode") as ssh, \
patch.object(bench, "python_runner") as runner, patch.object(bench.tempfile, "mkdtemp") as folder, \
contextlib.redirect_stdout(output):
self.assertEqual(bench.main(["--plan", "--preambles", "512,128,384,192,768,256", "--runs", "2", "--seed", "7"]), 0)
for mock in (serial, ssh, runner, folder): mock.assert_not_called()
plan = json.loads(output.getvalue())
self.assertEqual(len(plan["preambles"]), 12)
self.assertEqual(sorted(plan["preambles"][:6]), list(bench.DEFAULT_PREAMBLES))
def test_runner_subprocess_exact_json_no_shell_and_no_stderr_secrets(self):
with tempfile.TemporaryDirectory() as directory:
path = Path(directory) / "fixture.py"; path.touch()
completed = SimpleNamespace(stdout=json.dumps(result()), stderr="PRIVATE KEY", returncode=0)
with patch.object(bench.subprocess, "run", return_value=completed) as process:
value = bench.python_runner(path)(dict(preamble=128))
args, kwargs = process.call_args
self.assertEqual(args[0], [sys.executable, str(path)])
self.assertFalse(kwargs["shell"])
self.assertEqual(kwargs["timeout"], 240)
self.assertEqual(json.loads(kwargs["input"]), dict(preamble=128))
self.assertNotIn("PRIVATE KEY", json.dumps(value))
completed.returncode = 2
with patch.object(bench.subprocess, "run", return_value=completed):
self.assertFalse(bench.python_runner(path)(dict(preamble=128))["verified"])
if __name__ == "__main__":
unittest.main()
+217
View File
@@ -0,0 +1,217 @@
"""Physical SSH-runner protocol/ownership tests: no SSH, WSL, RF or key reads."""
import io
import json
from pathlib import Path
import shlex
import subprocess
import sys
import tempfile
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools/hil"))
import ethernet_preamble_ssh_runner as runner
def request(**changes):
return dict(index=1, case_id="case-001-p128", preamble=128, frequency_mhz=909.5,
bandwidth_khz=500, sf=5, cr=5, **changes)
def config(directory):
return dict(ssh_host="bench@example.test", ssh_key=str(Path(directory) / "ssh_key"),
pi_bench="/home/mikec/bench", server_endpoint="tcp://192.168.1.84:5003",
client_endpoint="tcp://192.168.1.87:5003", wsl_workspace="/home/mcarper/app",
flutter="/home/mcarper/flutter/bin/flutter", account="mcarper", out_root=directory)
def ready_text(config, pid=123):
return (f"READY: gateway={config['server_endpoint']} local=10.90.5.1 remote=10.90.5.2 SSH=2222 user=mikec\n"
"Ephemeral client private key file (never print contents): /home/mikec/bench/ethernet-phone.A1234567/client_key\n"
f"256 SHA256:{'a' * 43} root@host (ED25519)\n"
"Artifacts: /home/mikec/bench/ethernet-phone.A1234567\n"
f"Server helper PID: {pid}\n")
class EthernetPreambleSshRunnerTests(unittest.TestCase):
def test_config_uses_explicit_hosts_private_port5003_and_absolute_paths(self):
with tempfile.TemporaryDirectory() as directory:
value = config(directory); Path(value["ssh_key"]).touch()
self.assertEqual(runner.validate_config(value), value)
for field, bad in (("ssh_host", "-oProxyCommand=evil"), ("account", "root"),
("server_endpoint", "tcp://192.168.1.84:5002"),
("server_endpoint", "tcp://127.0.0.1:5003"),
("client_endpoint", value["server_endpoint"]),
("pi_bench", "/"), ("wsl_workspace", "/home/../etc"),
("flutter", "relative"), ("out_root", "relative")):
changed = dict(value); changed[field] = bad
with self.subTest(field=field, value=bad), self.assertRaises((ValueError, TypeError)):
runner.validate_config(changed)
def test_requests_accept_sf5_through_sf9_without_changing_case_identifier(self):
for bandwidth in (125, 500):
for sf in range(5, 10):
value = request(); value.update(sf=sf, bandwidth_khz=bandwidth)
with self.subTest(bandwidth=bandwidth, sf=sf):
self.assertIs(runner.validate_request(value), value)
self.assertEqual(value["case_id"], "case-001-p128")
def test_requests_only_allow_fixed_lab_frequency_bandwidth_cr_and_safe_case_id(self):
for field, bad in (("frequency_mhz", 910.525), ("bandwidth_khz", 250),
("cr", 7), ("preamble", 31), ("index", True),
("case_id", "../escape")):
changed = request(); changed[field] = bad
with self.subTest(field=field), self.assertRaises(ValueError):
runner.validate_request(changed)
def test_request_radio_fields_reject_wrong_types_and_invalid_sf(self):
invalid = {
"frequency_mhz": (True, False, "909.5", None),
"bandwidth_khz": (True, False, 125.0, 500.0, "125", "500", None),
"sf": (True, False, 4, 10, 5.0, 9.0, "5", None),
"cr": (True, False, 5.0, "5", None),
}
for field, values in invalid.items():
for bad in values:
changed = request(); changed[field] = bad
with self.subTest(field=field, value=bad), self.assertRaises(ValueError):
runner.validate_request(changed)
def test_windows_drive_mapping_never_expands_shell_metacharacters(self):
self.assertEqual(runner.wsl_path(r"C:\git\out with spaces\file.key"), "/mnt/c/git/out with spaces/file.key")
self.assertEqual(runner.wsl_path("/mnt/c/git/key"), "/mnt/c/git/key")
for path in (r"\\host\share\key", "relative", "/", "/tmp/../etc/key"):
with self.assertRaises(ValueError): runner.wsl_path(path)
def test_ready_requires_exact_endpoint_one_fingerprint_pid_and_scoped_artifacts(self):
value = config("/tmp/fixture")
ready = runner.parse_server_ready(ready_text(value), value)
self.assertEqual(ready["pid"], 123)
self.assertEqual(ready["fingerprint"], "SHA256:" + "a" * 43)
for text in (ready_text(value).replace("192.168.1.84", "192.168.1.87"),
ready_text(value).replace("Server helper PID: 123\n", ""),
ready_text(value) + "Server helper PID: 234\n",
ready_text(value) + "256 SHA256:" + "b" * 43 + " (ED25519)\n",
ready_text(value).replace("Artifacts: /home/mikec/bench/", "Artifacts: /etc/")):
with self.assertRaises(ValueError): runner.parse_server_ready(text, value)
def test_read_ready_waits_past_artifact_line_for_pid(self):
value = config("/tmp/fixture")
process = SimpleNamespace(stdout=io.StringIO(ready_text(value)))
self.assertEqual(runner.read_ready(process, value, .5)["pid"], 123)
with self.assertRaises(RuntimeError):
runner.read_ready(SimpleNamespace(stdout=io.StringIO(ready_text(value).split("Server helper PID")[0])), value, .5)
def test_flutter_artifact_path_must_be_exact_child_of_selected_workspace(self):
self.assertEqual(runner.artifact_dir("Flutter RF artifacts: /home/test/app/ethernet-flutter.Ab123456\n",
"/home/test/app", "Flutter RF artifacts"), "/home/test/app/ethernet-flutter.Ab123456")
for text in ("Flutter RF artifacts: /tmp/ethernet-flutter.Ab123456\n",
"Flutter RF artifacts: /home/test/app/sub/ethernet-flutter.Ab123456\n",
"Flutter RF artifacts: /home/test/app/ethernet-flutter.short\n"):
with self.assertRaises(ValueError): runner.artifact_dir(text, "/home/test/app", "Flutter RF artifacts")
def test_speed_normalization_uses_actual_time_and_both_sha256_checks(self):
speed = dict(success=True, host_verified=True, handshake_ms=7000, total_elapsed_ms=11000,
download=dict(bytes=8192, elapsed_ms=2000, kbps=9999, sha256_valid=True,
data_elapsed_ms=1000, data_kbps=65.536),
upload=dict(bytes=8192, elapsed_ms=4000, sha256_valid=True))
result = runner.normalized_speed(speed, 0)
self.assertTrue(result["verified"])
self.assertEqual(result["downlink"]["kbps"], 32.768)
self.assertEqual(result["uplink"]["elapsed_seconds"], 4)
self.assertEqual(result["downlink"]["data_kbps"], 65.536)
for field in ("success", "host_verified"):
changed = dict(speed); changed[field] = False
self.assertFalse(runner.normalized_speed(changed, 0)["verified"])
self.assertFalse(runner.normalized_speed(speed, 1)["verified"])
for bad in (None, dict(bytes=8192, elapsed_ms=0, sha256_valid=True),
dict(bytes=8192, elapsed_ms=float("nan"), sha256_valid=True),
dict(bytes=8192, elapsed_ms=2000, sha256_valid=False)):
changed = dict(speed); changed["download"] = bad
self.assertFalse(runner.normalized_speed(changed, 0)["verified"])
def test_remote_stop_checks_whole_pid_argv_script_and_unique_prefix_before_signal(self):
value = config("/tmp/fixture")
command = runner.stop_server_args(value, 123, "ethphone-speed-abcdef01")
self.assertIn("BatchMode=yes", command); self.assertIn("StrictHostKeyChecking=yes", command)
remote = shlex.split(command[-1])
self.assertEqual(remote[:4], ["sudo", "-n", "python3", "-c"])
self.assertEqual(remote[-3:], ["123", "/home/mikec/bench/ethernet_ssh_phone_bench.sh", "ethphone-speed-abcdef01"])
code = remote[4]
self.assertIn("script not in args", code)
self.assertIn("args[args.index('--prefix')+1] != prefix", code)
self.assertIn("os.kill(pid,signal.SIGTERM)", code)
self.assertNotIn("pkill", code)
for pid, prefix in ((1, "ethphone-speed-abcdef01"), (123, "ethphone-speed-abcdef01;killall")):
with self.assertRaises(ValueError): runner.stop_server_args(value, pid, prefix)
def test_failed_startup_still_removes_only_owned_key_and_owned_popen_child(self):
with tempfile.TemporaryDirectory() as directory:
value = config(directory); Path(value["ssh_key"]).touch()
process = Mock(); process.stdout = io.StringIO("")
process.wait.side_effect = [subprocess.TimeoutExpired("ssh", 3), None]
with patch.object(runner.subprocess, "Popen", return_value=process), \
patch.object(runner.subprocess, "run") as run:
result = runner.execute(value, request())
self.assertFalse(result["verified"])
process.kill.assert_called_once()
run.assert_not_called() # No WSL key was created; don't remove one.
self.assertFalse(any(path.name == "client_key.ephemeral" for path in Path(directory).rglob("*")))
def test_full_fake_case_normalizes_copies_evidence_and_cleans_only_its_resources(self):
with tempfile.TemporaryDirectory() as directory:
value = config(directory); Path(value["ssh_key"]).touch()
process = Mock(); process.stdout = io.StringIO(ready_text(value))
tcp_count = 0
commands = []
def run(argv, **kwargs):
nonlocal tcp_count
commands.append(argv)
self.assertFalse(kwargs["shell"])
self.assertNotIn("PRIVATE_KEY_CONTENT", " ".join(argv))
if argv[0] == "scp":
Path(argv[-1]).write_text("PRIVATE_KEY_CONTENT")
if argv[0] == "ssh" and "/proc/net/snmp" in argv[-1]:
tcp_count += 1
return SimpleNamespace(stdout=("Tcp: InSegs OutSegs RetransSegs\n"
f"Tcp: {tcp_count * 10} {tcp_count * 20} {tcp_count - 1}\n"), returncode=0)
if any(str(argument).endswith("/ethernet_ssh_flutter_bench.sh") for argument in argv):
return SimpleNamespace(stdout="Flutter RF artifacts: /home/mcarper/app/ethernet-flutter.Ab123456\n", returncode=0)
if argv[0] == "wsl" and "-c" in argv and "src=$1" in argv[argv.index("-c") + 1]:
output = next(Path(directory).glob("case-001-p128-*"))
(output / "speed.json").write_text(json.dumps(dict(success=True, host_verified=True,
handshake_ms=7000, total_elapsed_ms=11000,
download=dict(bytes=8192, elapsed_ms=2000, sha256_valid=True),
upload=dict(bytes=8192, elapsed_ms=2000, sha256_valid=True))))
(output / "tcp-stats.json").write_text(json.dumps(dict(available=True, delta=dict(OutSegs=40, RetransSegs=2))))
return SimpleNamespace(stdout="", returncode=0)
with patch.object(runner.subprocess, "Popen", return_value=process), \
patch.object(runner.subprocess, "run", side_effect=run):
result = runner.execute(value, request())
self.assertTrue(result["verified"], result.get("error"))
self.assertEqual(result["downlink"]["kbps"], 32.768)
self.assertEqual(result["tcp_server"]["delta"]["RetransSegs"], 1)
self.assertEqual(result["tcp_client"]["delta"]["RetransSegs"], 2)
self.assertTrue(any(argv[:7] == ["wsl", "--user", "root", "--exec", "rm", "-f", "--"] for argv in commands))
self.assertTrue(any(argv[0] == "ssh" and "os.kill(pid,signal.SIGTERM)" in argv[-1] for argv in commands))
self.assertFalse((Path(result["artifacts"]["directory"]) / "client_key.ephemeral").exists())
self.assertNotIn("PRIVATE_KEY_CONTENT", json.dumps(result))
process.kill.assert_not_called()
def test_source_has_no_network_mutation_or_key_content_copy_in_evidence_allowlist(self):
source = Path(runner.__file__).read_text()
self.assertNotIn("shell=True", source)
self.assertNotIn("pkill", source)
self.assertNotIn("killall", source)
self.assertNotIn("client_key", runner.ARTIFACTS)
self.assertNotIn("host_key", runner.ARTIFACTS)
self.assertNotIn("iptables", source)
self.assertIn('"--duration", "300"', source)
self.assertIn('deadline = time.monotonic() + 225', source)
if __name__ == "__main__":
unittest.main()
+74
View File
@@ -0,0 +1,74 @@
"""Primary/Ethernet coexistence harness guards; never opens hardware."""
from pathlib import Path
import sys
import unittest
from unittest.mock import Mock, patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools/hil"))
import ethernet_primary_probe as probe
class FakeNode:
def __init__(self, radio="> 909.5,500.000,5,5", ethernet="> 909.500,500,5,5,32; sync=0x16; saved"):
self.identity = {"port": "fake-only"}
self.commands = []
self.radio = radio
self.ethernet = ethernet
def command(self, text):
self.commands.append(text)
return {"get radio": self.radio, "get ethernet.radio": self.ethernet,
"get ethernet": "> ethernet=on; active (RX time-shared); TCP 5003 connected",
"get tx.channel 0": '> tx.channel 0 "Public" radio; active=radio',
"set tx.channel 0 radio": 'OK - tx.channel 0 "Public" radio; active=radio',
"show channels on": "Terminal display: channels on"}.get(text, "> public")
class EthernetPrimaryProbeTests(unittest.TestCase):
def test_lab_tuple_parser_is_strict_and_accepts_known_cli_formats(self):
for text in ("> 909.5,500,5,5", "get radio\r\n> 909.500,500.000,5,5\r\n> ",
"get radio\r\n -> > 909.5,500,5,5\r\n> ",
"> 909.500,500,5,5,32; sync=0x16; saved"):
probe.assert_lab_radio(text)
for text in ("> 910.525,500,5,5", "> 909.5,125,5,5", "> 909.5,500,7,5",
"> 909.5,500,5,7", "Error: unavailable", "set radio 909.5,500,5,5"):
with self.assertRaises(RuntimeError):
probe.assert_lab_radio(text)
def test_own_command_echo_is_never_an_incoming_message(self):
marker = "ETH-PRIMARY-12345678-a"
for text in ("channel 0 " + marker, "Sent to Public.\r\n> ",
"debug incoming " + marker, "CHANNEL MSG -> Public (FLOOD)\r\n wrong-message"):
self.assertFalse(probe.received_channel_marker(text, marker))
self.assertTrue(probe.received_channel_marker(
"\r\nCHANNEL MSG -> Public (FLOOD)\r\n EthernetA: " + marker + "\r\n> ", marker))
self.assertFalse(probe.received_channel_marker(
"CHANNEL MSG -> Public (FLOOD)\r\n Node: " + marker + "-different", marker))
def test_bad_primary_or_ethernet_profile_fails_before_any_tx(self):
for name in ("radio", "ethernet"):
a, b = FakeNode(), FakeNode()
setattr(b, name, "> 910.525,500,5,5")
factory = Mock()
with self.assertRaises(RuntimeError):
probe.run_primary(a, b, "tcp://a:5003", "tcp://b:5003", factory=factory)
factory.assert_not_called()
self.assertFalse(any(command.startswith("set ") or command.startswith("channel ")
for node in (a, b) for command in node.commands))
def test_channel_zero_primary_policy_is_set_and_confirmed_before_kiss(self):
a, b = FakeNode(), FakeNode()
factory = Mock(side_effect=RuntimeError("stop before opening test socket"))
with self.assertRaisesRegex(RuntimeError, "stop before"):
probe.run_primary(a, b, "tcp://a:5003", "tcp://b:5003", factory=factory)
for node in (a, b):
self.assertIn("set tx.channel 0 radio", node.commands)
self.assertIn("show channels on", node.commands)
self.assertLess(node.commands.index("set tx.channel 0 radio"),
len(node.commands) - 1 - node.commands[::-1].index("get tx.channel 0"))
if __name__ == "__main__":
unittest.main()
+185
View File
@@ -0,0 +1,185 @@
"""R2 sweep safety gates and honest RF-loss reporting, without hardware."""
import contextlib
import io
from pathlib import Path
import sys
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools/hil"))
import ethernet_radio2_sweep as sweep
class FakeNode:
def __init__(self):
self.identity = {"port": "fake-only"}
self.commands = []
self.guard = "> " + sweep.GUARD
self.radio = "> 910.525,62.5,7,5"
self.ethernet = "909.5,62.5,7,5,64"
self.primary = 0
self.bad_config = self.bad_set = False
def command(self, text, timeout=10):
self.commands.append(text)
if text == "get lab.tx": return self.guard
if text == "get radio": return self.radio
if text == "get radio2.status": return f"> rxtx; RX=0,4 TX={self.primary},2 switches=20"
if text == "get radio2.timing": return "> chirps=4.60,20.0; need=32,64; switch=2000us"
if text == "get ethernet": return "> ethernet=on; active (RX time-shared); TCP 5003 listening"
if text == "get ethernet.radio":
return "> " + ("910.525,62.5,7,5,64" if self.bad_config else self.ethernet) + "; sync=0x16; saved"
if text.startswith("set ethernet.radio "):
self.ethernet = text.removeprefix("set ethernet.radio ")
return "Error: test failed" if self.bad_set else "OK - Ethernet saved"
class FakeEndpoint:
def __init__(self):
self.decoder = SimpleNamespace(errors=0)
self.overflow = 0
self.error = None
self.closed = False
def close(self): self.closed = True
def fake_probe(*endpoints, **kwargs):
return {"passed": True, "transfers": [
{"received": True, "size": 244}, {"received": True, "size": 244}], "retries": 0}
class EthernetRadio2SweepTests(unittest.TestCase):
def run_fake(self, a=None, b=None, cases=None, probe=fake_probe, factory=None):
endpoints = []
def make(endpoint):
item = FakeEndpoint(); endpoints.append(item); return item
report = sweep.run_sweep(a or FakeNode(), b or FakeNode(), "tcp://a:5003", "tcp://b:5003",
cases or sweep.CANDIDATES[:2], probe_function=probe,
endpoint_factory=factory or make)
return report, endpoints
def test_guard_is_exact_structured_unambiguous_and_not_an_echo(self):
for text in ("> " + sweep.GUARD, "get lab.tx\r\n -> > " + sweep.GUARD + "\r\n> "):
sweep.require_guard(text)
for text in ("get lab.tx " + sweep.GUARD, "> lock=909.500 MHz", "> Error: unknown command",
"> " + sweep.GUARD.replace("909.500", "910.525"),
"> " + sweep.GUARD + "\n> " + sweep.GUARD):
with self.assertRaises(RuntimeError): sweep.require_guard(text)
def test_tuple_parser_rejects_echoes_ambiguous_and_wrong_preamble(self):
self.assertEqual(sweep.radio_tuple("get radio\n -> > 910.525,62.500,7,5\n> ")[:4], sweep.PRIMARY)
for text in ("set radio 910.525,62.5,7,5", "> unavailable",
"> 910.525,62.5,7,5\n> 910.525,62.5,7,5"):
with self.assertRaises(RuntimeError): sweep.radio_tuple(text)
def test_bad_initial_guard_or_primary_prevents_every_setting_and_socket(self):
for field, value in (("guard", "> Error: unknown command"), ("radio", "> 909.5,500,5,5")):
a, b = FakeNode(), FakeNode(); setattr(b, field, value)
factory = Mock()
report, _ = self.run_fake(a, b, factory=factory)
self.assertTrue(report["aborted"])
factory.assert_not_called()
self.assertFalse(any(command.startswith("set ") for node in (a, b) for command in node.commands))
def test_primary_nonzero_counter_refuses_configuration_and_tx(self):
a, b = FakeNode(), FakeNode(); b.primary = 1
report, endpoints = self.run_fake(a, b)
self.assertTrue(report["aborted"])
self.assertEqual(endpoints, [])
self.assertFalse(any(command.startswith("set ") for node in (a, b) for command in node.commands))
def test_primary_float_rounding_tolerates_only_50_hz_and_exact_phy(self):
for value in ("> 910.5250244,62.5,7,5", "> 910.5249756,62.5,7,5"):
a, b = FakeNode(), FakeNode(); b.radio = value
report, _ = self.run_fake(a, b)
self.assertTrue(report["passed"], report.get("error"))
for value in ("> 910.526,62.5,7,5", "> 910.5250501,62.5,7,5",
"> 910.5250244,62.5001,7,5", "> 910.5250244,62.5,8,5",
"> 910.5250244,62.5,7,6"):
a, b = FakeNode(), FakeNode(); b.radio = value
factory = Mock()
report, _ = self.run_fake(a, b, factory=factory)
self.assertTrue(report["aborted"])
factory.assert_not_called()
self.assertFalse(any(command.startswith("set ") for node in (a, b) for command in node.commands))
def test_only_predefined_9095_ethernet_settings_and_no_primary_writes(self):
a, b = FakeNode(), FakeNode()
report, endpoints = self.run_fake(a, b)
self.assertTrue(report["passed"])
self.assertTrue(all(endpoint.closed for endpoint in endpoints))
for node in (a, b):
writes = [command for command in node.commands if command.startswith("set ")]
self.assertEqual(writes, ["set ethernet off", "set ethernet.radio 909.5,62.5,7,5,64", "set ethernet on",
"set ethernet off", "set ethernet.radio 909.5,125,7,5,64", "set ethernet on"])
for item in report["cases"]:
self.assertEqual(item["primary_tx_before"], {"a": 0, "b": 0})
self.assertEqual(item["primary_tx_after"], {"a": 0, "b": 0})
self.assertIn("verified R2", item["probe"]["baseline"])
def test_wrong_readback_or_failed_setting_aborts_before_sockets(self):
for field in ("bad_config", "bad_set"):
a, b = FakeNode(), FakeNode(); setattr(b, field, True)
report, endpoints = self.run_fake(a, b)
self.assertTrue(report["aborted"])
self.assertEqual(endpoints, [])
self.assertEqual(len(report["cases"]), 1)
def test_rf_packet_loss_is_reported_and_remaining_cases_continue(self):
def loss(*endpoints, **kwargs):
report = fake_probe(); report["passed"] = False
report["transfers"][0]["received"] = False
return report
report, endpoints = self.run_fake(probe=loss)
self.assertFalse(report["aborted"])
self.assertFalse(report["passed"])
self.assertEqual(len(report["cases"]), 2)
self.assertTrue(all(item["measurement"]["loss_fraction"] == 0.5 for item in report["cases"]))
self.assertTrue(all(endpoint.closed for endpoint in endpoints))
def test_primary_counter_change_or_guard_change_stops_later_cases(self):
for field, value in (("primary", 1), ("guard", "> Error: lock disappeared")):
a, b = FakeNode(), FakeNode()
def changed(*endpoints, **kwargs):
setattr(b, field, value)
return fake_probe()
report, endpoints = self.run_fake(a, b, probe=changed)
self.assertTrue(report["aborted"])
self.assertEqual(len(report["cases"]), 1)
self.assertTrue(all(endpoint.closed for endpoint in endpoints))
def test_transport_error_aborts_without_retry_and_closes_both_sockets(self):
probe = Mock(side_effect=ConnectionError("WiFi lost"))
report, endpoints = self.run_fake(probe=probe)
self.assertTrue(report["aborted"])
self.assertEqual(probe.call_count, 1)
self.assertTrue(all(endpoint.closed for endpoint in endpoints))
def test_candidate_rate_ratios_and_real_payload_definition(self):
self.assertEqual([case.describe()["nominal_ratio_to_usa"] for case in sweep.CANDIDATES[:4]], [1, 2, 4, 8])
self.assertAlmostEqual(sweep.CANDIDATES[6].describe()["nominal_ratio_to_usa"], 22.857142857)
self.assertEqual(sweep.CANDIDATES[7].describe()["nominal_kbps"], 62.5)
self.assertEqual(sweep.CANDIDATES[7].preamble, 2048)
metrics = sweep.measurement(fake_probe(), 2)
self.assertEqual(metrics["ethernet_goodput_bps"], 1952)
self.assertEqual(metrics["ethernet_payload_goodput_bps"], 1840)
self.assertIn("not SSH", metrics["goodput_definition"])
def test_plan_does_not_open_serial_or_tcp_or_create_reports(self):
with patch.object(sweep, "SerialNode") as serial, patch.object(sweep, "KissEndpoint") as tcp, \
patch.object(sweep.tempfile, "mkdtemp") as directory, contextlib.redirect_stdout(io.StringIO()):
self.assertEqual(sweep.main(["--plan", "--cases", "2,3,4"]), 0)
serial.assert_not_called(); tcp.assert_not_called(); directory.assert_not_called()
def test_only_port5003_tcp_endpoints_without_credentials_are_valid(self):
for value in ("tcp://radio", "tcp://192.168.1.84:5003"):
self.assertTrue(sweep.endpoint_valid(value))
for value in ("serial:///dev/ttyUSB0", "tcp://radio:5002", "tcp://user:password@radio:5003",
"tcp://radio:5003/path", "tcp://radio:5003?x=1"):
self.assertFalse(sweep.endpoint_valid(value))
if __name__ == "__main__":
unittest.main()
+215
View File
@@ -0,0 +1,215 @@
"""Read-only serial ownership preflight and delayed USB reply regressions."""
import importlib.util
import contextlib
import io
import json
from pathlib import Path
import subprocess
import sys
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch
BASE = Path(__file__).resolve().parents[1] / "tools/hil"
sys.path.insert(0, str(BASE))
import serial_session as session
import ethernet_primary_probe as primary
class PortOwnershipTests(unittest.TestCase):
def run_inspector(self, result):
return patch.object(session.subprocess, "run", return_value=result)
def test_existing_owner_refuses_without_killing_anything(self):
with patch.object(session.sys, "platform", "linux"), \
patch.object(session.shutil, "which", return_value="/usr/bin/fuser"), \
self.run_inspector(SimpleNamespace(returncode=0)) as run:
with self.assertRaisesRegex(RuntimeError, "already in use"):
session.assert_port_idle("/dev/ttyUSB0")
self.assertEqual(run.call_args.args[0][:2], ["/usr/bin/fuser", "--"])
self.assertNotIn("-k", run.call_args.args[0])
def test_idle_port_is_allowed(self):
with patch.object(session.sys, "platform", "linux"), \
patch.object(session.shutil, "which", return_value="/usr/bin/fuser"), \
self.run_inspector(SimpleNamespace(returncode=1)):
session.assert_port_idle("/dev/ttyUSB0")
def test_inspection_failure_refuses(self):
with patch.object(session.sys, "platform", "linux"), \
patch.object(session.shutil, "which", return_value="/usr/bin/fuser"), \
self.run_inspector(SimpleNamespace(returncode=2)):
with self.assertRaisesRegex(RuntimeError, "fuser error"):
session.assert_port_idle("/dev/ttyUSB0")
def test_timeout_refuses(self):
with patch.object(session.sys, "platform", "linux"), \
patch.object(session.shutil, "which", return_value="/usr/bin/fuser"), \
patch.object(session.subprocess, "run", side_effect=subprocess.TimeoutExpired("fuser", 3)):
with self.assertRaisesRegex(RuntimeError, "Cannot safely inspect"):
session.assert_port_idle("/dev/ttyUSB0")
def test_missing_optional_inspector_does_not_spawn_a_process(self):
with patch.object(session.sys, "platform", "linux"), \
patch.object(session.shutil, "which", return_value=None), \
patch.object(session.subprocess, "run") as run:
session.assert_port_idle("/dev/ttyUSB0")
run.assert_not_called()
def test_non_linux_does_not_try_fuser(self):
with patch.object(session.sys, "platform", "win32"), \
patch.object(session.subprocess, "run") as run:
session.assert_port_idle("COM4")
run.assert_not_called()
class DelayedPort:
def __init__(self, clock, first_byte=1):
self.clock = clock
self.first_byte = first_byte
self.sent = False
self.in_waiting = 0
def read(self, count):
self.clock.now += 0.04
if not self.sent and self.clock.now >= self.first_byte:
self.sent = True
return b"delayed reply\r\n"
return b""
class DelayedReplyTests(unittest.TestCase):
def bench_module(self):
# Avoid importing optional pySerial just to exercise the pure read loop.
from types import ModuleType
serial = ModuleType("serial")
tools = ModuleType("serial.tools")
tools.list_ports = ModuleType("serial.tools.list_ports")
spec = importlib.util.spec_from_file_location("ethernet_bench_delayed_test", BASE / "ethernet_bench_serial.py")
module = importlib.util.module_from_spec(spec)
with patch.dict(sys.modules, {"serial": serial, "serial.tools": tools,
"serial.tools.list_ports": tools.list_ports}):
spec.loader.exec_module(module)
return module
def test_bench_waits_full_deadline_for_first_reply_byte(self):
module = self.bench_module()
clock = SimpleNamespace(now=0)
port = DelayedPort(clock, first_byte=1.2)
with patch.object(module.time, "monotonic", side_effect=lambda: clock.now):
self.assertEqual(module.read_quiet(port, timeout=3), "delayed reply\r\n")
self.assertGreaterEqual(clock.now, 1.2)
self.assertLess(clock.now, 2)
class TimedChunks:
def __init__(self, clock, events):
self.clock = clock
self.events = list(events)
self.pending = bytearray()
@property
def in_waiting(self):
return len(self.pending)
def read(self, count):
self.clock.now += 0.04
while self.events and self.events[0][0] <= self.clock.now:
self.pending.extend(self.events.pop(0)[1])
out = bytes(self.pending[:count])
del self.pending[:count]
return out
class CommandTransactionTests(unittest.TestCase):
def transact(self, events, command="set radio 909.5,500,5,5", timeout=3):
clock = SimpleNamespace(now=0)
port = TimedChunks(clock, events)
with patch.object(session.time, "monotonic", side_effect=lambda: clock.now):
reply = session.read_command_reply(port, command, timeout)
return reply, clock.now
def test_immediate_echo_then_long_nvs_pause_waits_for_actual_reply(self):
reply, elapsed = self.transact([
(0.04, b"> set radio 909.5,500,5,5\r\n"),
(1.1, b" -> OK - saved\r\n> ")])
self.assertIn("OK - saved", reply)
self.assertGreaterEqual(elapsed, 1.1)
def test_stale_other_command_reply_and_prompt_cannot_complete_this_command(self):
reply, elapsed = self.transact([
(0.04, b" -> OK - previous command\r\n> "),
(0.7, b"set radio 909.5,500,5,5\r\n"),
(1.4, b" -> OK - current command\r\n> ")])
self.assertIn("OK - current command", reply)
self.assertGreaterEqual(elapsed, 1.4)
def test_wrong_echo_does_not_count_even_with_prompt(self):
clock = SimpleNamespace(now=0)
port = TimedChunks(clock, [(0.04, b"> get radio\r\n -> > 909.5,500,5,5\r\n> ")])
with patch.object(session.time, "monotonic", side_effect=lambda: clock.now):
with self.assertRaises(session.SerialCommandTimeout) as caught:
session.read_command_reply(port, "get radio2", timeout=1)
self.assertIn("get radio", caught.exception.partial_reply)
self.assertGreaterEqual(clock.now, 1)
def test_getter_value_prefix_is_not_a_completion_prompt(self):
reply, elapsed = self.transact([
(0.04, b"> get radio\r\n -> > 909.5,500,5,5\r\n"),
(0.9, b">"), (1.2, b" ")], command="get radio")
self.assertTrue(reply.endswith("> "))
self.assertGreaterEqual(elapsed, 1.2)
def test_masked_password_echo_is_matched_exactly_without_requiring_plaintext(self):
reply, _ = self.transact([(0.04, b"> set wifi.pwd ********\r\n"),
(1.0, b" -> OK\r\n> ")], command="set wifi.pwd secret42")
self.assertNotIn("secret42", reply)
self.assertIn("OK", reply)
def test_timeout_stops_remaining_commands_and_still_attempts_stop(self):
module = DelayedReplyTests().bench_module()
port = Mock(is_open=True)
info = SimpleNamespace(device="/dev/ttyUSB-test", serial_number="test", vid=0x1A86, pid=0x7523)
output = io.StringIO()
args = ["ethernet_bench_serial.py", "--port", "/dev/ttyUSB-test",
"--command", "set radio 909.5,500,5,5", "--command", "get radio"]
with patch.object(module.serial, "Serial", return_value=port, create=True), \
patch.object(module.list_ports, "comports", return_value=[info], create=True), \
patch.object(module, "assert_port_idle"), \
patch.object(module, "open_configured_session"), \
patch.object(module, "read_quiet", return_value="WELCOME"), \
patch.object(module, "read_command_reply", side_effect=session.SerialCommandTimeout("echo only")), \
patch.object(sys, "argv", args), contextlib.redirect_stdout(output):
self.assertEqual(module.main(), 1)
written = [call.args[0] for call in port.write.call_args_list]
self.assertNotIn(b"get radio\n", written)
self.assertEqual(written[-1], b"+++MESHCORE-TERM-STOP\n")
self.assertFalse(json.loads(output.getvalue())["commands"][0]["completed"])
port.close.assert_called_once()
def test_bench_stops_after_post_reply_quiet_interval(self):
module = DelayedReplyTests().bench_module()
clock = SimpleNamespace(now=0)
with patch.object(module.time, "monotonic", side_effect=lambda: clock.now):
module.read_quiet(DelayedPort(clock, first_byte=0), timeout=3)
self.assertLess(clock.now, 0.6)
def test_bench_silence_waits_whole_deadline(self):
module = DelayedReplyTests().bench_module()
clock = SimpleNamespace(now=0)
with patch.object(module.time, "monotonic", side_effect=lambda: clock.now):
self.assertEqual(module.read_quiet(DelayedPort(clock, first_byte=99), timeout=3), "")
self.assertGreaterEqual(clock.now, 3)
def test_primary_helper_also_waits_for_delayed_first_byte(self):
clock = SimpleNamespace(now=0)
node = SimpleNamespace(port=DelayedPort(clock, first_byte=1.2))
with patch.object(primary.time, "monotonic", side_effect=lambda: clock.now):
self.assertEqual(primary.SerialNode.read(node, timeout=3), "delayed reply\r\n")
self.assertGreaterEqual(clock.now, 1.2)
if __name__ == "__main__":
unittest.main()
+206
View File
@@ -0,0 +1,206 @@
"""Host-only SSH/serial framing and ownership tests; never opens hardware."""
import io
import json
from pathlib import Path
import shlex
import subprocess
import sys
import tempfile
import threading
import unittest
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools" / "hil"))
import ethernet_serial_rpc as rpc
from ethernet_ssh_serial import SshSerialNode, ssh_command
import ethernet_radio2_sweep as sweep
def ready(**changes):
message = dict(type="ready", protocol=rpc.PROTOCOL, identity={"port": "/dev/ttyUSB0"}, hello="WELCOME")
message.update(changes)
return message
class Node:
identity = {"port": "/dev/ttyUSB0"}
hello = "WELCOME"
def __init__(self, error=None):
self.commands = []
self.closed = False
self.error = error
def command(self, command, timeout):
self.commands.append((command, timeout))
if self.error:
raise self.error
return "> OK\n"
def close(self):
self.closed = True
class BlockingOutput:
def __init__(self):
self.event = threading.Event()
def readline(self, limit):
self.event.wait(5)
return b""
def close(self):
self.event.set()
class Process:
def __init__(self, output, stall_close=False):
self.stdin = io.BytesIO()
self.stdout = output if hasattr(output, "readline") else io.BytesIO(output)
self.returncode = None
self.kills = self.terminates = 0
self.stall_close = stall_close
def poll(self):
return self.returncode
def kill(self):
self.kills += 1
self.returncode = -9
if isinstance(self.stdout, BlockingOutput):
self.stdout.event.set()
def terminate(self):
self.terminates += 1
self.returncode = -15
def wait(self, timeout):
if self.stall_close and self.returncode is None:
raise subprocess.TimeoutExpired("owned ssh", timeout)
self.returncode = self.returncode or 0
return self.returncode
class SerialRpcTests(unittest.TestCase):
def setUp(self):
self.folder = tempfile.TemporaryDirectory()
self.key = Path(self.folder.name) / "key"
self.key.touch()
def tearDown(self):
self.folder.cleanup()
def client(self, messages, **kwargs):
process = Process(b"".join(rpc.encode_message(item) for item in messages))
node = SshSerialNode("bench@example.test", str(self.key), "/tmp/helper.py", "/dev/ttyUSB0",
process_factory=lambda *args, **kw: process, **kwargs)
return node, process
def test_server_success_eof_closes_only_owned_serial(self):
node, output = Node(), io.BytesIO()
request = rpc.encode_message(dict(id=1, command="get lab.tx", timeout=12))
self.assertEqual(rpc.serve(node, io.BytesIO(request), output), 0)
messages = [json.loads(line) for line in output.getvalue().splitlines()]
self.assertEqual(messages[0]["protocol"], rpc.PROTOCOL)
self.assertEqual(messages[1], dict(id=1, completed=True, reply="> OK\n"))
self.assertEqual(node.commands, [("get lab.tx", 12)])
self.assertTrue(node.closed)
def test_invalid_requests_never_reach_usb_and_close(self):
invalid = [b"{}\n", b"[]\n", b"{}", b"x" * (rpc.MAX_LINE + 1),
rpc.encode_message(dict(id=1, command="get radio\nset radio bad", timeout=1)),
rpc.encode_message(dict(id=True, command="get radio", timeout=1)),
rpc.encode_message(dict(id=1, command="get radio", timeout=31)),
b'{"id":1,"command":"get radio","timeout":NaN}\n']
for line in invalid:
with self.subTest(line=line[:80]):
node, output = Node(), io.BytesIO()
self.assertEqual(rpc.serve(node, io.BytesIO(line), output), 1)
self.assertFalse(node.commands)
self.assertTrue(node.closed)
self.assertFalse(json.loads(output.getvalue().splitlines()[-1])["completed"])
def test_serial_timeout_not_reported_completed_or_followed_by_new_command(self):
node, output = Node(TimeoutError("serial deadline")), io.BytesIO()
request = rpc.encode_message(dict(id=1, command="get radio", timeout=.1))
rpc.serve(node, io.BytesIO(request + request), output)
self.assertEqual(len(node.commands), 1)
self.assertFalse(json.loads(output.getvalue().splitlines()[-1])["completed"])
self.assertTrue(node.closed)
def test_shell_arguments_are_quoted_and_host_keys_stay_pinned(self):
helper, port = "/tmp/has ' quotes/helper; nope.py", "/dev/serial/by-id/a; $(false)"
argv = ssh_command("bench@example.test", str(self.key), helper, port)
self.assertIn("StrictHostKeyChecking=yes", argv)
self.assertIn("BatchMode=yes", argv)
self.assertEqual(shlex.split(argv[-1]), ["python3", helper, "--port", port])
for host in ("-oProxyCommand=bad", "host\nother", "user@host;touch x", "host other"):
with self.assertRaises(ValueError):
ssh_command(host, str(self.key), helper, port)
with self.assertRaises(ValueError):
ssh_command("host", str(self.key), "relative.py", port)
def test_client_complete_response_logged_and_normal_close_is_eof(self):
node, process = self.client([ready(), dict(id=1, completed=True, reply="> OK\n")])
self.assertEqual(node.command("get radio", timeout=1), "> OK\n")
request = json.loads(process.stdin.getvalue())
self.assertEqual(request, dict(id=1, command="get radio", timeout=1))
self.assertEqual(node.commands, [dict(command="get radio", reply="> OK\n")])
node.close()
node.close()
self.assertTrue(process.stdin.closed)
self.assertEqual(process.kills, 0)
def test_client_refuses_incomplete_mismatched_and_eof_replies(self):
for messages in ([ready()], [ready(), dict(id=2, completed=True, reply="OK")],
[ready(), dict(id=1, completed=False, error="timeout")],
[ready(), dict(id=1, completed=1, reply="OK")]):
with self.subTest(messages=messages):
node, process = self.client(messages)
with self.assertRaises(RuntimeError):
node.command("get radio", timeout=.01)
self.assertFalse(node.commands)
self.assertTrue(node._closed)
self.assertEqual(process.kills, 1)
def test_invalid_startup_identity_protocol_or_framing(self):
for data in (rpc.encode_message(ready(protocol="other")), rpc.encode_message(ready(identity={})),
b"debug text\n", b"x" * (rpc.MAX_LINE + 1), b"{}"):
process = Process(data)
with self.subTest(data=data[:80]), self.assertRaises((ValueError, RuntimeError)):
SshSerialNode("host", str(self.key), "/tmp/helper", "/dev/ttyUSB0",
process_factory=lambda *args, **kw: process)
self.assertTrue(process.stdin.closed)
def test_startup_deadline_kills_only_owned_child(self):
process = Process(BlockingOutput())
with self.assertRaises(TimeoutError):
SshSerialNode("host", str(self.key), "/tmp/helper", "/dev/ttyUSB0",
process_factory=lambda *args, **kw: process, startup_timeout=.01)
self.assertEqual(process.kills, 1)
self.assertTrue(process.stdin.closed)
def test_normal_close_has_bounded_terminate_fallback(self):
node, process = self.client([ready()])
process.stall_close = True
node.close()
self.assertEqual(process.terminates, 1)
self.assertEqual(process.kills, 0)
def test_client_rejects_bad_commands_before_writing(self):
node, process = self.client([ready()])
for command, timeout in (("get x\nset x", 1), ("x" * 513, 1), ("get x", float("inf")), ("get x", True)):
with self.assertRaises(ValueError):
node.command(command, timeout)
self.assertEqual(process.stdin.getvalue(), b"")
node.close()
def test_fast_board_cases_are_exact_and_909_5_only(self):
self.assertEqual([case.tuple_text for case in sweep.CANDIDATES[9:]], [
"909.5,500,5,5,32", "909.5,500,5,5,64", "909.5,500,5,5,128", "909.5,500,5,5,256",
"909.5,500,7,5,32", "909.5,500,7,5,64", "909.5,250,7,5,32", "909.5,500,8,5,32",
"909.5,125,7,5,32", "909.5,250,8,5,32", "909.5,250,7,5,48",
"909.5,500,5,5,512", "909.5,125,7,5,48"])
if __name__ == "__main__":
unittest.main()
+61
View File
@@ -0,0 +1,61 @@
"""Isolated KISS SSH/SFTP fixture contract; no services, namespaces or radios."""
from pathlib import Path
import re
import shlex
import unittest
ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "tools/hil/ethernet_ssh_phone_bench.sh"
def sshd_arguments(source):
block = re.search(r"^options=\((.*?)\)\s*$", source, re.M | re.S)
if block is None:
raise AssertionError("Missing explicit isolated sshd option array")
return shlex.split(block[1])
class EthernetPhoneSftpFixtureTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.source = SCRIPT.read_text()
def test_dev_null_configuration_explicitly_enables_internal_sftp(self):
arguments = sshd_arguments(self.source)
self.assertEqual(arguments[:2], ["-f", "/dev/null"])
self.assertIn("Subsystem=sftp internal-sftp", arguments)
position = arguments.index("Subsystem=sftp internal-sftp")
self.assertEqual(arguments[position - 1], "-o")
self.assertEqual(arguments.count("Subsystem=sftp internal-sftp"), 1)
# With -f /dev/null the previous array inherited no system subsystem.
old = self.source.replace(" -o 'Subsystem=sftp internal-sftp'", "")
self.assertNotIn("Subsystem=sftp internal-sftp", sshd_arguments(old))
def test_server_remains_private_key_only_and_namespace_scoped(self):
arguments = sshd_arguments(self.source)
for option in ("ListenAddress=10.90.5.2", "AllowUsers=mikec",
"PermitRootLogin=no", "UsePAM=no",
"PasswordAuthentication=no", "KbdInteractiveAuthentication=no",
"AuthenticationMethods=publickey", "DisableForwarding=yes",
"PermitUserEnvironment=no", "X11Forwarding=no"):
self.assertIn(option, arguments)
self.assertIn('chmod 700 -- "$run_dir"', self.source)
self.assertIn('ip netns exec "$namespace" /usr/sbin/sshd -D -e "${options[@]}"', self.source)
self.assertIn('/usr/sbin/sshd -t "${options[@]}"', self.source)
for forbidden in ("systemctl", "service ssh", "/etc/ssh/sshd_config",
"useradd", "usermod", "iptables", "nft ", "ip_forward"):
self.assertNotIn(forbidden, self.source)
def test_sftp_change_preserves_kiss_mtu_and_owned_cleanup(self):
for required in ('mtu 230 up', '--interface lora0',
'"$observed" == "$inode"', '"$parent" == "$$"',
'rm -f -- "$run_dir/host_key" "$run_dir/client_key"',
'Namespace already exists; refusing to reuse it.'):
self.assertIn(required, self.source)
self.assertNotIn("killall", self.source)
self.assertNotIn("pkill", self.source)
self.assertNotIn("rm -rf", self.source)
if __name__ == "__main__":
unittest.main()
+279
View File
@@ -0,0 +1,279 @@
"""TAP relay byte/backpressure checks without privileged resources or sockets."""
from pathlib import Path
import sys
from types import SimpleNamespace
import unittest
from unittest.mock import Mock, patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools/hil"))
import ethernet_tap_relay as relay
from ethernet_kiss_probe import kiss_encode, KissDecoder
class EthernetTapRelayTests(unittest.TestCase):
def test_windows_kiss_transport_keeps_raw_bytes_and_separate_fds(self):
shared = Mock(read_fd=77, write_fd=78)
payload = b"\xc0\x10\0\xdb\xdc\xff\xc0"
shared.read.return_value = payload
shared.write.return_value = len(payload)
with patch.object(relay, "WindowsUsbStream", return_value=shared) as opened:
stream = relay.KissStream("winusb://COM6", 115200, windows_python="/mnt/c/python.exe",
expect_serial="EXACT", expect_vid=0x303a, expect_pid=0x1001,
serial_dtr=True, duration=300)
self.assertEqual((stream.read_fd, stream.write_fd), (77, 78))
self.assertEqual(stream.read(99), payload)
self.assertEqual(stream.write(payload), len(payload))
stream.close()
opened.assert_called_once_with("winusb://COM6", python="/mnt/c/python.exe", expected_serial="EXACT",
expected_vid=0x303a, expected_pid=0x1001, baud=115200, serial_dtr=True, duration=300)
shared.write.assert_called_once_with(payload)
shared.close.assert_called_once()
def test_kiss_relay_uses_separate_pipe_fds_without_changing_modem_pacing(self):
stream = Mock(read_fd=77, write_fd=78)
stream.write.side_effect = len
selections = []
def selected(reads, writes, errors, timeout):
selections.append((reads, writes))
return [], [78], []
with patch.object(relay.select, "select", side_effect=selected):
stats = relay.relay(stream, 88, duration=1, kiss_modem=True, kiss_port=1,
stop=lambda: len(selections) >= 3)
self.assertEqual(selections, [([88, 77], [78])] * 3)
self.assertEqual(stats["modem_controls_written"], 3)
self.assertEqual(stats["kiss_written"], 0)
def test_selected_port_preserves_bytes_and_ignores_other_data_and_full_modem_metadata(self):
frame = bytes(range(255))
metadata = b"\xc0\x06\xf9\x00\x01\xc0\x06\xf8\x01\xc0"
for port in (0, 1):
buffers = relay.RelayBuffers(kiss_port=port)
buffers.queue_tap_frame(frame)
self.assertEqual(buffers.to_kiss[0][1], port << 4)
self.assertEqual(KissDecoder(port=port).feed(buffers.to_kiss[0]), [frame])
buffers.feed_kiss(metadata + kiss_encode(b"other" * 12, port=1 - port) + kiss_encode(frame, port=port))
self.assertEqual(list(buffers.to_tap), [frame])
self.assertEqual(buffers.stats()["decode_errors"], 0)
for invalid in (True, -1, 2, 1.0):
with self.assertRaises(ValueError): relay.RelayBuffers(kiss_port=invalid)
with self.assertRaises(ValueError): kiss_encode(frame, port=invalid)
def test_bench_endpoint_allowlist_is_explicit_and_does_not_access_devices(self):
for value in ("serial:///dev/serial/by-id/usb-RAK_4631_ABC-if00", "serial:///dev/serial/by-path/platform-usb-0:1.2:1.0",
"serial:///dev/ttyACM0", "serial:///dev/ttyUSB0", "tcp://192.168.1.84:5003"):
self.assertIn(relay.validate_endpoint(value, strict_tcp=True).scheme, ("serial", "tcp"))
for value in ("serial:///dev/mem", "serial:///etc/passwd", "serial://host/dev/ttyACM0",
"serial:///dev/serial/by-id/../by-path/device", "serial:///dev/ttyACM0?baud=1",
"serial:///dev/serial/by-id/%2fetc", "tcp://radio:5003", "tcp://192.168.1.84:5002",
"tcp://user:secret@192.168.1.84:5003", "tcp://192.168.1.84:5003/path"):
with self.assertRaises(ValueError): relay.validate_endpoint(value, strict_tcp=True)
with patch.object(relay.os.path, "realpath", return_value="/dev/mem"), \
patch.object(relay, "open_configured_session") as opened:
with self.assertRaises(ValueError):
relay.KissStream("serial:///dev/serial/by-id/explicit-device", 115200)
opened.assert_not_called()
def test_explicit_full_modem_initialization_then_one_data_until_successful_tx_done(self):
clock = [100.0]
buffers = relay.RelayBuffers(kiss_port=1, kiss_modem=True, now=lambda: clock[0])
frames = [b"a" * 60, b"b" * 60]
for frame in frames: buffers.queue_tap_frame(frame)
written = []
while buffers.controls:
buffers.write_kiss(lambda data: written.append(data) or len(data))
self.assertEqual(written, [b"\xc0\x01\0\xc0", b"\xc0\x02\xff\xc0", b"\xc0\x03\x01\xc0"])
buffers.write_kiss(lambda data: written.append(data) or len(data))
self.assertEqual(written[-1], kiss_encode(frames[0], port=1))
self.assertFalse(buffers.can_write_kiss())
writer = Mock(); buffers.write_kiss(writer); writer.assert_not_called()
buffers.feed_kiss(b"\xc0\x06\xf9\x00\x01\xc0")
self.assertEqual(buffers.stats()["modem_rx_meta"], 1)
self.assertFalse(buffers.can_write_kiss())
buffers.feed_kiss(b"\xc0\x16\xf8\x01\xc0") # Wrong control port cannot release it.
self.assertFalse(buffers.can_write_kiss())
buffers.feed_kiss(b"\xc0\x06\xf8\x01\xc0")
self.assertTrue(buffers.can_write_kiss())
self.assertEqual(buffers.stats()["modem_tx_done"], 1)
buffers.write_kiss(lambda data: written.append(data) or len(data))
self.assertEqual(written[-1], kiss_encode(frames[1], port=1))
clock[0] += 10
with self.assertRaises(relay.ModemFlowError) as failure: buffers.check_tx_done()
self.assertEqual(failure.exception.stats["modem_tx_timeout"], 1)
def test_full_modem_errors_fail_closed_no_next_data_or_automatic_retry(self):
for control in (b"\xc0\x06\xf8\x02\xc0", b"\xc0\x06\xf1\x07\xc0", b"\xc0\x06\xf8\xc0"):
buffers = relay.RelayBuffers(kiss_modem=True)
buffers.queue_tap_frame(b"a" * 60); buffers.queue_tap_frame(b"b" * 60)
while buffers.controls: buffers.write_kiss(len)
buffers.write_kiss(len)
with self.assertRaises(relay.ModemFlowError) as failure: buffers.feed_kiss(control)
self.assertEqual(failure.exception.stats["modem_tx_error"], 1)
self.assertEqual(len(buffers.to_kiss), 1)
self.assertFalse(buffers.can_write_kiss())
def test_rejected_tx_done_drops_one_frame_then_allows_next_without_replay(self):
buffers = relay.RelayBuffers(kiss_port=1, kiss_modem=True)
frames = [b"first" * 12, b"other" * 12]
for frame in frames: buffers.queue_tap_frame(frame)
written = []
while buffers.controls: buffers.write_kiss(len)
buffers.write_kiss(lambda data: written.append(data) or len(data))
self.assertFalse(buffers.can_write_kiss())
buffers.feed_kiss(b"\xc0\x06\xf8\x00\xc0")
self.assertTrue(buffers.can_write_kiss())
self.assertEqual(buffers.stats()["modem_tx_error"], 1)
self.assertEqual(buffers.stats()["modem_tx_done"], 0)
buffers.write_kiss(lambda data: written.append(data) or len(data))
self.assertEqual(written, [kiss_encode(frame, port=1) for frame in frames])
self.assertEqual(list(buffers.to_kiss), [])
self.assertEqual(buffers.stats()["kiss_written"], 2)
self.assertTrue(buffers.stats()["awaiting_tx_done"])
buffers.feed_kiss(b"\xc0\x06\xf8\x01\xc0")
self.assertEqual(buffers.stats()["modem_tx_done"], 1)
self.assertFalse(buffers.stats()["awaiting_tx_done"])
def test_unsolicited_tx_done_does_not_create_a_credit_or_change_default_tcp_flow(self):
buffers = relay.RelayBuffers(kiss_modem=True)
buffers.feed_kiss(b"\xc0\x06\xf8\x01\xc0")
self.assertEqual(buffers.stats()["modem_unexpected_tx_done"], 1)
default = relay.RelayBuffers()
self.assertFalse(default.controls)
default.queue_tap_frame(b"a" * 60); default.queue_tap_frame(b"b" * 60)
default.write_kiss(len); default.write_kiss(len)
self.assertEqual(default.stats()["kiss_written"], 2)
self.assertFalse(default.stats()["awaiting_tx_done"])
def test_phone_fixture_keeps_default_tcp_and_explicitly_forwards_serial_modem_options(self):
source = (Path(relay.__file__).parent / "ethernet_ssh_phone_bench.sh").read_text()
self.assertIn("kiss_port=0; baud=115200; serial_dtr=off; kiss_modem=off", source)
self.assertIn('validate_endpoint(sys.argv[1], strict_tcp=True)', source)
self.assertIn('--baud "$baud" --kiss-port "$kiss_port" --serial-dtr "$serial_dtr"', source)
self.assertIn('modem_args+=(--kiss-modem)', source)
self.assertIn('Server helper PID: %s', source)
self.assertIn('"$observed" == "$inode"', source)
def test_tcp_transport_defaults_to_dedicated_kiss_port(self):
with patch.object(relay.socket, "create_connection") as connect:
endpoint = relay.KissStream("tcp://bench-radio", 115200)
connect.assert_called_once_with(("bench-radio", 5003), timeout=5)
endpoint.close()
def test_current_host_namespace_is_rejected_before_setns(self):
identity = SimpleNamespace(st_dev=5, st_ino=700)
with patch.object(relay.os, "open", return_value=88), \
patch.object(relay.os, "O_CLOEXEC", 0, create=True), \
patch.object(relay.os, "fstat", return_value=identity), \
patch.object(relay.os, "stat", return_value=identity), \
patch.object(relay.os, "close") as close, \
patch.object(relay.os, "setns", create=True) as setns:
with self.assertRaisesRegex(ValueError, "current host"):
relay.enter_network_namespace("/proc/self/ns/net")
setns.assert_not_called()
close.assert_called_once_with(88)
def test_distinct_namespace_is_entered_and_handle_closed(self):
with patch.object(relay.os, "open", return_value=88), \
patch.object(relay.os, "O_CLOEXEC", 0, create=True), \
patch.object(relay.os, "fstat", return_value=SimpleNamespace(st_dev=5, st_ino=701)), \
patch.object(relay.os, "stat", return_value=SimpleNamespace(st_dev=5, st_ino=700)), \
patch.object(relay.os, "close") as close, \
patch.object(relay.os, "setns", create=True) as setns:
relay.enter_network_namespace("/var/run/netns/bench-a")
setns.assert_called_once_with(88, 0x40000000)
close.assert_called_once_with(88)
def test_kiss_is_opened_before_namespace_or_tap(self):
calls = []
stream = Mock()
with patch.object(relay, "KissStream", side_effect=lambda *args, **kwargs: calls.append("socket") or stream), \
patch.object(relay, "enter_network_namespace", side_effect=lambda path: calls.append("netns")), \
patch.object(relay, "create_tap", side_effect=lambda name: calls.append("tap") or (77, name)):
self.assertEqual(relay.open_transport_then_tap("tcp://host:5003", 115200, "/ns", "lora0"),
(stream, 77, "lora0"))
self.assertEqual(calls, ["socket", "netns", "tap"])
stream.close.assert_not_called()
def test_failed_namespace_or_tap_closes_kiss(self):
for stage in ("enter_network_namespace", "create_tap"):
stream = Mock()
with patch.object(relay, "KissStream", return_value=stream), \
patch.object(relay, "enter_network_namespace"), \
patch.object(relay, "create_tap", return_value=(77, "lora0")), \
patch.object(relay, stage, side_effect=OSError("denied")):
with self.assertRaises(OSError):
relay.open_transport_then_tap("tcp://host:5003", 115200, "/ns", "lora0")
stream.close.assert_called_once()
def test_partial_stream_writes_preserve_exact_frame(self):
buffers = relay.RelayBuffers(1)
frame = bytes(range(255))
self.assertTrue(buffers.queue_tap_frame(frame))
self.assertFalse(buffers.can_read_tap())
with self.assertRaises(BufferError):
buffers.queue_tap_frame(frame)
written = bytearray()
def write(data):
written.extend(data[:3])
return min(3, len(data))
buffers.write_kiss(lambda data: 0)
while buffers.to_kiss:
buffers.write_kiss(write)
self.assertTrue(buffers.can_read_tap())
self.assertEqual(bytes(written), kiss_encode(frame))
self.assertEqual(KissDecoder().feed(bytes(written)), [frame])
self.assertEqual(buffers.stats()["kiss_written"], 1)
def test_incoming_staging_applies_backpressure_without_losing_frames(self):
buffers = relay.RelayBuffers(1)
frames = [bytes((index,)) * 255 for index in range(3)]
buffers.feed_kiss(b"".join(kiss_encode(frame) for frame in frames))
self.assertEqual(len(buffers.to_tap), 1)
self.assertTrue(buffers.staging)
self.assertFalse(buffers.can_read_kiss())
with self.assertRaises(BufferError):
buffers.feed_kiss(b"more bytes")
written = []
for frame in frames:
buffers.write_tap(lambda data: written.append(data) or len(data))
buffers.parse_pending()
self.assertEqual(written, frames)
self.assertTrue(buffers.can_read_kiss())
self.assertEqual(buffers.stats()["kiss_read"], 3)
self.assertEqual(buffers.stats()["tap_written"], 3)
def test_partial_tap_write_is_not_retried_as_a_new_frame(self):
buffers = relay.RelayBuffers()
buffers.feed_kiss(kiss_encode(b"x" * 60))
with self.assertRaisesRegex(IOError, "partial TAP"):
buffers.write_tap(lambda frame: 20)
self.assertEqual(buffers.stats()["tap_written"], 0)
self.assertEqual(len(buffers.to_tap), 1)
def test_invalid_frame_sizes_are_dropped_not_truncated(self):
buffers = relay.RelayBuffers()
for size in (0, 13, 256, 1500):
self.assertFalse(buffers.queue_tap_frame(b"x" * size))
self.assertEqual(len(buffers.to_kiss), 0)
buffers.feed_kiss(kiss_encode(b"x" * 13))
self.assertEqual(len(buffers.to_tap), 0)
self.assertEqual(buffers.stats()["bad_frame_length"], 5)
for size in (14, 255):
self.assertTrue(buffers.queue_tap_frame(b"x" * size))
def test_bad_stream_cannot_grow_unbounded(self):
buffers = relay.RelayBuffers()
with self.assertRaises(BufferError):
buffers.feed_kiss(b"x" * 4097)
buffers.feed_kiss(b"\xc0\0" + b"x" * 3500)
self.assertEqual(buffers.staging, b"")
self.assertLessEqual(len(buffers.decoder.buffer), 257)
self.assertEqual(buffers.stats()["decode_errors"], 1)
if __name__ == "__main__":
unittest.main()
+57
View File
@@ -0,0 +1,57 @@
"""No network namespaces or radio access: production TCP evidence parser tests."""
import json
from pathlib import Path
import sys
import tempfile
import unittest
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools" / "hil"))
import ethernet_tcp_stats as stats
class TcpStatsTests(unittest.TestCase):
def test_linux_snmp_counter_pair_and_handshake_inclusive_delta(self):
before = stats.parse_tcp("Ip: ignored\nTcp: RtoAlgorithm CurrEstab InSegs OutSegs RetransSegs OutRsts\nTcp: 1 0 20 30 2 0\n")
after = stats.parse_tcp("Tcp: RtoAlgorithm CurrEstab InSegs OutSegs RetransSegs OutRsts\nTcp: 1 1 40 80 7 1\n")
report = stats.summarize(before, after)
self.assertEqual(report["delta"], dict(InSegs=20, OutSegs=50, RetransSegs=5, OutRsts=1))
self.assertEqual(report["out_segment_retransmission_ratio"], .1)
self.assertNotIn("CurrEstab", report["delta"])
self.assertIn("total only", report["retransmission_scope"])
def test_ambiguous_missing_negative_or_reset_counters_are_not_evidence(self):
for text in ("", "Tcp: InSegs OutSegs\nTcp: 1 2\n", "Tcp: InSegs OutSegs RetransSegs\nTcp: 1 2 -1\n",
"Tcp: InSegs OutSegs RetransSegs\nTcp: 1 2 0\nTcp: 2 3 0\n"):
with self.assertRaises(ValueError):
stats.parse_tcp(text)
with self.assertRaises(ValueError):
stats.summarize(dict(InSegs=2, OutSegs=2, RetransSegs=0), dict(InSegs=1, OutSegs=2, RetransSegs=0))
def test_zero_out_segments_has_no_invented_ratio(self):
value = dict(InSegs=0, OutSegs=0, RetransSegs=0)
self.assertIsNone(stats.summarize(value, value)["out_segment_retransmission_ratio"])
def test_missing_snapshot_produces_explicit_unavailable_artifact(self):
with tempfile.TemporaryDirectory() as directory:
self.assertEqual(stats.main(["--directory", directory]), 0)
report = json.loads((Path(directory) / "tcp-stats.json").read_text())
self.assertFalse(report["available"])
def test_shell_default_mode_bounds_namespace_ownership_and_evidence_paths(self):
source = (Path(__file__).resolve().parents[1] / "tools/hil/ethernet_ssh_flutter_bench.sh").read_text()
self.assertIn("test_name=rf; test_timeout=210", source)
self.assertIn('--test-name) test_name=$2', source)
self.assertIn('10#$2 <= 180', source)
self.assertIn('--phase-timeout)', source)
self.assertIn('LORA_SSH_SPEED_TIMEOUT_SECONDS="$phase_timeout"', source)
self.assertIn('LORA_SSH_SPEED_REPORT="$run_dir/speed.json"', source)
self.assertIn('capture_tcp after || true', source)
self.assertIn('wait "$test_pid" || test_result=$?', source)
self.assertIn('exit "$test_result"', source)
self.assertIn('local remaining=300', source)
self.assertIn('"$observed" == "$inode"', source)
self.assertNotIn('cat /proc/net/snmp > /', source)
if __name__ == "__main__":
unittest.main()
+378
View File
@@ -0,0 +1,378 @@
"""Host-only RAW point IPv4 framing/MTU/state tests; no sockets, USB or TUN."""
import io
from pathlib import Path
import struct
import sys
from types import SimpleNamespace
import unittest
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools" / "hil"))
import meshcore_ip_relay as ip
def packet(size=40, reverse=False, flags=0x4000, protocol=6, header=20):
result = bytearray(size)
result[0] = 0x40 | (header // 4)
result[8], result[9] = 64, protocol
result[12:16] = bytes((10, 90, 5, 2 if reverse else 1))
result[16:20] = bytes((10, 90, 5, 1 if reverse else 2))
result[header + 12] = 0x50
struct.pack_into("!H", result, 2, size)
struct.pack_into("!H", result, 6, flags)
checksum = sum(struct.unpack("!" + "H" * (header // 2), result[:header]))
while checksum >> 16:
checksum = (checksum & 0xffff) + (checksum >> 16)
struct.pack_into("!H", result, 10, ~checksum & 0xffff)
return bytes(result)
def codec(path=None, back=0, **kwargs):
return ip.IpCodec("10.90.5.1", "10.90.5.2", path or ip.MeshPath(), back,
session=0x11223344, **kwargs)
def response(frame):
return b">" + struct.pack("<H", len(frame)) + frame
def incoming(size=40):
peer = ip.IpCodec("10.90.5.2", "10.90.5.1", ip.MeshPath(), 0, session=7)
return bytes((0x84, 0, 0, 255)) + peer.command(packet(size, reverse=True))[2:]
def started_buffers(**kwargs):
buffers = ip.IpRelayBuffers(codec(), **kwargs)
buffers.write_companion(lambda data: len(data))
buffers.feed_companion(response(bytes((5, 0))))
return buffers
class MeshCoreIpTests(unittest.TestCase):
def test_optional_ready_callback_follows_real_app_start_reply_once(self):
buffers = ip.IpRelayBuffers(codec())
events = []
iteration = [0]
stream = mock.Mock(read_fd=77, write_fd=77)
stream.write.side_effect = lambda data: events.append("app-start-written") or len(data)
stream.read.side_effect = lambda count: events.append("self-info-received") or response(bytes((5, 0)))
def ready(reads, writes, errors, timeout):
iteration[0] += 1
if iteration[0] == 1: return [], [77], []
if iteration[0] == 2: return [77], [], []
return [], [], []
with mock.patch.object(ip.select, "select", side_effect=ready):
stats = ip.relay(stream, 88, buffers, duration=1,
stop=lambda: iteration[0] >= 4, on_ready=lambda: events.append("ready"))
self.assertEqual(events, ["app-start-written", "self-info-received", "ready"])
self.assertTrue(stats["ready"])
def test_path_width_count_and_no_flood(self):
self.assertEqual(ip.MeshPath.from_hex("aabbccdd", 2).encoded, 0x42)
self.assertEqual(ip.MeshPath().encoded, 0)
for data, width in ((bytes(64), 1), (bytes(65), 2), (bytes(3), 2), (b"", 4), (b"", 0)):
with self.subTest(width=width, length=len(data)), self.assertRaises(ValueError):
ip.MeshPath(data, width)
self.assertEqual(ip.MeshPath(bytes(64), 2).encoded, 0x60)
with self.assertRaises(ValueError):
ip.MeshPath.from_hex("f")
def test_mtu_derived_for_both_paths_and_host_frame_cap(self):
self.assertEqual(ip.mtu_budget(0, 0), (156, 156))
self.assertEqual(ip.mtu_budget(0, 64), (94, 94))
self.assertEqual(ip.mtu_budget(64, 0), (94, 94))
self.assertEqual(ip.mtu_budget(4, 2), (154, 154))
self.assertEqual(ip.mtu_budget(0, 0, requested=230), (156, 156))
self.assertEqual(ip.mtu_budget(0, 0, requested=128), (156, 128))
self.assertEqual(ip.mtu_budget(64, 64, transport_bytes=4), (94, 94))
for values in ((65, 0, None), (0, -1, None), (0, 0, 93)):
with self.assertRaises(ValueError):
ip.mtu_budget(*values)
def test_command_wire_golden_matches_dart_and_nonces_change(self):
link = codec(ip.MeshPath.from_hex("aabb", 2))
first, second = link.command(packet()), link.command(packet())
self.assertEqual(first[:16].hex(), "1941aabb535348014433221100000000")
self.assertEqual(second[12:16], b"\x01\0\0\0")
self.assertNotEqual(first, second)
self.assertEqual(first[16:], packet())
self.assertEqual(ip.companion_frame(first)[:3], b"<\x38\0")
def test_sequence_exhaustion_never_wraps(self):
link = codec(sequence=0xffffffff)
self.assertEqual(link.command(packet())[10:14], b"\xff" * 4)
with self.assertRaises(OverflowError):
link.command(packet())
def test_ipv4_limits_direction_checksum_fragments_protocol_and_padding(self):
link = codec()
self.assertEqual(len(link.command(packet(156))), 170)
for value in (packet(157), packet(reverse=True), packet(flags=0x2000), packet(flags=1),
packet(flags=0x8000), packet(protocol=17), packet() + b"\0", packet()[:10]):
with self.subTest(value=value[:10]), self.assertRaises(ValueError):
link.command(value)
damaged = bytearray(packet())
damaged[8] ^= 1
with self.assertRaises(ValueError):
link.command(damaged)
self.assertTrue(ip.valid_ipv4(packet(header=24, size=44), link.local, link.remote, 156))
def test_raw_push_only_valid_peer_packet_not_text_or_other_apps(self):
link = codec()
self.assertEqual(link.raw_push(incoming()), packet(reverse=True))
for frame in (incoming()[:35], b"\x08" + incoming()[1:], incoming()[:4] + b"OTHER" + incoming()[9:],
incoming() + b"\0", bytes((0x84, 0, 0, 255)) + link.command(packet())[2:]):
self.assertIsNone(link.raw_push(frame))
def test_stream_every_split_boundary_and_wrong_direction_length(self):
wire = response(incoming())
for split in range(len(wire) + 1):
decoder = ip.CompanionDecoder()
decoder.feed(wire[:split])
first = decoder.pop()
decoder.feed(wire[split:])
self.assertEqual(first if first is not None else decoder.pop(), incoming())
self.assertIsNone(decoder.pop())
for wire in (b"<\x01\0x", b">\0\0", b">\xb1\0"):
decoder = ip.CompanionDecoder()
decoder.feed(wire)
with self.assertRaises(ValueError):
decoder.pop()
with self.assertRaises(BufferError):
ip.CompanionDecoder().feed(bytes(4097))
def test_actual176byte_rx_limit_does_not_increase_command_or_ip_budget(self):
buffers = started_buffers()
for size in (173, 176):
frame = bytes((0x88,)) + bytes(size - 1)
buffers.feed_companion(response(frame))
self.assertEqual(buffers.counters["ignored_frames"], 2)
self.assertTrue(buffers.can_read_tun())
self.assertEqual(codec().mtu, 156)
with self.assertRaises(ValueError): ip.companion_frame(bytes(173))
decoder = ip.CompanionDecoder()
decoder.feed(response(bytes(177)))
with self.assertRaises(ValueError): decoder.pop()
def test_serial_printable_banner_is_bounded_and_only_before_binary(self):
decoder = ip.CompanionDecoder(ascii_preamble=True)
banner = b"WELCOME HIL\r\n(+++MESHCORE-TERM-STOP returns to Binary mode)\r\n> "
wire = banner + response(bytes((5, 0)))
for byte in wire:
decoder.feed(bytes((byte,)))
result = decoder.pop()
self.assertEqual(result, bytes((5, 0)))
self.assertEqual(decoder.preamble_bytes, len(banner))
decoder.feed(b"bad")
with self.assertRaises(ValueError): decoder.pop()
for wire, exception in ((b"bad\x00", ValueError), (b"a" * 4096, None)):
decoder = ip.CompanionDecoder(ascii_preamble=True)
decoder.feed(wire)
if exception:
with self.assertRaises(exception): decoder.pop()
else:
decoder.pop()
decoder.feed(b"x")
with self.assertRaises(BufferError): decoder.pop()
def test_startup_and_one_acknowledged_command_at_a_time(self):
buffers = ip.IpRelayBuffers(codec())
self.assertFalse(buffers.can_read_tun())
self.assertEqual(buffers.output[:4], b"<\x13\0\x01")
buffers.write_companion(lambda data: len(data))
buffers.feed_companion(response(bytes((5, 0))))
self.assertTrue(buffers.can_read_tun())
self.assertTrue(buffers.queue_tun(packet()))
with self.assertRaises(BufferError):
buffers.queue_tun(packet())
buffers.write_companion(lambda data: 2)
self.assertTrue(buffers.output)
buffers.write_companion(lambda data: len(data))
buffers.feed_companion(response(b"\0"))
self.assertTrue(buffers.can_read_tun())
self.assertEqual(buffers.stats()["accepted_commands"], 1)
self.assertIn("not RF-delivered", buffers.stats()["acceptance_note"])
def test_partial_stream_continues_reading_and_receive_queue_backpressures(self):
buffers = started_buffers(depth=1)
wire = response(incoming())
buffers.feed_companion(wire[:1])
self.assertTrue(buffers.can_read_companion())
buffers.feed_companion(wire[1:] + wire)
self.assertEqual(len(buffers.to_tun), 1)
self.assertFalse(buffers.can_read_companion())
buffers.write_tun(lambda data: len(data))
buffers.parse_pending()
self.assertEqual(len(buffers.to_tun), 1)
self.assertFalse(buffers.decoder.staging)
with self.assertRaises(IOError):
buffers.write_tun(lambda data: 1)
def test_error_acceptance_invalid_packet_and_deadline_are_honest(self):
now = [0.]
buffers = started_buffers(now=lambda: now[0])
self.assertFalse(buffers.queue_tun(b"bad IPv4"))
buffers.queue_tun(packet())
buffers.write_companion(lambda data: len(data))
buffers.feed_companion(response(b"\x01\x03"))
self.assertEqual(buffers.stats()["rejected_commands"], 1)
buffers.queue_tun(packet())
now[0] = 12
with self.assertRaises(TimeoutError):
buffers.check_deadline()
def test_socket_before_namespace_before_new_tun_only(self):
events = []
stream = mock.Mock()
with mock.patch.object(ip, "CompanionStream", side_effect=lambda endpoint, **kwargs: events.append("socket") or stream), \
mock.patch.object(ip, "enter_network_namespace", side_effect=lambda path: events.append("namespace")), \
mock.patch.object(ip, "create_tun", side_effect=lambda name: events.append("tun") or (99, "mcip0")), \
mock.patch.object(ip.subprocess, "run") as run:
self.assertEqual(ip.open_transport_then_tun("tcp://192.168.1.84:5000", "/var/run/netns/test", "mcip0",
"10.90.5.1", "10.90.5.2", 156), (stream, 99, "mcip0"))
self.assertEqual(events, ["socket", "namespace", "tun"])
self.assertEqual(run.call_args_list[1].args[0], ["ip", "address", "add", "10.90.5.1/32", "peer", "10.90.5.2/32", "dev", "mcip0"])
stream.close.assert_not_called()
def test_namespace_rejection_closes_own_socket_without_touching_tun(self):
stream = mock.Mock()
with mock.patch.object(ip, "CompanionStream", return_value=stream), \
mock.patch.object(ip, "enter_network_namespace", side_effect=ValueError("same host")), \
mock.patch.object(ip, "create_tun") as create:
with self.assertRaises(ValueError):
ip.open_transport_then_tun("endpoint", "host", "mcip0", "10.90.5.1", "10.90.5.2", 156)
create.assert_not_called()
stream.close.assert_called_once()
class CompanionStreamTests(unittest.TestCase):
endpoint = "serial:///dev/serial/by-id/usb-RAK_USB_TEST_SERIAL-if00"
identity = ("/dev/ttyACM3", "USB_TEST_SERIAL", 11, 22, 33)
def test_windows_stream_shared_binary_transport_and_explicit_identity(self):
shared = mock.Mock(read_fd=77, write_fd=78)
payload = b"\0\xc0\xdb\xff"
shared.read.return_value = payload
shared.write.return_value = len(payload)
with mock.patch.object(ip, "WindowsUsbStream", return_value=shared) as opened:
stream = ip.CompanionStream("winusb://COM6", windows_python="/mnt/c/python.exe",
expect_serial="EXACT", expect_vid=0x303a, expect_pid=0x1001,
serial_dtr=True, duration=300)
self.assertEqual((stream.read_fd, stream.write_fd), (77, 78))
self.assertEqual(stream.read(99), payload)
self.assertEqual(stream.write(payload), len(payload))
stream.close()
opened.assert_called_once_with("winusb://COM6", python="/mnt/c/python.exe", expected_serial="EXACT",
expected_vid=0x303a, expected_pid=0x1001, baud=115200, serial_dtr=True, duration=300)
shared.write.assert_called_once_with(payload)
shared.close.assert_called_once()
with self.assertRaises(ValueError): ip.CompanionStream("winusb://COM6")
with self.assertRaises(ValueError):
ip.CompanionStream("tcp://192.168.1.87:5000", windows_python="/mnt/c/python.exe")
def test_relay_selects_separate_pipe_read_and_write_descriptors(self):
stream = mock.Mock(read_fd=77, write_fd=78)
stream.write.side_effect = len
stream.read.return_value = response(bytes((5, 0)))
selections = []
def selected(reads, writes, errors, timeout):
selections.append((reads, writes))
return ([], [78], []) if len(selections) == 1 else ([77], [], [])
with mock.patch.object(ip.select, "select", side_effect=selected):
stats = ip.relay(stream, 88, ip.IpRelayBuffers(codec()), duration=1, stop=lambda: len(selections) >= 2)
self.assertEqual(selections, [([77], [78]), ([77], [])])
self.assertTrue(stats["ready"])
def test_tcp_defaults_unchanged_and_no_serial_open(self):
sock = mock.Mock()
sock.recv.return_value = b">\x01\0\0"
sock.send.return_value = 2
with mock.patch.object(ip.socket, "create_connection", return_value=sock) as connect, \
mock.patch.object(ip, "serial_identity") as identity:
stream = ip.CompanionStream("tcp://192.168.1.87:5000")
self.assertEqual(stream.read(10), b">\x01\0\0")
self.assertEqual(stream.write(b"<\x01\0\x01"), 2)
stream.close()
connect.assert_called_once_with(("192.168.1.87", 5000), timeout=5)
identity.assert_not_called()
sock.setblocking.assert_called_once_with(False)
sock.close.assert_called_once()
def test_unstable_alias_wrong_protocol_and_credentials_rejected_without_open(self):
for endpoint in ("serial:///dev/ttyACM3", "serial:///dev/serial/by-path/port0",
"serial:///dev/serial/by-id/../../ttyACM3", "serial:///tmp/radio",
"tcp://192.168.1.87:5003", "tcp://user:secret@192.168.1.87:5000"):
with self.subTest(endpoint=endpoint), self.assertRaises(ValueError):
ip.validate_companion_endpoint(endpoint)
with self.assertRaises(ValueError):
ip.validate_companion_endpoint("tcp://192.168.1.87:5000", expect_serial="9AB3B64C641BA927")
def test_busy_serial_never_opens_or_changes_lines(self):
with mock.patch.object(ip, "serial_identity", return_value=self.identity), \
mock.patch.object(ip, "assert_port_idle", side_effect=RuntimeError("busy")), \
mock.patch.object(ip, "open_configured_session") as opened:
with self.assertRaisesRegex(RuntimeError, "busy"):
ip.CompanionStream(self.endpoint)
opened.assert_not_called()
def test_exact_identity_exclusive_safe_dtr_selection_read_write_and_close(self):
for selected in (False, True):
port = mock.Mock()
port.fileno.return_value = 77
fake_serial = SimpleNamespace(Serial=mock.Mock(return_value=port))
info = SimpleNamespace(st_mode=0o020600, st_dev=11, st_ino=22, st_rdev=33)
with mock.patch.dict(sys.modules, serial=fake_serial), \
mock.patch.object(ip, "serial_identity", return_value=self.identity) as identity, \
mock.patch.object(ip, "assert_port_idle") as idle, \
mock.patch.object(ip, "open_configured_session") as opened, \
mock.patch.object(ip.os, "fstat", return_value=info), \
mock.patch.object(ip.os, "read", return_value=b">\x02\0\x05\0") as read, \
mock.patch.object(ip.os, "write", return_value=2) as write:
stream = ip.CompanionStream(self.endpoint, baud=115200, serial_dtr=selected,
expect_serial="9AB3B64C641BA927")
self.assertEqual(stream.read(4096), b">\x02\0\x05\0")
self.assertEqual(stream.write(b"<\x01\0\x01"), 2)
stream.close()
self.assertEqual(identity.call_count, 2)
idle.assert_called_once_with("/dev/ttyACM3")
opened.assert_called_once_with(port)
self.assertEqual(port.port, "/dev/ttyACM3")
self.assertTrue(port.exclusive)
self.assertIs(port.dtr, selected)
self.assertIs(port.rts, False)
self.assertEqual((port.timeout, port.write_timeout), (0, 0))
read.assert_called_once_with(77, 4096)
write.assert_called_once_with(77, b"<\x01\0\x01")
port.close.assert_called_once()
def test_alias_change_after_open_closes_owned_port(self):
port = mock.Mock()
fake_serial = SimpleNamespace(Serial=mock.Mock(return_value=port))
changed = ("/dev/ttyACM4", *self.identity[1:])
with mock.patch.dict(sys.modules, serial=fake_serial), \
mock.patch.object(ip, "serial_identity", side_effect=[self.identity, changed]), \
mock.patch.object(ip, "assert_port_idle"), mock.patch.object(ip, "open_configured_session"), \
mock.patch.object(ip.os, "fstat"):
with self.assertRaisesRegex(ValueError, "identity changed"):
ip.CompanionStream(self.endpoint)
port.close.assert_called_once()
def test_wrong_usb_serial_and_alias_target_fail_closed(self):
with mock.patch.object(ip.Path, "resolve", return_value=Path("/tmp/not-a-radio")):
with self.assertRaisesRegex(ValueError, "ttyACM/ttyUSB"):
ip.serial_identity("/dev/serial/by-id/radio", "EXPECTED")
info = SimpleNamespace(st_mode=0o020600, st_dev=11, st_ino=22, st_rdev=33)
with mock.patch.object(ip.Path, "resolve", side_effect=["/dev/ttyACM3", Path("/sys/devices/usb/1-1")]), \
mock.patch.object(ip.os, "stat", return_value=info), \
mock.patch.object(ip.Path, "is_file", return_value=True), \
mock.patch.object(ip.Path, "read_text", return_value="WRONG"):
with self.assertRaisesRegex(ValueError, "USB serial differs"):
ip.serial_identity("/dev/serial/by-id/radio", "EXPECTED")
if __name__ == "__main__":
unittest.main()
+83
View File
@@ -0,0 +1,83 @@
"""RAW phone SSH/SFTP fixture contract; no root, services, sockets or radios."""
from pathlib import Path
import re
import subprocess
import sys
import unittest
ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "tools/hil/meshcore_ip_ssh_phone_bench.sh"
class MeshCoreIpPhoneFixtureTests(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.source = SCRIPT.read_text()
cls.validator = re.search(r"python3 - \"\$endpoint\" \"\$script_dir\" \"\$expect_serial\" <<'PY'\n(.*?)\nPY", cls.source, re.S)[1]
def test_endpoint_validator_only_explicit_private_numeric_companion_tcp5000(self):
for endpoint in ("tcp://192.168.1.87:5000", "tcp://10.1.2.3:5000", "tcp://172.20.1.2:5000/",
"serial:///dev/serial/by-id/usb-RAK_USB_TEST_SERIAL-if00"):
result = subprocess.run([sys.executable, "-B", "-c", self.validator, endpoint, str(ROOT / "tools/hil")],
capture_output=True, timeout=3, check=False)
self.assertEqual(result.returncode, 0, result.stderr.decode())
for endpoint in ("tcp://8.8.8.8:5000", "tcp://127.0.0.1:5000", "tcp://radio.local:5000",
"tcp://192.168.1.87:5003", "tcp://user:secret@192.168.1.87:5000",
"serial:///dev/ttyACM0", "tcp://192.168.1.87:5000/path",
"tcp://192.168.1.87:5000?x=1", "tcp://192.168.1.0:5000"):
result = subprocess.run([sys.executable, "-B", "-c", self.validator, endpoint, str(ROOT / "tools/hil")],
capture_output=True, timeout=3, check=False)
self.assertNotEqual(result.returncode, 0, endpoint)
def test_raw_point_ip_configuration_and_actual_handshake_readiness(self):
self.assertIn('--interface mcip0 --local-address 10.90.5.2 --remote-address 10.90.5.1', self.source)
self.assertIn("--path-hex '' --path-hash-width 1 --return-path-byte-count 0 --mtu 156", self.source)
self.assertIn('--ready-after-handshake', self.source)
self.assertIn("state.get('state') == 'ready'", self.source)
self.assertIn("state.get('mtu') == 156", self.source)
self.assertIn('Companion APP_START/SelfInfo handshake did not become ready.', self.source)
self.assertLess(self.source.index('((ready))'), self.source.index("printf 'READY:"))
self.assertNotIn('ethernet_tap_relay.py --endpoint', self.source)
def test_serial_identity_and_explicit_line_options_forwarded_without_kiss(self):
for required in ('--baud) baud=$2', '--serial-dtr) serial_dtr=$2', '--expect-serial) expect_serial=$2',
'serial_dtr=off', 'serial_options=(--baud "$baud" --serial-dtr "$serial_dtr")',
'serial_options+=(--expect-serial "$expect_serial")', '"${serial_options[@]}"'):
self.assertIn(required, self.source)
def test_account_private_keys_and_internal_sftp_not_system_ssh(self):
for required in ('--account) account=$2', '$(id -u "$account") -ne 0',
'ssh_user=${ssh_user:-$account}', '"$ssh_user" == "$account"',
'chown "$account": "$run_dir"', 'chmod 700 -- "$run_dir"',
'runuser -u "$account" -- ssh-keygen', '-o "AllowUsers=$ssh_user"',
"-o 'Subsystem=sftp internal-sftp'", '-o ListenAddress=10.90.5.2',
'-o AuthenticationMethods=publickey', '-o PasswordAuthentication=no',
'-o DisableForwarding=yes', 'options=(-f /dev/null'):
self.assertIn(required, self.source)
self.assertIn('umask 077', self.source)
self.assertNotIn('chown mikec', self.source)
self.assertNotIn('AllowUsers=mikec', self.source)
for forbidden in ('systemctl', 'service ssh', 'useradd', 'usermod', 'passwd ', '/etc/ssh/sshd_config',
'iptables', 'nft ', 'route add default', 'ip_forward'):
self.assertNotIn(forbidden, self.source)
def test_cleanup_is_owned_namespace_inode_and_explicit_children_only_with_key_removal(self):
for required in ('^mcphone-', '"$duration" -ge 60', '"$duration" -le 1800',
'Namespace already exists; refusing to reuse it.',
'mktemp -d -- "$bench_dir/meshcore-phone.XXXXXXXX"',
'trap cleanup EXIT', "trap 'exit 130' INT", "trap 'exit 143' TERM HUP",
'"$observed" == "$inode"', '"$parent" == "$$"',
'for child in "$relay_pid" "$sshd_pid"', 'for _ in {1..20}',
'rm -f -- "$run_dir/host_key" "$run_dir/client_key"'):
self.assertIn(required, self.source)
self.assertEqual(self.source.count('ip netns delete "$namespace"'), 1)
self.assertNotIn('pkill', self.source)
self.assertNotIn('killall', self.source)
self.assertNotIn('rm -rf', self.source)
self.assertNotIn('cat "$run_dir/client_key"', self.source)
self.assertNotIn('cat "$run_dir/host_key"', self.source)
self.assertIn('Server helper PID: %s', self.source)
if __name__ == "__main__":
unittest.main()
+44
View File
@@ -0,0 +1,44 @@
"""Non-hardware contract checks for the matched USB SFTP runner."""
from pathlib import Path
import shutil
import subprocess
import unittest
SCRIPT = Path(__file__).resolve().parents[1] / "tools/hil/meshcore_usb_sftp_bench.sh"
class UsbSftpBenchContract(unittest.TestCase):
def test_bash_syntax(self):
if not shutil.which("bash"):
self.skipTest("bash is unavailable")
# Parse stdin so Windows' WSL bash never interprets a C:\ path as a
# Linux filename (and no host-specific path conversion is required).
result = subprocess.run(["bash", "-n"], input=SCRIPT.read_text().encode("utf-8"),
capture_output=True, timeout=20)
self.assertEqual(result.returncode, 0, result.stderr)
def test_distinct_fixed_transports_and_real_mtu_check(self):
source = SCRIPT.read_text()
self.assertIn("mtu=156; interface=mcip0", source)
self.assertIn("mtu=230; interface=lora0", source)
self.assertIn("meshcore_ip_relay.py", source)
self.assertIn("ethernet_tap_relay.py", source)
self.assertIn("--ready-after-handshake", source)
self.assertIn("--kiss-port 1 --kiss-modem", source)
self.assertIn('[[ "$actual_mtu" == "$mtu" ]]', source)
self.assertIn('ip -n "$prefix" -j link show dev "$interface"', source)
def test_cleanup_scope_and_no_host_route_or_firewall(self):
source = SCRIPT.read_text()
self.assertIn('[[ -n "$inode" && "$observed" == "$inode" ]]', source)
self.assertIn('[[ "$parent" == "$$" ]]', source)
self.assertIn('ip netns delete "$prefix"', source)
self.assertNotIn("chown -R", source)
self.assertNotIn("iptables", source)
self.assertNotIn("nft ", source)
self.assertNotIn("route add default", source)
self.assertNotIn("rm -rf", source)
if __name__ == "__main__":
unittest.main()
+207
View File
@@ -0,0 +1,207 @@
"""Binary/identity/ownership tests; no USB hardware, namespaces or listeners."""
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import threading
import time
from types import SimpleNamespace
import unittest
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools/hil"))
import windows_usb_stdio_bridge as usb
SERIAL = "AABBCCDDEEFF"
IDENTITY = dict(port="COM6", serial=SERIAL, vid=0x303a, pid=0x1001)
def device_info(serial=SERIAL, port="COM6", vid=0x303a, pid=0x1001):
return SimpleNamespace(device=port, serial_number=serial, vid=vid, pid=pid)
class WindowsBridgeTests(unittest.TestCase):
def test_exact_endpoint_and_required_settings_fail_before_open(self):
self.assertEqual(usb.validate_windows_endpoint("winusb://com6").netloc, "com6")
for value in ("winusb://COM0", "winusb://COM6/", "winusb://COM6:2", "winusb://user@COM6",
"winusb://COM6?baud=1", "winusb://COM6#x", "winusb:///COM6", "tcp://COM6"):
with self.subTest(value=value), self.assertRaises(ValueError):
usb.validate_windows_endpoint(value)
for serial, vid, pid, duration in ((None, 1, 1, 5), ("bad\n", 1, 1, 5), (SERIAL, None, 1, 5),
(SERIAL, 1, 0, 5), (SERIAL, 1, 1, 0), (SERIAL, 1, 1, float("nan")), (SERIAL, 1, 1, 3601)):
with self.assertRaises(ValueError):
usb.validate_settings("COM6", serial, vid, pid, 115200, False, duration)
def test_wrong_missing_ambiguous_usb_identity_never_opens(self):
for inventory in ([], [device_info("OTHER")], [device_info(vid=1)], [device_info(pid=2)],
[device_info(port="COM7")], [device_info(), device_info()]):
factory = mock.Mock()
with self.assertRaises(ValueError):
usb.open_selected_port("COM6", SERIAL, 0x303a, 0x1001, baud=115200, serial_dtr=False,
inventory=lambda: inventory, serial_factory=factory)
factory.assert_not_called()
def test_safe_exclusive_open_checks_twice_and_selects_dtr_without_reset(self):
for dtr in (False, True):
device = mock.Mock()
inventory = mock.Mock(return_value=[device_info()])
with mock.patch.object(usb, "open_configured_session") as opened:
result, identity = usb.open_selected_port("com6", SERIAL, 0x303a, 0x1001,
baud=115200, serial_dtr=dtr, inventory=inventory, serial_factory=lambda: device)
self.assertIs(result, device)
self.assertEqual(identity, dict(IDENTITY, baud=115200, serial_dtr=dtr))
self.assertEqual(inventory.call_count, 2)
opened.assert_called_once_with(device)
self.assertEqual(device.port, "COM6")
self.assertIs(device.exclusive, True)
self.assertIs(device.dtr, dtr)
self.assertIs(device.rts, False)
self.assertEqual((device.timeout, device.write_timeout), (.05, .5))
device.close.assert_not_called()
def test_busy_or_swapped_identity_closes_only_owned_handle(self):
for swap in (False, True):
device = mock.Mock()
inventory = mock.Mock(side_effect=[[device_info()], [device_info("OTHER")]])
with mock.patch.object(usb, "open_configured_session", side_effect=None if swap else OSError("busy")):
with self.assertRaises((ValueError, OSError)):
usb.open_selected_port("COM6", SERIAL, 0x303a, 0x1001, baud=115200, serial_dtr=True,
inventory=inventory, serial_factory=lambda: device)
device.close.assert_called_once()
def test_duplex_short_writes_preserve_every_binary_byte(self):
payload = b"\0\xc0\xdb\xff\r\n" + bytes(range(256))
written, output = bytearray(), bytearray()
output_done = threading.Event()
supplied = [False]
returned = [False]
device = mock.Mock()
device.in_waiting = 0
def host_read(fd, count):
if not supplied[0]:
supplied[0] = True
return payload
output_done.wait(1)
return b""
def serial_read(count):
if not returned[0] and bytes(written) == payload:
returned[0] = True
return payload
time.sleep(.001)
return b""
def serial_write(data):
written.extend(data[:3])
return min(3, len(data))
def host_write(fd, data):
output.extend(data[:2])
if len(output) == len(payload):
output_done.set()
return min(2, len(data))
device.write.side_effect, device.read.side_effect = serial_write, serial_read
stats = usb.pump_stdio(device, 11, 12, duration=2, read=host_read, write=host_write)
self.assertEqual(bytes(written), payload)
self.assertEqual(bytes(output), payload)
self.assertEqual(stats, dict(to_usb=len(payload), from_usb=len(payload), reason="stdin_eof"))
device.close.assert_called_once()
def test_eof_and_deadline_close_usb_even_if_stdout_worker_is_blocked(self):
for eof in (False, True):
blocked, release = threading.Event(), threading.Event()
device = mock.Mock()
device.in_waiting = 4
device.read.return_value = b"\0\xc0\xdb\xff"
def read(fd, count):
blocked.wait(.2)
if not eof:
release.wait(1)
return b""
def write(fd, data):
blocked.set()
release.wait(1)
return len(data)
started = time.monotonic()
try:
stats = usb.pump_stdio(device, 11, 12, duration=.05, read=read, write=write)
self.assertLess(time.monotonic() - started, .6)
device.close.assert_called_once()
self.assertEqual(stats["reason"], "stdin_eof" if eof else "deadline")
finally:
release.set()
def test_parent_uses_absolute_trusted_argv_no_shell_and_distinct_nonblocking_fds(self):
with tempfile.TemporaryDirectory() as directory:
executable = Path(directory) / "python.exe"
executable.touch()
child = mock.Mock()
child.stdout.fileno.return_value, child.stdin.fileno.return_value = 11, 12
identity = dict(IDENTITY, baud=115200, serial_dtr=True)
with mock.patch.object(usb.sys, "platform", "linux"), \
mock.patch.object(usb.subprocess, "run", return_value=SimpleNamespace(stdout="C:\\git\\MeshCore\\tools\\hil\\windows_usb_stdio_bridge.py\n")), \
mock.patch.object(usb.subprocess, "Popen", return_value=child) as popen, \
mock.patch.object(usb, "_ready_record", return_value=dict(identity=identity)), \
mock.patch.object(usb.os, "set_blocking") as nonblocking:
stream = usb.WindowsUsbStream("winusb://COM6", python=str(executable), expected_serial=SERIAL,
expected_vid=0x303a, expected_pid=0x1001, serial_dtr=True, duration=300)
self.assertEqual((stream.read_fd, stream.write_fd), (11, 12))
self.assertEqual(stream.identity, identity)
argv = popen.call_args.args[0]
self.assertEqual(argv[:3], [str(executable), "-u", "C:\\git\\MeshCore\\tools\\hil\\windows_usb_stdio_bridge.py"])
self.assertIn(SERIAL, argv)
self.assertIs(popen.call_args.kwargs["shell"], False)
self.assertEqual(nonblocking.call_args_list, [mock.call(11, False), mock.call(12, False)])
stream.close()
child.stdin.close.assert_called_once()
child.wait.assert_called_once_with(timeout=2)
child.terminate.assert_not_called()
def test_parent_rejects_changed_ready_identity_and_closes_owned_child(self):
with tempfile.TemporaryDirectory() as directory:
executable = Path(directory) / "python.exe"
executable.touch()
child = mock.Mock()
with mock.patch.object(usb.sys, "platform", "linux"), \
mock.patch.object(usb.subprocess, "run", return_value=SimpleNamespace(stdout="C:\\trusted\\windows_usb_stdio_bridge.py")), \
mock.patch.object(usb.subprocess, "Popen", return_value=child), \
mock.patch.object(usb, "_ready_record", return_value=dict(identity=dict(IDENTITY, serial="OTHER"))):
with self.assertRaisesRegex(ValueError, "identity"):
usb.WindowsUsbStream("winusb://COM6", python=str(executable), expected_serial=SERIAL,
expected_vid=0x303a, expected_pid=0x1001)
child.stdin.close.assert_called_once()
child.wait.assert_called_once_with(timeout=2)
def test_parent_close_eof_then_bounded_owned_process_cleanup(self):
stream = usb.WindowsUsbStream.__new__(usb.WindowsUsbStream)
child = mock.Mock()
child.wait.side_effect = [subprocess.TimeoutExpired("owned", 2), subprocess.TimeoutExpired("owned", 1), 0]
stream.process = child
stream.close()
stream.close()
child.stdin.close.assert_called_once()
child.terminate.assert_called_once()
child.kill.assert_called_once()
self.assertEqual(child.wait.call_args_list, [mock.call(timeout=2), mock.call(timeout=1), mock.call(timeout=1)])
def test_ready_status_bound_and_invalid_records_do_not_expose_child_stderr(self):
for data in (b"secret payload, not JSON\n", b'{"state":"error","error":"private"}\n', b"x" * 4097):
child = mock.Mock()
child.stderr.fileno.return_value = 11
with mock.patch.object(usb.select, "select", return_value=([11], [], [])), \
mock.patch.object(usb.os, "read", return_value=data):
with self.assertRaises(RuntimeError) as failure:
usb._ready_record(child)
self.assertNotIn("secret", str(failure.exception))
self.assertNotIn("private", str(failure.exception))
if __name__ == "__main__":
unittest.main()