test: comprehensive SniffSession live streaming tests

10 tests covering streaming flags, trace entry parsing, status transitions, button-toggle lifecycle, BREAK_LOOP, entry accumulation across pause/resume cycles, and full end-to-end workflow.
2026-03-19 12:34:10 -07:00
parent e90a907630
commit 83c8dd5546
1 changed files with 327 additions and 0 deletions
--- a/tests/test_sniff_session.py
+++ b/tests/test_sniff_session.py
@@ -0,0 +1,327 @@
+"""Tests for SniffSession live streaming mode."""
+import struct
+import time
+
+import pytest
+
+from pm3py.core.protocol import (
+    Cmd,
+    SNIFF_FLAG_STREAMING, SNIFF_FLAG_BUTTON_TOGGLE,
+    SNIFF_STATE_STARTED, SNIFF_STATE_PAUSED,
+    SNIFF_STATE_RESUMED, SNIFF_STATE_STOPPED,
+    RESP_PREAMBLE_MAGIC, RESP_POSTAMBLE_NOCRC,
+)
+from pm3py.core.transport import encode_ng_frame
+from pm3py.sniff.session import SniffSession
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+class FakeSerial:
+    """Mock serial port that feeds pre-built response frames."""
+
+    def __init__(self, frames: list[bytes] | None = None):
+        self._buf = bytearray()
+        if frames:
+            for f in frames:
+                self._buf.extend(f)
+        self._pos = 0
+        self._closed = False
+        self._written: list[bytes] = []
+
+    @property
+    def in_waiting(self):
+        return len(self._buf) - self._pos
+
+    def read(self, size=1):
+        if self._closed or self._pos >= len(self._buf):
+            time.sleep(0.01)
+            return b""
+        end = min(self._pos + size, len(self._buf))
+        data = bytes(self._buf[self._pos:end])
+        self._pos = end
+        return data
+
+    def write(self, data):
+        self._written.append(bytes(data))
+        return len(data)
+
+    def reset_input_buffer(self):
+        pass
+
+    def reset_output_buffer(self):
+        pass
+
+    def close(self):
+        self._closed = True
+
+    def feed(self, data: bytes):
+        """Append more data to the read buffer."""
+        self._buf.extend(data)
+
+
+def _make_resp_frame(cmd: int, status: int, data: bytes) -> bytes:
+    """Build a raw NG response frame (USB, no CRC)."""
+    payload_len = len(data)
+    preamble = struct.pack("<IHBBH",
+        RESP_PREAMBLE_MAGIC,
+        payload_len | 0x8000,  # NG bit
+        status & 0xFF,
+        0,   # reason
+        cmd,
+    )
+    postamble = struct.pack("<H", RESP_POSTAMBLE_NOCRC)
+    return preamble + data + postamble
+
+
+def _make_sniff_stream_frame(protocol, direction, duration, timestamp, data):
+    """Build a CMD_HF_SNIFF_STREAM response frame."""
+    payload = struct.pack("<BBHI", protocol, direction, duration, timestamp) + data
+    return _make_resp_frame(Cmd.HF_SNIFF_STREAM.value, 0, payload)
+
+
+def _make_status_frame(state, count=0):
+    """Build a CMD_HF_SNIFF_STATUS response frame."""
+    payload = struct.pack("<BH", state, count)
+    return _make_resp_frame(Cmd.HF_SNIFF_STATUS.value, 0, payload)
+
+
+# ---------------------------------------------------------------------------
+# Tests
+# ---------------------------------------------------------------------------
+
+class TestSniffSession:
+
+    def test_start_15693_sends_streaming_flags(self):
+        """start_15693(live=True) sends payload with idle_timeout_ms=400, flags=0x03."""
+        ser = FakeSerial()
+        session = SniffSession(ser)
+
+        # Feed a status frame so the reader thread has something to consume
+        ser.feed(_make_status_frame(SNIFF_STATE_STARTED))
+
+        session.start_15693(live=True, idle_timeout_ms=400)
+        time.sleep(0.05)
+        session._active = False
+        time.sleep(0.05)
+
+        # Check the written command frame
+        assert len(ser._written) >= 1
+        frame = ser._written[0]
+
+        # Decode the command frame: magic(4) + length_ng(2) + cmd(2) + payload + postamble(2)
+        magic, length_ng, cmd = struct.unpack_from("<IHH", frame, 0)
+        payload = frame[8:-2]
+
+        assert cmd == Cmd.HF_ISO15693_SNIFF
+        idle_timeout, flags = struct.unpack_from("<HB", payload)
+        assert idle_timeout == 400
+        assert flags == (SNIFF_FLAG_STREAMING | SNIFF_FLAG_BUTTON_TOGGLE)
+        assert flags == 0x03
+
+    def test_start_15693_legacy_sends_zero_flags(self):
+        """start_15693(live=False) sends flags=0."""
+        ser = FakeSerial()
+        session = SniffSession(ser)
+
+        # Feed a final response so legacy path unblocks
+        ser.feed(_make_resp_frame(Cmd.HF_ISO15693_SNIFF.value, 0, b""))
+
+        session.start_15693(live=False, idle_timeout_ms=400)
+
+        assert len(ser._written) >= 1
+        frame = ser._written[0]
+        payload = frame[8:-2]
+        idle_timeout, flags = struct.unpack_from("<HB", payload)
+        assert idle_timeout == 400
+        assert flags == 0
+
+    def test_receives_trace_entries(self):
+        """Feed CMD_HF_SNIFF_STREAM frames, verify parsed entries."""
+        frames = [
+            _make_status_frame(SNIFF_STATE_STARTED),
+            _make_sniff_stream_frame(
+                protocol=0, direction=0, duration=100,
+                timestamp=1000, data=bytes([0x26, 0x01, 0x00])),
+            _make_sniff_stream_frame(
+                protocol=0, direction=1, duration=200,
+                timestamp=2000, data=bytes([0x00, 0x0F, 0xAB, 0xCD])),
+        ]
+        ser = FakeSerial(frames)
+        session = SniffSession(ser)
+
+        session._formatter = None  # suppress printing
+        session._active = True
+        session._state = "sniffing"
+        session._reader_thread = __import__("threading").Thread(
+            target=session._stream_reader, daemon=True)
+        session._reader_thread.start()
+
+        time.sleep(0.15)
+        session._active = False
+        session._reader_thread.join(timeout=1.0)
+
+        entries = session.entries
+        assert len(entries) == 2
+
+        assert entries[0]["direction"] == 0
+        assert entries[0]["duration"] == 100
+        assert entries[0]["timestamp"] == 1000
+        assert entries[0]["data"] == bytes([0x26, 0x01, 0x00])
+
+        assert entries[1]["direction"] == 1
+        assert entries[1]["duration"] == 200
+        assert entries[1]["timestamp"] == 2000
+        assert entries[1]["data"] == bytes([0x00, 0x0F, 0xAB, 0xCD])
+
+    def test_status_updates(self):
+        """Feed STATUS frames, verify session.state transitions."""
+        frames = [
+            _make_status_frame(SNIFF_STATE_STARTED),
+            _make_status_frame(SNIFF_STATE_PAUSED, count=5),
+            _make_status_frame(SNIFF_STATE_RESUMED),
+            _make_status_frame(SNIFF_STATE_STOPPED, count=10),
+        ]
+        ser = FakeSerial(frames)
+        session = SniffSession(ser)
+
+        session._formatter = None
+        session._active = True
+        session._state = "starting"
+        session._reader_thread = __import__("threading").Thread(
+            target=session._stream_reader, daemon=True)
+        session._reader_thread.start()
+
+        time.sleep(0.15)
+        session._reader_thread.join(timeout=1.0)
+
+        # After STOPPED, _active should be False and state should be "stopped"
+        assert session._active is False
+        assert session.state == "stopped"
+
+    def test_stop_sends_break_loop(self):
+        """stop() sends a BREAK_LOOP command."""
+        ser = FakeSerial()
+        session = SniffSession(ser)
+        session._active = True
+        # No reader thread — just test the command sending
+        session._reader_thread = None
+
+        session.stop()
+
+        assert len(ser._written) == 1
+        frame = ser._written[0]
+        _, _, cmd = struct.unpack_from("<IHH", frame, 0)
+        assert cmd == Cmd.BREAK_LOOP
+        assert session._active is False
+        assert session.state == "stopped"
+
+    def test_entries_accumulate_across_cycles(self):
+        """Entries persist across pause/resume cycles."""
+        frames = [
+            _make_status_frame(SNIFF_STATE_STARTED),
+            _make_sniff_stream_frame(0, 0, 50, 100, b"\x26"),
+            _make_status_frame(SNIFF_STATE_PAUSED, count=1),
+            _make_status_frame(SNIFF_STATE_RESUMED),
+            _make_sniff_stream_frame(0, 1, 60, 200, b"\x00\x0F"),
+            _make_status_frame(SNIFF_STATE_STOPPED, count=2),
+        ]
+        ser = FakeSerial(frames)
+        session = SniffSession(ser)
+
+        session._formatter = None
+        session._active = True
+        session._state = "starting"
+        session._reader_thread = __import__("threading").Thread(
+            target=session._stream_reader, daemon=True)
+        session._reader_thread.start()
+
+        time.sleep(0.15)
+        session._reader_thread.join(timeout=1.0)
+
+        entries = session.entries
+        assert len(entries) == 2
+        assert entries[0]["data"] == b"\x26"
+        assert entries[1]["data"] == b"\x00\x0F"
+
+    def test_clear_entries(self):
+        """clear() resets entries list."""
+        ser = FakeSerial()
+        session = SniffSession(ser)
+        session._entries.append({"direction": 0, "data": b"\x26"})
+        session._entries.append({"direction": 1, "data": b"\x00"})
+
+        assert len(session.entries) == 2
+        session.clear()
+        assert len(session.entries) == 0
+
+    def test_full_workflow(self):
+        """Integration: started -> stream -> pause -> resume -> more -> stopped."""
+        frames = [
+            _make_status_frame(SNIFF_STATE_STARTED),
+            # Reader command
+            _make_sniff_stream_frame(0, 0, 100, 1000, b"\x26\x01\x00"),
+            # Tag response
+            _make_sniff_stream_frame(0, 1, 200, 1500, b"\x00\x0F\xAA\xBB"),
+            # Another reader command
+            _make_sniff_stream_frame(0, 0, 100, 3000, b"\x20\x01"),
+            _make_status_frame(SNIFF_STATE_PAUSED, count=3),
+            _make_status_frame(SNIFF_STATE_RESUMED),
+            # More frames after resume
+            _make_sniff_stream_frame(0, 0, 100, 5000, b"\x26\x01\x00"),
+            _make_sniff_stream_frame(0, 1, 200, 5500, b"\x00\x0F\xCC\xDD"),
+            _make_status_frame(SNIFF_STATE_STOPPED, count=5),
+        ]
+        ser = FakeSerial(frames)
+        session = SniffSession(ser)
+
+        session._formatter = None
+        session._active = True
+        session._state = "starting"
+        session._reader_thread = __import__("threading").Thread(
+            target=session._stream_reader, daemon=True)
+        session._reader_thread.start()
+
+        time.sleep(0.2)
+        session._reader_thread.join(timeout=1.0)
+
+        entries = session.entries
+        assert len(entries) == 5
+
+        # Verify timestamps are monotonically increasing
+        timestamps = [e["timestamp"] for e in entries]
+        assert timestamps == sorted(timestamps)
+        assert timestamps == [1000, 1500, 3000, 5000, 5500]
+
+        # Verify directions alternate for first pair
+        assert entries[0]["direction"] == 0  # reader
+        assert entries[1]["direction"] == 1  # tag
+
+        # Final state
+        assert session.state == "stopped"
+        assert session._active is False
+
+    def test_close_cleans_up(self):
+        """close() stops session and closes serial."""
+        ser = FakeSerial()
+        session = SniffSession(ser)
+        session._active = True
+        session._reader_thread = None
+
+        session.close()
+
+        assert session._active is False
+        assert session.state == "stopped"
+        assert ser._closed is True
+
+    def test_entries_returns_copy(self):
+        """entries property returns a copy, not the internal list."""
+        ser = FakeSerial()
+        session = SniffSession(ser)
+        session._entries.append({"direction": 0, "data": b"\x26"})
+
+        entries = session.entries
+        entries.append({"direction": 1, "data": b"\x00"})
+        assert len(session.entries) == 1  # internal list unchanged