motian_mavbridge/tests/test_bridge.py

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"},
                        {"request_params_on_start": 1}, {"param_request_timeout": 0},
                        {"param_request_max_attempts": 0}):
            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()