micromesh/tests/test_micromesh.py

import unittest

from micromesh import Data, FromRadio, MeshPacket, PortNum, Position, SerialInterface, ToRadio
from micromesh.protobuf import DecodeError, decode_varint, encode_varint
from micromesh.stream import StreamParser, frame


class FakeUART:
    def __init__(self, incoming=b""):
        self.incoming = bytearray(incoming)
        self.written = bytearray()

    def any(self):
        return len(self.incoming)

    def read(self, count=None):
        count = len(self.incoming) if count is None else count
        result = bytes(self.incoming[:count])
        del self.incoming[:count]
        return result

    def write(self, data):
        self.written.extend(data)
        return len(data)


class MicroMeshTests(unittest.TestCase):
    def test_varint_round_trip(self):
        for value in (0, 1, 127, 128, 16384, 0xFFFFFFFF, 0xFFFFFFFFFFFFFFFF):
            encoded = encode_varint(value)
            self.assertEqual(decode_varint(encoded), (value, len(encoded)))

    def test_known_data_wire_bytes(self):
        message = Data(portnum=PortNum.TEXT_MESSAGE_APP, payload=b"hello", want_response=True)
        self.assertEqual(message.SerializeToString(), b"\x08\x01\x12\x05hello\x18\x01")
        self.assertEqual(Data().ParseFromString(message.SerializeToString()).to_dict(), message.to_dict())

    def test_generated_style_imports(self):
        from micromesh import mesh_pb2, portnums_pb2
        self.assertEqual(mesh_pb2.Constants.DATA_PAYLOAD_LEN, 233)
        self.assertEqual(portnums_pb2.PortNum.TEXT_MESSAGE_APP, 1)

    def test_proto3_scalar_defaults_are_omitted(self):
        self.assertEqual(Data(portnum=0, payload=b"", want_response=False).SerializeToString(), b"")
        self.assertEqual(ToRadio(disconnect=False).SerializeToString(), b"\x20\x00")
        self.assertEqual(Position(latitude_i=0).SerializeToString(), b"\x0d\x00\x00\x00\x00")

    def test_mesh_packet_fixed_fields_and_nested_message(self):
        packet = MeshPacket(**{"from": 0x12345678, "to": 0xFFFFFFFF, "id": 42})
        packet.decoded.portnum = PortNum.TEXT_MESSAGE_APP
        packet.decoded.payload = b"hi"
        decoded = MeshPacket().ParseFromString(packet.SerializeToString())
        self.assertEqual(decoded.from_, 0x12345678)
        self.assertEqual(decoded.to, 0xFFFFFFFF)
        self.assertEqual(decoded.id, 42)
        self.assertEqual(decoded.decoded.payload, b"hi")

    def test_signed_position(self):
        original = Position(latitude_i=-1220000000, longitude_i=455000000, altitude=-12)
        decoded = Position().ParseFromString(original.SerializeToString())
        self.assertEqual(decoded.latitude_i, -1220000000)
        self.assertEqual(decoded.altitude, -12)

    def test_unknown_fields_are_preserved(self):
        raw = b"\x08\x01" + encode_varint(99 << 3) + b"\x07"
        parsed = Data().ParseFromString(raw)
        self.assertEqual(parsed.portnum, 1)
        self.assertEqual(parsed.SerializeToString(), raw)

    def test_stream_parser_handles_chunks_logs_and_resync(self):
        logs = []
        parser = StreamParser(on_log=logs.append)
        wrapped = b"debug line\n" + frame(b"one") + frame(b"two")
        packets = []
        for chunk in (wrapped[:3], wrapped[3:14], wrapped[14:19], wrapped[19:]):
            packets.extend(parser.feed(chunk))
        self.assertEqual(packets, [b"one", b"two"])
        self.assertEqual(logs, ["debug line"])

    def test_stream_parser_sanitizes_invalid_utf8_logs(self):
        logs = []
        parser = StreamParser(on_log=logs.append)
        self.assertEqual(parser.feed(b"bad:\xff\xfe\n"), [])
        self.assertEqual(logs, ["bad:??"])

    def test_send_text_builds_to_radio_frame(self):
        import micromesh.interface as interface_module
        old_packet_id = interface_module._packet_id
        interface_module._packet_id = lambda: 0x1234
        try:
            uart = FakeUART()
            interface = SerialInterface(uart)
            packet = interface.sendText("hello", destinationId="!aabbccdd", wantAck=True)
        finally:
            interface_module._packet_id = old_packet_id
        sent = ToRadio().ParseFromString(StreamParser().feed(uart.written)[0])
        self.assertEqual(sent.packet.to, 0xAABBCCDD)
        self.assertEqual(sent.packet.id, 0x1234)
        self.assertIs(sent.packet.want_ack, True)
        self.assertEqual(sent.packet.decoded.portnum, PortNum.TEXT_MESSAGE_APP)
        self.assertEqual(sent.packet.decoded.payload, b"hello")
        self.assertEqual(packet.id, 0x1234)

    def test_poll_decodes_and_tracks_state(self):
        config_id = 123
        uart = FakeUART(frame(FromRadio(config_complete_id=config_id).SerializeToString()))
        interface = SerialInterface(uart)
        interface.config_id = config_id
        self.assertEqual(len(interface.poll()), 1)
        self.assertIs(interface.config_complete, True)

    def test_poll_can_report_and_skip_bad_frames(self):
        errors = []
        bad = frame(b"\x12\x05no")
        good = frame(FromRadio(config_complete_id=123).SerializeToString())
        interface = SerialInterface(
            FakeUART(bad + good),
            on_error=lambda error, payload: errors.append((error, payload)),
        )
        interface.config_id = 123
        messages = interface.poll()
        self.assertEqual(len(messages), 1)
        self.assertEqual(len(errors), 1)
        self.assertEqual(errors[0][1], b"\x12\x05no")
        self.assertIs(interface.config_complete, True)

    def test_nested_decode_errors_include_the_field_path(self):
        with self.assertRaisesRegex(DecodeError, "FromRadio.packet"):
            FromRadio().ParseFromString(b"\x12\x03\x22\x05x")

    def test_truncated_message_raises(self):
        with self.assertRaises(DecodeError):
            Data().ParseFromString(b"\x12\x05no")


if __name__ == "__main__":
    unittest.main()