chore: Add better tests for serial data.

Try different combinations of serial data and SMP data. Test error handling.

Author: JaagupAverin

tests/test_smp_serial_transport.py

@@ -10,6 +10,7 @@
 from smp import packet as smppacket
 
 from smpclient.requests.os_management import EchoWrite
+from smpclient.transport import SMPTransportDisconnected
 from smpclient.transport.serial import SMPSerialTransport
 
 
@@ -129,57 +130,188 @@ async def test_read_one_smp_packet() -> None:
     for p in packets:
         assert p == await t._read_one_smp_packet()
 
+    await t.disconnect()
+
 
 @pytest.mark.asyncio
-async def test_read_one_smp_packet_with_smp_server_logging(
-    caplog: pytest.LogCaptureFixture,
-) -> None:
+async def test_send_and_receive() -> None:
+    t = SMPSerialTransport()
+    t.send = AsyncMock()  # type: ignore
+    t.receive = AsyncMock()  # type: ignore
+
+    await t.send_and_receive(b"some data")
+
+    t.send.assert_awaited_once_with(b"some data")
+    t.receive.assert_awaited_once_with()
+
+
+@pytest.mark.asyncio
+async def test_only_serial_data_no_smp(caplog: pytest.LogCaptureFixture) -> None:
+    t = SMPSerialTransport()
+    await t.connect("/dev/ttyACM0", timeout_s=1.0)
+
+    t._conn.read_all = MagicMock(  # type: ignore
+        side_effect=[
+            b"First line\n",
+            b"Second line\nThird line\n",
+            b"Partial line",
+            b" continues here\n",
+            b"No newline at end",
+        ]
+    )
+
+    with caplog.at_level(logging.WARNING):
+        await asyncio.sleep(0.05)  # Flush mock serial.
+        messages = {r.message for r in caplog.records}
+        assert "/dev/ttyACM0: First line" in messages
+        assert "/dev/ttyACM0: Second line" in messages
+        assert "/dev/ttyACM0: Third line" in messages
+        assert "/dev/ttyACM0: Partial line continues here" in messages
+        assert "/dev/ttyACM0: No newline at end" not in messages  # NOT!
+
+        await t.disconnect()
+        messages = {r.message for r in caplog.records}
+        assert "/dev/ttyACM0: No newline at end" in messages
+
+
+@pytest.mark.asyncio
+async def test_only_smp_data_no_serial() -> None:
     t = SMPSerialTransport()
     await t.connect("/dev/ttyUSB0", timeout_s=1.0)
 
-    m1 = EchoWrite._Response.get_default()(sequence=0, r="Hello pytest!")  # type: ignore
-    m2 = EchoWrite._Response.get_default()(sequence=1, r="Hello computer!")  # type: ignore
-    p1 = [p for p in smppacket.encode(m1.BYTES, 8)]
-    p2 = [p for p in smppacket.encode(m2.BYTES, 8)]
-    packets = p1 + p2
+    m1 = EchoWrite._Response.get_default()(sequence=0, r="SMP Message 1")  # type: ignore
+    m2 = EchoWrite._Response.get_default()(sequence=1, r="SMP Message 2")  # type: ignore
+    m3 = EchoWrite._Response.get_default()(sequence=2, r="SMP Message 3")  # type: ignore
+
+    packets = (
+        list(smppacket.encode(m1.BYTES, 512))
+        + list(smppacket.encode(m2.BYTES, 512))
+        + list(smppacket.encode(m3.BYTES, 512))
+    )
+
+    t._conn.read_all = MagicMock(side_effect=packets)  # type: ignore
+
+    for expected_msg in [m1.BYTES, m2.BYTES, m3.BYTES]:
+        received = await t.receive()
+        assert received == expected_msg
+
+    await t.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_serial_and_smp(caplog: pytest.LogCaptureFixture) -> None:
+    t = SMPSerialTransport()
+    await t.connect("/dev/ttyUSB0", timeout_s=1.0)
+
+    m1 = EchoWrite._Response.get_default()(sequence=0, r="SMP1")  # type: ignore
+    m2 = EchoWrite._Response.get_default()(sequence=1, r="SMP2")  # type: ignore
+
+    p1 = next(smppacket.encode(m1.BYTES, 512))
+    p2 = next(smppacket.encode(m2.BYTES, 512))
 
     t._conn.read_all = MagicMock(  # type: ignore
-        side_effect=(
-            [b"Hi, there!"]
-            + [b"newline\n"]
-            + [b"Another line\nAgain\n"]
-            + [b"log with no newline,"]
-            + p1
-            + [b" ends here\n"]
-            + [bytes([0, 1, 2, 3])]
-            + [b"Bye!\n"]
-            + p2
-            + [b"One more thing...\n"]
-            + [b"We \n could \n use \n newlines\n"]
-        )
+        side_effect=[
+            b"Start\n" + p1[:10],
+            p1[10:] + b"Mid1\n",
+            b"Mid2\n" + p2,
+            b"End\n",
+        ]
     )
 
     with caplog.at_level(logging.WARNING):
-        for p in packets:
-            assert p == await t._read_one_smp_packet()
+        received1 = await t.receive()
+        assert received1 == m1.BYTES
 
-        await t.disconnect()  # Flush remaining serial data
+        received2 = await t.receive()
+        assert received2 == m2.BYTES
 
+        await asyncio.sleep(0.05)  # Flush mock serial.
         messages = {r.message for r in caplog.records}
-        assert "/dev/ttyUSB0: Hi, there!newline" in messages
-        assert "/dev/ttyUSB0: Another line" in messages
-        assert "/dev/ttyUSB0: Again" in messages
-        assert "/dev/ttyUSB0: log with no newline, ends here" in messages
-        assert "/dev/ttyUSB0: \x00\x01\x02\x03Bye!" in messages
+        assert "/dev/ttyUSB0: Start" in messages
+        assert "/dev/ttyUSB0: Mid1" in messages
+        assert "/dev/ttyUSB0: Mid2" in messages
+        assert "/dev/ttyUSB0: End" in messages
+
+    await t.disconnect()
 
 
 @pytest.mark.asyncio
-async def test_send_and_receive() -> None:
+async def test_disconnect_flushes_partial_smp_as_serial(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
     t = SMPSerialTransport()
-    t.send = AsyncMock()  # type: ignore
-    t.receive = AsyncMock()  # type: ignore
+    await t.connect("/dev/ttyACM0", timeout_s=1.0)
 
-    await t.send_and_receive(b"some data")
+    partial_smp = smppacket.START_DELIMITER + b"AAAA"
 
-    t.send.assert_awaited_once_with(b"some data")
-    t.receive.assert_awaited_once_with()
+    t._conn.read_all = MagicMock(  # type: ignore
+        side_effect=[
+            b"Normal serial\n",
+            partial_smp,
+        ]
+    )
+
+    with caplog.at_level(logging.WARNING):
+        await asyncio.sleep(0.05)  # Flush mock serial.
+        await t.disconnect()
+        messages = {r.message for r in caplog.records}
+        assert "/dev/ttyACM0: Normal serial" in messages
+        assert any(
+            b"AAAA" in r.message.encode("utf-8", errors="ignore")
+            for r in caplog.records
+        )
+
+
+@pytest.mark.asyncio
+async def test_not_connected_exception_handling() -> None:
+    t = SMPSerialTransport()
+
+    with pytest.raises(SMPTransportDisconnected):
+        await t.receive()
+
+
+@pytest.mark.asyncio
+async def test_serial_callback_exception_handling(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    async def bad_callback(data: bytes) -> None:
+        raise ValueError("Callback error")
+
+    t = SMPSerialTransport(serial_line_callback=bad_callback)
+    await t.connect("/dev/ttyACM0", timeout_s=1.0)
+
+    t._conn.read_all = MagicMock(  # type: ignore
+        side_effect=[b"Line 1\n", b"Line 2\n"]
+    )
+
+    with caplog.at_level(logging.ERROR):
+        await asyncio.sleep(0.05)  # Flush mock serial.
+        assert any(
+            "Serial data callback raised an exception" in r.message
+            for r in caplog.records
+        )
+
+    await t.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_zero_length_serial_lines(caplog: pytest.LogCaptureFixture) -> None:
+    t = SMPSerialTransport()
+    await t.connect("/dev/ttyUSB0", timeout_s=1.0)
+
+    t._conn.read_all = MagicMock(  # type: ignore
+        side_effect=[
+            b"\n",
+            b"\n\n\n",
+            b"Text\n\n",
+            b"\nText\n",
+            b"",
+        ]
+    )
+
+    with caplog.at_level(logging.WARNING):
+        await asyncio.sleep(0.05)  # Flush mock serial.
+        empty_messages = [r for r in caplog.records if r.message == "/dev/ttyUSB0: "]
+        assert len(empty_messages) >= 4
+
+    await t.disconnect()