fd16fa8f90
Live sniff/sim stream trace frames from a background thread. bpython replaces sys.stdout with a capture object whose write() does a greenlet switch back to the main thread — which raises off-thread; the text buffers but only repaints on the next keypress, so streamed frames 'pile up until you touch a key'. New trace/output.py emit() buffers via the repl's on_write and fires bpython's threadsafe repaint trigger; a plain write everywhere else, with a safe fallback on any internals mismatch. Route TraceFormatter.print(), SimSession's sim-ended status, and SniffSession's state message through emit(). Flip the trace-line newline convention from leading to trailing so each emitted line self-terminates (format_sniff_line + test updated to match). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
582 lines
23 KiB
Python
582 lines
23 KiB
Python
"""Tests for pm3py.sim.trace_fmt — trace formatting and protocol decoding."""
|
|
import os
|
|
import pytest
|
|
|
|
from pm3py.trace.decode_iso15 import decode_15693, decode_15693_nxp
|
|
from pm3py.trace.decode_iso14a import decode_14443a
|
|
from pm3py.trace.format import TraceFormatter
|
|
|
|
|
|
class TestDecode15693Request:
|
|
"""Decode reader->tag (direction=0) 15693 commands."""
|
|
|
|
def test_inventory(self):
|
|
payload = bytes([0x26, 0x01, 0x00])
|
|
result = decode_15693(0, payload)
|
|
assert result == "INVENTORY"
|
|
|
|
def test_inventory_with_mask(self):
|
|
payload = bytes([0x26, 0x01, 0x08, 0xAB])
|
|
result = decode_15693(0, payload)
|
|
assert result == "INVENTORY mask=8"
|
|
|
|
def test_stay_quiet(self):
|
|
payload = bytes([0x22, 0x02]) + b"\x01" * 8
|
|
result = decode_15693(0, payload)
|
|
assert result == "STAY QUIET"
|
|
|
|
def test_read_single_block_addressed(self):
|
|
payload = bytes([0x22, 0x20]) + b"\x01" * 8 + bytes([0x03])
|
|
result = decode_15693(0, payload)
|
|
assert result == "READ SINGLE BLOCK #3"
|
|
|
|
def test_read_single_block_unaddressed(self):
|
|
payload = bytes([0x02, 0x20, 0x00])
|
|
result = decode_15693(0, payload)
|
|
assert result == "READ SINGLE BLOCK #0"
|
|
|
|
def test_write_single_block(self):
|
|
payload = bytes([0x22, 0x21]) + b"\x01" * 8 + bytes([0x05]) + b"\xDE\xAD\xBE\xEF"
|
|
result = decode_15693(0, payload)
|
|
assert result == "WRITE SINGLE BLOCK #5 [4B]"
|
|
|
|
def test_read_multiple_block(self):
|
|
payload = bytes([0x22, 0x23]) + b"\x01" * 8 + bytes([0x00, 0x0D])
|
|
result = decode_15693(0, payload)
|
|
assert result == "READ MULTIPLE BLOCKS #0+13"
|
|
|
|
def test_reset_to_ready(self):
|
|
payload = bytes([0x22, 0x26]) + b"\x01" * 8
|
|
result = decode_15693(0, payload)
|
|
assert result == "RESET TO READY"
|
|
|
|
def test_get_system_info(self):
|
|
payload = bytes([0x22, 0x2B]) + b"\x01" * 8
|
|
result = decode_15693(0, payload)
|
|
assert result == "GET SYSTEM INFO"
|
|
|
|
def test_get_multiple_block_security(self):
|
|
payload = bytes([0x22, 0x2C]) + b"\x01" * 8 + bytes([0x00, 0x3F])
|
|
result = decode_15693(0, payload)
|
|
assert result == "GET MULTIPLE BLOCK SECURITY #0+63"
|
|
|
|
def test_nxp_command_decoded(self):
|
|
payload = bytes([0x22, 0xA1, 0x04, 0x02])
|
|
result = decode_15693(0, payload)
|
|
assert result == "NXP FAST INVENTORY READ"
|
|
|
|
def test_unknown_command(self):
|
|
payload = bytes([0x22, 0xFF])
|
|
result = decode_15693(0, payload)
|
|
assert result == "UNKNOWN CMD 0xFF"
|
|
|
|
def test_too_short(self):
|
|
result = decode_15693(0, bytes([0x26]))
|
|
assert result is None
|
|
|
|
|
|
class TestDecode15693NxpRequest:
|
|
"""Decode NXP custom reader->tag (direction=0) commands."""
|
|
|
|
def test_nxp_fast_inventory_read(self):
|
|
payload = bytes([0x22, 0xA1, 0x04, 0x02])
|
|
result = decode_15693_nxp(0, payload)
|
|
assert result == "NXP FAST INVENTORY READ"
|
|
|
|
def test_nxp_set_eas(self):
|
|
payload = bytes([0x22, 0xA2, 0x04, 0x03, 0xFF])
|
|
result = decode_15693_nxp(0, payload)
|
|
assert result == "NXP SET EAS"
|
|
|
|
def test_nxp_get_random(self):
|
|
payload = bytes([0x22, 0xB2, 0x04])
|
|
result = decode_15693_nxp(0, payload)
|
|
assert result == "NXP GET RANDOM"
|
|
|
|
def test_nxp_set_password(self):
|
|
payload = bytes([0x22, 0xB3, 0x04, 0x01]) + b"\x00" * 4
|
|
result = decode_15693_nxp(0, payload)
|
|
assert result == "NXP SET PASSWORD id=1"
|
|
|
|
def test_response_returns_none(self):
|
|
payload = bytes([0x00])
|
|
result = decode_15693_nxp(1, payload)
|
|
assert result is None
|
|
|
|
def test_unknown_nxp_command(self):
|
|
payload = bytes([0x22, 0xC0, 0x04])
|
|
result = decode_15693_nxp(0, payload)
|
|
assert result is None
|
|
|
|
|
|
class TestDecode15693Response:
|
|
"""Decode tag->reader (direction=1) 15693 responses."""
|
|
|
|
def test_ok_empty(self):
|
|
result = decode_15693(1, bytes([0x00]))
|
|
assert result == "OK"
|
|
|
|
def test_inventory_response(self):
|
|
uid_lsb = bytes([0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x04, 0xE0])
|
|
payload = bytes([0x00, 0x00]) + uid_lsb
|
|
result = decode_15693(1, payload)
|
|
assert result == "OK INVENTORY UID=E004010101010101"
|
|
|
|
def test_error_response(self):
|
|
payload = bytes([0x01, 0x0F])
|
|
result = decode_15693(1, payload)
|
|
assert result == "ERROR 0x0F"
|
|
|
|
def test_error_block_not_available(self):
|
|
payload = bytes([0x01, 0x10])
|
|
result = decode_15693(1, payload)
|
|
assert result == "ERROR 0x10 block not available"
|
|
|
|
def test_data_response(self):
|
|
payload = bytes([0x00, 0xDE, 0xAD, 0xBE, 0xEF])
|
|
result = decode_15693(1, payload)
|
|
assert result == "OK [4B]"
|
|
|
|
def test_too_short(self):
|
|
result = decode_15693(1, b"")
|
|
assert result is None
|
|
|
|
|
|
class TestDecode14443aRequest:
|
|
"""Decode reader->tag 14443-A commands."""
|
|
|
|
def test_reqa(self):
|
|
assert decode_14443a(0, bytes([0x26])) == "REQA"
|
|
|
|
def test_wupa(self):
|
|
assert decode_14443a(0, bytes([0x52])) == "WUPA"
|
|
|
|
def test_hlta(self):
|
|
assert decode_14443a(0, bytes([0x50, 0x00])) == "HLTA"
|
|
|
|
def test_anticol_cl1(self):
|
|
assert decode_14443a(0, bytes([0x93, 0x20])) == "ANTICOL CL1"
|
|
|
|
def test_select_cl1(self):
|
|
payload = bytes([0x93, 0x70]) + b"\x01\x02\x03\x04\x04"
|
|
assert decode_14443a(0, payload) == "SELECT CL1"
|
|
|
|
def test_anticol_cl2(self):
|
|
assert decode_14443a(0, bytes([0x95, 0x20])) == "ANTICOL CL2"
|
|
|
|
def test_select_cl2(self):
|
|
payload = bytes([0x95, 0x70]) + b"\x01\x02\x03\x04\x04"
|
|
assert decode_14443a(0, payload) == "SELECT CL2"
|
|
|
|
def test_anticol_cl3(self):
|
|
assert decode_14443a(0, bytes([0x97, 0x20])) == "ANTICOL CL3"
|
|
|
|
def test_select_cl3(self):
|
|
payload = bytes([0x97, 0x70]) + b"\x01\x02\x03\x04\x04"
|
|
assert decode_14443a(0, payload) == "SELECT CL3"
|
|
|
|
def test_rats(self):
|
|
assert decode_14443a(0, bytes([0xE0, 0x50])) == "RATS"
|
|
|
|
def test_iblock_even(self):
|
|
assert decode_14443a(0, bytes([0x02, 0x00, 0xA4])) == "I-BLOCK(0)"
|
|
|
|
def test_iblock_odd(self):
|
|
assert decode_14443a(0, bytes([0x03, 0x00, 0xA4])) == "I-BLOCK(1)"
|
|
|
|
def test_rack(self):
|
|
assert decode_14443a(0, bytes([0xA2])) == "R-ACK(0)"
|
|
assert decode_14443a(0, bytes([0xA3])) == "R-ACK(1)"
|
|
|
|
def test_rnak(self):
|
|
assert decode_14443a(0, bytes([0xB2])) == "R-NAK(0)"
|
|
assert decode_14443a(0, bytes([0xB3])) == "R-NAK(1)"
|
|
|
|
def test_deselect(self):
|
|
assert decode_14443a(0, bytes([0xC2])) == "S(DESELECT)"
|
|
|
|
def test_wtx(self):
|
|
assert decode_14443a(0, bytes([0xF2, 0x01])) == "S(WTX)"
|
|
|
|
def test_unknown(self):
|
|
assert decode_14443a(0, bytes([0xFF])) is None
|
|
|
|
def test_empty(self):
|
|
assert decode_14443a(0, b"") is None
|
|
|
|
|
|
class TestDecode14443aResponse:
|
|
"""Decode tag->reader 14443-A responses."""
|
|
|
|
def test_atqa(self):
|
|
assert decode_14443a(1, bytes([0x04, 0x00])) == "ATQA 04 00"
|
|
|
|
def test_sak(self):
|
|
assert decode_14443a(1, bytes([0x20])) == "SAK 20"
|
|
|
|
def test_ats(self):
|
|
ats = bytes([0x05, 0x78, 0x80, 0x70, 0x02])
|
|
assert decode_14443a(1, ats) == "ATS [5]"
|
|
|
|
def test_iblock_response(self):
|
|
assert decode_14443a(1, bytes([0x02, 0x90, 0x00])) == "I-BLOCK(0)"
|
|
|
|
def test_empty(self):
|
|
assert decode_14443a(1, b"") is None
|
|
|
|
|
|
class TestTraceFormatterBasic:
|
|
"""Test formatting output (colors stripped for assertion)."""
|
|
|
|
def _strip_ansi(self, s: str) -> str:
|
|
import re
|
|
return re.sub(r'\033\[[0-9;]*m', '', s)
|
|
|
|
def test_ends_with_newline(self):
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693)
|
|
result = fmt.format(0, bytes([0x26, 0x01, 0x00]))
|
|
assert result.endswith("\n")
|
|
|
|
def test_sim_mode_tag(self):
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693)
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
assert "[Sim]" in result
|
|
|
|
def test_reader_mode_tag(self):
|
|
fmt = TraceFormatter(mode="reader", decoder=decode_15693)
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
assert "[Rdr]" in result
|
|
|
|
def test_sniff_mode_tag(self):
|
|
fmt = TraceFormatter(mode="sniff", decoder=decode_15693)
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
assert "[Snf]" in result
|
|
|
|
def test_reader_to_tag_arrow(self):
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693)
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
assert "Reader \u2192 Tag:" in result
|
|
|
|
def test_tag_to_reader_arrow(self):
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693)
|
|
result = self._strip_ansi(fmt.format(1, bytes([0x00])))
|
|
assert "Tag \u2192 Reader:" in result
|
|
|
|
def test_hex_space_separated(self):
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693)
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x22, 0x20, 0x03])))
|
|
assert "22 20 03" in result
|
|
|
|
def test_hex_uppercase(self):
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693)
|
|
result = self._strip_ansi(fmt.format(0, bytes([0xDE, 0xAD])))
|
|
assert "DE AD" in result
|
|
|
|
def test_annotation_present(self):
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693)
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
assert "INVENTORY" in result
|
|
|
|
def test_no_annotation_for_unknown(self):
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693)
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x22, 0xFF])))
|
|
assert "22 FF" in result
|
|
|
|
def test_no_decoder(self):
|
|
fmt = TraceFormatter(mode="sim")
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
assert "26 01 00" in result
|
|
assert "INVENTORY" not in result
|
|
|
|
def test_custom_decoder(self):
|
|
def my_decoder(d, p):
|
|
return "CUSTOM"
|
|
fmt = TraceFormatter(mode="sim", decoder=my_decoder)
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x00])))
|
|
assert "CUSTOM" in result
|
|
|
|
|
|
class TestTraceFormatterDimOurSide:
|
|
"""Our side (the one we're simulating) should be dimmed."""
|
|
|
|
def test_sim_mode_dims_tag_response(self):
|
|
"""In sim mode, tag→reader (our side) should have dim annotation."""
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, is_tty=True)
|
|
result = fmt.format(1, bytes([0x00]))
|
|
# dim (\033[2m) should appear before direction color for annotation
|
|
assert "\033[2m" in result
|
|
|
|
def test_sim_mode_bright_reader_command(self):
|
|
"""In sim mode, reader→tag (their side) should NOT be dimmed."""
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, is_tty=True)
|
|
result = fmt.format(0, bytes([0x26, 0x01, 0x00]))
|
|
# Annotation should use direction color (cyan), not dim
|
|
assert "\033[36mINVENTORY\033[0m" in result
|
|
|
|
def test_reader_mode_dims_reader_command(self):
|
|
"""In reader mode, reader→tag (our side) should be dimmed."""
|
|
fmt = TraceFormatter(mode="reader", decoder=decode_15693, is_tty=True)
|
|
result = fmt.format(0, bytes([0x26, 0x01, 0x00]))
|
|
assert "\033[2m" in result
|
|
|
|
def test_sniff_mode_dims_neither(self):
|
|
"""In sniff mode, neither side is dimmed — annotations use direction colors."""
|
|
fmt = TraceFormatter(mode="sniff", decoder=decode_15693, is_tty=True)
|
|
r0 = fmt.format(0, bytes([0x26, 0x01, 0x00]))
|
|
r1 = fmt.format(1, bytes([0x00]))
|
|
# reader→tag uses cyan, tag→reader uses yellow
|
|
assert "\033[36mINVENTORY\033[0m" in r0
|
|
assert "\033[33mOK\033[0m" in r1
|
|
|
|
|
|
class TestTraceFormatterWrapping:
|
|
"""Test column-aligned wrapping for long payloads."""
|
|
|
|
def _strip_ansi(self, s: str) -> str:
|
|
import re
|
|
return re.sub(r'\033\[[0-9;]*m', '', s)
|
|
|
|
def test_short_payload_inline_annotation(self):
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, width=80)
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
lines = result.strip().split("\n")
|
|
assert len(lines) == 1
|
|
assert "INVENTORY" in lines[0]
|
|
assert "26 01 00" in lines[0]
|
|
|
|
def test_annotation_right_justified(self):
|
|
"""Annotation should end at the terminal's right edge."""
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, width=80)
|
|
result = self._strip_ansi(fmt.format(0, bytes([0x26, 0x01, 0x00])))
|
|
line = result.strip()
|
|
assert line.endswith("INVENTORY")
|
|
assert len(line) == 80
|
|
|
|
def test_wrapped_annotation_right_justified(self):
|
|
"""When wrapped, annotation line should be right-justified."""
|
|
# Use wider terminal so annotation fits within avail
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, width=80)
|
|
# Short response that still wraps at width=80 won't work, so test
|
|
# with a response that has a short annotation
|
|
payload = bytes([0x00, 0xDE, 0xAD, 0xBE, 0xEF])
|
|
# At width=80 this fits on one line — annotation right-justified
|
|
result = self._strip_ansi(fmt.format(1, payload))
|
|
line = result.strip()
|
|
assert line.endswith("OK [4B]")
|
|
assert len(line) == 80
|
|
|
|
def test_long_payload_wraps(self):
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, width=60)
|
|
long_payload = bytes([0x00]) + bytes(40)
|
|
result = self._strip_ansi(fmt.format(1, long_payload))
|
|
lines = result.strip().split("\n")
|
|
assert len(lines) > 1
|
|
# Continuation lines should be indented to prefix column
|
|
first_hex_col = lines[0].index("00")
|
|
for line in lines[1:]:
|
|
if line.strip():
|
|
leading = len(line) - len(line.lstrip())
|
|
assert leading >= first_hex_col - 1
|
|
|
|
def test_wrapped_annotation_on_own_line(self):
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, width=50)
|
|
uid_lsb = bytes([0x01, 0x01, 0x01, 0x01, 0x01, 0x01, 0x04, 0xE0])
|
|
payload = bytes([0x00, 0x00]) + uid_lsb
|
|
result = self._strip_ansi(fmt.format(1, payload))
|
|
lines = result.strip().split("\n")
|
|
if len(lines) > 1:
|
|
annotation_lines = [l for l in lines if "INVENTORY" in l]
|
|
hex_lines = [l for l in lines if "00 01" in l.lower() or "04 e0" in l.lower()]
|
|
if len(hex_lines) > 1:
|
|
assert len(annotation_lines) == 1
|
|
assert annotation_lines[0] not in hex_lines
|
|
|
|
def test_long_annotation_wraps_to_next_line(self):
|
|
cc = bytes([0xE1, 0x40, 0x0D, 0x01])
|
|
from pm3py.transponders.hf.iso15693.type5 import ndef_text
|
|
msg = ndef_text("A" * 30)
|
|
tlv = bytes([0x03, len(msg)]) + msg + bytes([0xFE])
|
|
payload = bytes([0x00]) + cc + tlv
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, width=80)
|
|
result = self._strip_ansi(fmt.format(1, payload))
|
|
lines = result.strip().split("\n")
|
|
ann_lines = [l for l in lines if "NDEF" in l]
|
|
assert len(ann_lines) >= 1
|
|
|
|
|
|
class TestTraceFormatterCRC:
|
|
"""CRC bytes rendered separately and excluded from decoding."""
|
|
|
|
def _strip_ansi(self, s: str) -> str:
|
|
import re
|
|
return re.sub(r'\033\[[0-9;]*m', '', s)
|
|
|
|
def test_crc_bytes_in_output(self):
|
|
"""CRC bytes still appear in hex output."""
|
|
# Inventory response with CRC: flags + dsfid + uid(8) + crc(2)
|
|
uid_lsb = bytes([0x04, 0x03, 0x02, 0x01, 0xEF, 0xBE, 0xAD, 0xDE])
|
|
payload = bytes([0x00, 0x00]) + uid_lsb + bytes([0x84, 0x62])
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, crc_len=2, width=120)
|
|
result = self._strip_ansi(fmt.format(1, payload))
|
|
assert "84 62" in result
|
|
|
|
def test_crc_excluded_from_decode(self):
|
|
"""Decoder sees data without CRC — inventory response recognized."""
|
|
uid_lsb = bytes([0x04, 0x03, 0x02, 0x01, 0xEF, 0xBE, 0xAD, 0xDE])
|
|
payload = bytes([0x00, 0x00]) + uid_lsb + bytes([0x84, 0x62])
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, crc_len=2, width=120)
|
|
result = self._strip_ansi(fmt.format(1, payload))
|
|
assert "OK INVENTORY UID=" in result
|
|
|
|
def test_read_response_with_crc(self):
|
|
"""Read response with CRC decoded as OK [4B] not OK [6B]."""
|
|
payload = bytes([0x00, 0xDE, 0xAD, 0xBE, 0xEF, 0x48, 0xD1])
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, crc_len=2, width=120)
|
|
result = self._strip_ansi(fmt.format(1, payload))
|
|
assert "OK [4B]" in result
|
|
|
|
def test_crc_different_color(self):
|
|
"""CRC bytes use distinct color from data bytes."""
|
|
payload = bytes([0x00, 0xDE, 0xAD, 0x48, 0xD1])
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, crc_len=2, is_tty=True, width=120)
|
|
result = fmt.format(1, payload)
|
|
# Data uses dim, CRC uses bold white
|
|
assert "\033[2m" in result # dim for data
|
|
assert "\033[1;37m" in result # bold white for CRC
|
|
|
|
def test_no_crc_when_crc_len_zero(self):
|
|
"""No CRC splitting when crc_len=0."""
|
|
payload = bytes([0x00, 0xDE, 0xAD, 0xBE, 0xEF])
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, crc_len=0, width=120)
|
|
result = self._strip_ansi(fmt.format(1, payload))
|
|
assert "OK [4B]" in result
|
|
|
|
def test_short_payload_no_crc_split(self):
|
|
"""Payload shorter than crc_len doesn't crash."""
|
|
payload = bytes([0x00])
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, crc_len=2, width=120)
|
|
result = self._strip_ansi(fmt.format(1, payload))
|
|
assert "00" in result
|
|
|
|
def test_crc_in_commands(self):
|
|
"""Reader→tag commands include CRC — all bytes appear in hex output."""
|
|
# INVENTORY with CRC: 26 01 00 F6 0A — CRC stripped for decoding
|
|
payload = bytes([0x26, 0x01, 0x00, 0xF6, 0x0A])
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, crc_len=2, width=120)
|
|
result = self._strip_ansi(fmt.format(0, payload))
|
|
assert "INVENTORY" in result
|
|
assert "26 01 00" in result
|
|
assert "F6 0A" in result
|
|
|
|
|
|
class TestTraceFormatterNoColor:
|
|
"""Colors suppressed when is_tty=False."""
|
|
|
|
def test_no_ansi_when_not_tty(self):
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, is_tty=False)
|
|
result = fmt.format(0, bytes([0x26, 0x01, 0x00]))
|
|
assert "\033[" not in result
|
|
|
|
def test_ansi_present_when_tty(self):
|
|
fmt = TraceFormatter(mode="sim", decoder=decode_15693, is_tty=True)
|
|
result = fmt.format(0, bytes([0x26, 0x01, 0x00]))
|
|
assert "\033[" in result
|
|
|
|
|
|
class TestTransponderDecodeTrace:
|
|
"""Verify decode_trace on transponder hierarchy."""
|
|
|
|
def test_base_transponder_returns_none(self):
|
|
from pm3py.sim.transponder import Transponder
|
|
# Can't instantiate ABC directly, but we can check the method exists
|
|
assert hasattr(Transponder, 'decode_trace')
|
|
|
|
def test_tag15693_decodes_standard(self):
|
|
from pm3py.transponders.hf.iso15693.base import Tag15693
|
|
tag = Tag15693(uid=bytes(8))
|
|
result = tag.decode_trace(0, bytes([0x26, 0x01, 0x00]))
|
|
assert result == "INVENTORY"
|
|
|
|
def test_tag15693_decodes_nxp_names(self):
|
|
"""Base Tag15693 decodes NXP command names (via standard decoder)."""
|
|
from pm3py.transponders.hf.iso15693.base import Tag15693
|
|
tag = Tag15693(uid=bytes(8))
|
|
result = tag.decode_trace(0, bytes([0x22, 0xA1, 0x04, 0x02]))
|
|
assert result == "NXP FAST INVENTORY READ"
|
|
|
|
def test_nxp_icode_decodes_nxp(self):
|
|
from pm3py.transponders.hf.iso15693.nxp.nxp_icode import NxpIcodeTag
|
|
tag = NxpIcodeTag(uid=b"\xE0\x04" + bytes(6))
|
|
result = tag.decode_trace(0, bytes([0x22, 0xA1, 0x04, 0x02]))
|
|
assert result == "NXP FAST INVENTORY READ"
|
|
|
|
def test_nxp_icode_falls_back_to_standard(self):
|
|
from pm3py.transponders.hf.iso15693.nxp.nxp_icode import NxpIcodeTag
|
|
tag = NxpIcodeTag(uid=b"\xE0\x04" + bytes(6))
|
|
result = tag.decode_trace(0, bytes([0x26, 0x01, 0x00]))
|
|
assert result == "INVENTORY"
|
|
|
|
def test_icode_slix2_inherits_nxp(self):
|
|
from pm3py.transponders.hf.iso15693.nxp.icode_slix2 import IcodeSlix2Tag
|
|
tag = IcodeSlix2Tag(uid=b"\xE0\x04\x02" + bytes(5))
|
|
result = tag.decode_trace(0, bytes([0x22, 0xB2, 0x04]))
|
|
assert result == "NXP GET RANDOM"
|
|
|
|
def test_tag14443a_decodes(self):
|
|
from pm3py.transponders.hf.iso14443a.base import Tag14443A
|
|
tag = Tag14443A(uid=bytes(4))
|
|
result = tag.decode_trace(0, bytes([0x26]))
|
|
assert result == "REQA"
|
|
|
|
def test_tag14443a_no_15693(self):
|
|
from pm3py.transponders.hf.iso14443a.base import Tag14443A
|
|
tag = Tag14443A(uid=bytes(4))
|
|
result = tag.decode_trace(0, bytes([0x26, 0x01, 0x00]))
|
|
# 14443a decoder sees 0x26 as REQA (single byte check happens first)
|
|
# This is fine — the protocol context determines which decoder runs
|
|
assert result is not None # it will decode as REQA since b0=0x26
|
|
|
|
|
|
from unittest.mock import MagicMock, patch
|
|
import struct
|
|
|
|
|
|
class TestSimSessionTraceWiring:
|
|
"""Verify SimSession uses TraceFormatter with tag.decode_trace."""
|
|
|
|
def _make_trace_frame(self, direction: int, payload: bytes) -> bytes:
|
|
from pm3py.core.transport import RESP_POSTAMBLE_NOCRC
|
|
from pm3py.core.protocol import RESP_PREAMBLE_MAGIC
|
|
from pm3py.sim.sim_session import CMD_HF_ISO15693_SIM_TRACE
|
|
data = bytes([direction]) + payload
|
|
length = len(data) | 0x8000
|
|
header = struct.pack("<IHbbH", RESP_PREAMBLE_MAGIC, length, 0, 0, CMD_HF_ISO15693_SIM_TRACE)
|
|
postamble = struct.pack("<H", RESP_POSTAMBLE_NOCRC)
|
|
return header + data + postamble
|
|
|
|
@patch("pm3py.sim.sim_session.TraceFormatter")
|
|
def test_trace_uses_tag_decoder(self, MockFormatter):
|
|
from pm3py.sim.sim_session import SimSession
|
|
mock_fmt = MockFormatter.return_value
|
|
session = SimSession()
|
|
session._formatter = mock_fmt
|
|
session._active = True
|
|
session._print_frames = True # start_15693(trace=True) sets this
|
|
|
|
frame = self._make_trace_frame(0, bytes([0x26, 0x01, 0x00]))
|
|
mock_serial = MagicMock()
|
|
mock_serial.in_waiting = len(frame)
|
|
call_count = 0
|
|
|
|
def read_side_effect(n):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count == 1:
|
|
return frame
|
|
session._active = False
|
|
return b""
|
|
|
|
mock_serial.read.side_effect = read_side_effect
|
|
session._trace_reader(mock_serial)
|
|
mock_fmt.print.assert_called_once_with(0, bytes([0x26, 0x01, 0x00]), crc_fail=False)
|