83c8dd5546
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.
328 lines
11 KiB
Python
328 lines
11 KiB
Python
"""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
|