import json import os import threading import time import unittest from types import SimpleNamespace from unittest.mock import patch import serial from motian_bridge import Bridge, Config, decode_command, make_report from motian_bridge.bridge import decode_hex def config(**changes): values = dict(system_id="UAV000001", uav_type=7, serial_port="loop://", serial_baudrate=57600, mqtt_host="localhost") values.update(changes) return Config(**values) def command(data_type=526, data="FE0600FF", **changes): value = dict(DataType=data_type, UAVType=7, Data=data, SystemID="UAV000001", TimeStamp="2026-09-28T12:00:00+08:00") value.update(changes) return json.dumps(value).encode() class FakeMqtt: def __init__(self): self.published = [] self.subscriptions = [] self.active = False def subscribe(self, topic, qos): self.subscriptions.append((topic, qos)) return (0, 1) def publish(self, topic, payload, qos, retain): self.published.append((topic, payload, qos, retain)) return SimpleNamespace(rc=0) def connect_async(self, *args): self.active = True def loop_start(self): pass def disconnect(self): self.active = False def loop_stop(self): pass class FakeSerial: def __init__(self): self.written = [] self.closed = False self.to_read = [] def write(self, data): self.written.append(data) return len(data) def read(self, size): time.sleep(0.01) return self.to_read.pop(0) if self.to_read else b"" def close(self): self.closed = True class BridgeTests(unittest.TestCase): def setUp(self): self.cfg = config() self.mqtt = FakeMqtt() self.serial = FakeSerial() self.bridge = Bridge(self.cfg, self.mqtt, lambda *args, **kwargs: self.serial) self.bridge._serial = self.serial self.bridge._on_connect(self.mqtt, None, None, 0, None) def deliver(self, payload, topic=None, retain=False): message = SimpleNamespace(topic=topic or self.cfg.obtain_topic, payload=payload, retain=retain) self.bridge._on_message(self.mqtt, None, message) def test_passthrough_bytes_and_exact_topic(self): self.deliver(command(data="00Feff01")) self.assertEqual(self.serial.written, [bytes.fromhex("00FEFF01")]) self.assertEqual(self.mqtt.subscriptions, [(self.cfg.obtain_topic, 0)]) def test_invalid_messages_never_write(self): bad = [b"not-json", b"[]", command(DataType=True), command(DataType=2049, SystemID="other"), command(UAVType=8), command(data="F"), command(data="FG"), command(data=""), command(data="AA BB"), command(data="AA" * 4097)] for payload in bad: self.deliver(payload) self.deliver(command(), retain=True) self.deliver(command(), topic="/OBTAIN/other") self.assertEqual(self.serial.written, []) def test_no_replay_on_disconnect_and_no_unsupported_action(self): self.bridge._on_disconnect(self.mqtt, None, None, 1, None) self.deliver(command()) self.assertEqual(self.serial.written, []) self.assertFalse(self.bridge.publish_report(526, "AA", passthrough=True)) self.bridge._on_connect(self.mqtt, None, None, 0, None) self.deliver(command(2054, None)) self.assertEqual(self.serial.written, []) self.deliver(command()) self.assertEqual(self.serial.written, [bytes.fromhex("FE0600FF")]) def test_extension_handler_and_report(self): seen = [] def handler(message, bridge): seen.append(message.data) bridge.publish_report(270, {"Text": "ok", "Cmd": message.data_type, "Code": 0}) self.bridge.register_handler(2054, handler) self.deliver(command(2054, None)) self.assertEqual(seen, [None]) self.assertEqual(self.mqtt.published[0][0], self.cfg.receive_topic) self.assertEqual(self.mqtt.published[0][2:], (0, False)) def test_serial_chunk_upstream_envelope(self): self.serial.to_read.append(b"\xfe\x00\xff") thread = threading.Thread(target=self.bridge._serial_loop) thread.start() try: deadline = time.monotonic() + 1 while not self.mqtt.published and time.monotonic() < deadline: time.sleep(0.01) self.assertEqual(len(self.mqtt.published), 1) topic, payload, qos, retain = self.mqtt.published[0] self.assertEqual((topic, qos, retain), (self.cfg.publish_topic, 0, False)) value = json.loads(payload) self.assertEqual(value["DataType"], 526) self.assertEqual(value["Data"], "FE00FF") self.assertEqual(value["UAVType"], 7) self.assertEqual(value["SystemID"], "UAV000001") self.assertIn("+", value["TimeStamp"]) self.assertNotIn("DataType ", value) finally: self.bridge._stop.set() thread.join(1) class ValidationTests(unittest.TestCase): def test_envelope_and_hex(self): decoded = decode_command(command(), config()) self.assertEqual(decoded.data_type, 526) self.assertEqual(decode_hex(decoded.data, 8), bytes.fromhex("FE0600FF")) self.assertEqual(json.loads(make_report(config(), 526, "AB"))["Data"], "AB") def test_config_validation(self): for changes in ({"system_id": "x/#"}, {"uav_type": -1}, {"max_chunk_bytes": 0}, {"mqtt_password_env": "PASS"}): with self.subTest(changes=changes), self.assertRaises(ValueError): config(**changes) def test_password_is_read_from_environment(self): cfg = config(mqtt_username="user", mqtt_password_env="BRIDGE_TEST_SECRET") with patch.dict(os.environ, {"BRIDGE_TEST_SECRET": "secret"}): self.assertEqual(cfg.mqtt_password(), "secret") with patch.dict(os.environ, {}, clear=True), self.assertRaises(ValueError): cfg.mqtt_password() def test_real_pyserial_loopback(self): cfg = config() mqtt = FakeMqtt() bridge = Bridge(cfg, mqtt, serial.serial_for_url) bridge.start() bridge._on_connect(mqtt, None, None, 0, None) try: deadline = time.monotonic() + 2 while bridge._serial is None and time.monotonic() < deadline: time.sleep(0.01) self.assertIsNotNone(bridge._serial) bridge._on_message(mqtt, None, SimpleNamespace( topic=cfg.obtain_topic, payload=command(data="00FEFF"), retain=False)) while not mqtt.published and time.monotonic() < deadline: time.sleep(0.01) self.assertEqual(len(mqtt.published), 1) self.assertEqual(json.loads(mqtt.published[0][1])["Data"], "00FEFF") finally: bridge.stop() def test_serial_reconnect_discards_old_bytes(self): class BrokenSerial(FakeSerial): def read(self, size): raise OSError("link lost") first, second = BrokenSerial(), FakeSerial() second.to_read.append(b"\x02") opened = [] def factory(*args, **kwargs): opened.append(True) return first if len(opened) == 1 else second cfg = config(reconnect_seconds=0.01) mqtt = FakeMqtt() bridge = Bridge(cfg, mqtt, factory) bridge.start() bridge._on_connect(mqtt, None, None, 0, None) try: deadline = time.monotonic() + 1 while not mqtt.published and time.monotonic() < deadline: time.sleep(0.01) self.assertGreaterEqual(len(opened), 2) self.assertTrue(first.closed) self.assertEqual(json.loads(mqtt.published[0][1])["Data"], "02") finally: bridge.stop() if __name__ == "__main__": unittest.main()