init

tests/test_micromesh.py

@@ -0,0 +1,112 @@
+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_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_truncated_message_raises(self):
+        with self.assertRaises(DecodeError):
+            Data().ParseFromString(b"\x12\x05no")
+
+
+if __name__ == "__main__":
+    unittest.main()