"""Synthetic protocol and ownership tests; never opens a real serial port.""" import base64 import binascii from datetime import datetime, timedelta, timezone import json from pathlib import Path import struct import sys import tempfile import unittest import uuid sys.path.insert(0, str(Path(__file__).resolve().parents[1])) sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "src")) from runtime import SCHEMA from runtime.protocol import Decoder, firmware, request, values from runtime.serial import Attachment from runtime.service import Service def frame(payload): header = bytes([2, len(payload)]) if len(payload) < 256 else b"\x03" + len(payload).to_bytes(2, "big") return header + payload + binascii.crc_hqx(payload, 0).to_bytes(2, "big") + b"\x03" def fw(index=1): return b"\x00\x06\x06SYNTHETIC\x00" + bytes([index]) * 12 + b"\x00\x00\x00\x00" def telemetry(): return b"\x04" + struct.pack(">hhiiiihihiiiiiiB", 215, 190, -123, 456, 0, 0, 123, 600, 481, 100, 0, 200, 0, 30, 30, 0) class ProtocolTests(unittest.TestCase): def test_transmit_has_no_motor_or_write_escape_hatch(self): for code in range(256): if code in (0, 4, 14, 17, 31, 62): self.assertEqual(Decoder().feed(request(code)), [bytes([code])]) else: with self.assertRaises(ValueError): request(code) for value in (True, b"\0", "0", -1, 256): with self.assertRaises(ValueError): request(value) def test_fragmented_and_combined_frames(self): samples = [fw(), bytes([14]) + bytes(range(256)) * 4, telemetry()] decoder = Decoder(); actual = [] for byte in b"noise" + b"".join(map(frame, samples)): actual += decoder.feed(bytes([byte])) self.assertEqual(actual, samples) def test_crc_and_bounds(self): bad = bytearray(frame(fw())); bad[-2] ^= 1 self.assertEqual(Decoder().feed(bad), []) self.assertEqual(Decoder().feed(b"\x04\xff\xff\xff" + frame(fw())), [fw()]) with self.assertRaises(ValueError): Decoder().feed(bytes(30000)) def test_identity_requires_uuid_and_bounded_name(self): self.assertEqual(firmware(fw())["uuid"], "01" * 12) for data in (b"", fw()[:5], fw()[:20], b"\0\6\6" + b"x" * 150, fw(0)): with self.assertRaises(ValueError): firmware(data) def test_telemetry_scales_signed_values_and_truncation(self): result = values(telemetry()) self.assertEqual(result["motor_current_a"], -1.23) self.assertEqual(result["input_voltage_v"], 48.1) self.assertEqual(result["erpm"], 600) self.assertNotIn("timeout", result) for length in range(54): with self.assertRaises(ValueError): values(telemetry()[:length]) class FakeLink: instances = [] identities = {} def __init__(self, attachment): self.attachment = attachment; self.commands = []; self.closed = False self.instances.append(self) def query(self, code, timeout=2): request(code) self.commands.append(code) if code == 0: return fw(self.identities.get(self.attachment.usb, int(self.attachment.address))) if code == 4: return telemetry() return bytes([code]) + b"\x12\x34\x56\x78opaque-config" def close(self): self.closed = True class ServiceTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.attachments = [Attachment("1-2", "1", "ttyACM0", "12"), Attachment("1-3", "2", "ttyACM1", "12")] FakeLink.instances = []; FakeLink.identities = {} self.service = Service(self.temp.name, lambda: self.attachments, FakeLink) self.service.scan() def tearDown(self): self.temp.cleanup() def command(self, item=None, action="vesc.telemetry.read"): item = item or self.service.inventory("node_synthetic")["items"][0] now = datetime.now(timezone.utc); identifier = "op_" + uuid.uuid4().hex return {"api_version": SCHEMA, "kind": "OperationRequest", "operation_id": identifier, "idempotency_key": identifier, "session": {"session_id": item["snapshot"]["context"]["session_id"], "device_id": item["id"]}, "requested_at": now.isoformat(), "deadline_at": (now + timedelta(seconds=60)).isoformat(), "action_id": action, "parameters": {}} def test_two_attachments_promote_independently(self): items = self.service.inventory("node_synthetic")["items"] self.assertEqual(len({i["id"] for i in items}), 2) self.assertTrue(all(i["id"] != i["attachment_id"] and i["verified"] for i in items)) result = self.service.execute(self.command(items[1])) self.assertEqual(result["state"], "complete") self.assertEqual(FakeLink.instances[0].commands, [0]) self.assertEqual(FakeLink.instances[1].commands, [0, 0, 4]) def test_uuid_collision_never_aliases_a_controller(self): FakeLink.identities = {"1-2": 1, "1-3": 1} for device in self.service.devices.values(): device.close() self.service.scan() items = self.service.inventory("node_synthetic")["items"] self.assertEqual(len({i["id"] for i in items}), 2) self.assertTrue(all(not i["verified"] for i in items)) with self.assertRaises(ValueError): self.service.execute(self.command(items[0])) def test_reconnect_invalidates_old_session_but_retains_uuid(self): old = self.service.inventory("node_synthetic")["items"][0]; command = self.command(old) self.attachments = self.attachments[1:]; self.service.scan() self.assertTrue(FakeLink.instances[0].closed) FakeLink.identities["1-2"] = 1 self.attachments.append(Attachment("1-2", "3", "ttyACM4", "12")); self.service.scan() current = next(i for i in self.service.inventory("node_synthetic")["items"] if i["id"] == old["id"]) self.assertNotEqual(current["snapshot"]["context"]["session_id"], old["snapshot"]["context"]["session_id"]) with self.assertRaises(ValueError): self.service.execute(command) def test_receipt_replay_never_repeats_serial_query(self): command = self.command(); first = self.service.execute(command) before = list(FakeLink.instances[0].commands) self.assertEqual(self.service.execute(command), first) self.assertEqual(FakeLink.instances[0].commands, before) command["action_id"] = "vesc.config.backup" with self.assertRaises(ValueError): self.service.execute(command) def test_native_process_exit_invalidates_session_before_reconnection(self): old = self.service.inventory("node_synthetic")["items"][0] command = self.command(old) process = FakeLink.instances[0] process.alive = False self.service.scan() current = next(i for i in self.service.inventory("node_synthetic")["items"] if i["id"] == old["id"]) self.assertTrue(process.closed) self.assertNotEqual(current["snapshot"]["context"]["session_id"], old["snapshot"]["context"]["session_id"]) with self.assertRaises(ValueError): self.service.execute(command) def test_backup_is_raw_hashed_and_bound_to_identity(self): result = self.service.execute(self.command(action="vesc.config.backup")) self.assertEqual(result["state"], "complete") value = result["result"] self.assertFalse(value["decoded"]) self.assertEqual(base64.b64decode(value["configs"]["motor"]["payload"])[0], 14) backups = list(Path(self.temp.name).glob("backup_*.json")) self.assertEqual(json.loads(backups[0].read_text()), value) self.assertEqual(backups[0].stat().st_mode & 0o777, 0o600) self.assertEqual(FakeLink.instances[0].commands, [0, 0, 14, 17, 0]) def test_arbitrary_writes_parameters_and_expired_requests_rejected(self): for action in ("start", "stop", "settings.apply", "raw", "firmware.write"): with self.assertRaises(ValueError): self.service.execute(self.command(action=action)) command = self.command(); command["parameters"] = {"packet": "anything"} with self.assertRaises(ValueError): self.service.execute(command) command = self.command(); command["deadline_at"] = command["requested_at"] with self.assertRaises(ValueError): self.service.execute(command) self.assertTrue(all(link.commands == [0] for link in FakeLink.instances)) if __name__ == "__main__": unittest.main()