micromesh/tests/test_micromesh.py

import unittest

from micromesh import (
    Data, DeviceMetrics, FromRadio, MeshPacket, PortNum, Position,
    SerialInterface, ToRadio, node_num,
)
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_varint_rejects_values_wider_than_64_bits(self):
        with self.assertRaises(ValueError):
            encode_varint(1 << 64)
        with self.assertRaises(DecodeError):
            decode_varint(b"\xff" * 9 + b"\x02")

    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")
        decoded = Data().ParseFromString(message.SerializeToString())
        self.assertEqual(decoded.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):
        empty = Data(portnum=0, payload=b"", want_response=False)
        self.assertEqual(empty.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_optional_device_metric_zero_is_preserved(self):
        metrics = DeviceMetrics(battery_level=0, voltage=0.0)
        decoded = DeviceMetrics().ParseFromString(metrics.SerializeToString())
        self.assertTrue(decoded.HasField("battery_level"))
        self.assertTrue(decoded.HasField("voltage"))

    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_current_signature_fields_round_trip(self):
        data = Data(xeddsa_signature=b"signature")
        packet = MeshPacket(decoded=data, xeddsa_signed=True)
        decoded = MeshPacket().ParseFromString(packet.SerializeToString())
        self.assertEqual(decoded.decoded.xeddsa_signature, b"signature")
        self.assertTrue(decoded.xeddsa_signed)

    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_send_position_keeps_zero_altitude_and_validates_coordinates(self):
        uart = FakeUART()
        packet = SerialInterface(uart).sendPosition(0, 0, altitude=0)
        self.assertTrue(Position().ParseFromString(packet.decoded.payload).HasField("altitude"))
        with self.assertRaises(ValueError):
            SerialInterface(FakeUART()).sendPosition(91, 0)

    def test_node_numbers_must_fit_fixed32(self):
        self.assertEqual(node_num("!ffffffff"), 0xFFFFFFFF)
        with self.assertRaises(ValueError):
            node_num(-1)
        with self.assertRaises(ValueError):
            node_num(1 << 32)

    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_known_field_with_wrong_wire_type_is_preserved_as_unknown(self):
        raw = b"\x0a\x01x"
        parsed = Data().ParseFromString(raw)
        self.assertFalse(parsed.HasField("portnum"))
        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")

    def test_field_number_zero_raises(self):
        with self.assertRaises(DecodeError):
            Data().ParseFromString(b"\x00\x00")


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