f76663f798
- completer.py: RawCompleter completes the leading command token — control verbs + the identified transponder's catalog commands — with each command's help as the completion meta (tooltip). `help <cmd>` completes catalog command names. - app: wire the completer with complete_while_typing (live menu) + AutoSuggestFromHistory (inline hints) for the ipython feel. Completes the 6-phase rawcli plan. 4 new tests (verb/catalog completion, tooltips, help-arg completion). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
329 lines
12 KiB
Python
329 lines
12 KiB
Python
"""rawcli Phase 1 — session state, breadcrumb, trace rendering, CLI dispatch. Hardware-free."""
|
|
import re
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from pm3py.cli.main import build_parser, main
|
|
from pm3py.cli.rawcli.session import RawSession
|
|
from pm3py.cli.rawcli.trace_view import render_exchange
|
|
from pm3py.cli.rawcli.parser import parse_token, parse_bytes, parse_line
|
|
from pm3py.cli.rawcli.entry import byte_space
|
|
from pm3py.cli.rawcli.identify import identify, clear, name_14a
|
|
from pm3py.cli.rawcli.catalog import TYPE2, ISO15, catalog_for, catalog_for as _cf
|
|
from pm3py.cli.rawcli.tlv import parse_tlv, format_tlv
|
|
from pm3py.cli.rawcli.completer import RawCompleter
|
|
from pm3py.cli.rawcli.app import dispatch
|
|
|
|
_ANSI = re.compile(r"\x1b\[[0-9;]*m")
|
|
|
|
|
|
def _plain(s: str) -> str:
|
|
return _ANSI.sub("", s)
|
|
|
|
|
|
class TestBreadcrumb:
|
|
def test_default(self):
|
|
assert RawSession().breadcrumb() == "[raw / 0x]"
|
|
|
|
def test_toggle_to_binary(self):
|
|
s = RawSession()
|
|
assert s.toggle_entry_mode() == "bin"
|
|
assert s.entry_prefix == "0b"
|
|
assert s.breadcrumb() == "[raw / 0b]"
|
|
assert s.toggle_entry_mode() == "hex"
|
|
assert s.breadcrumb() == "[raw / 0x]"
|
|
|
|
def test_identified_fills_segments(self):
|
|
s = RawSession()
|
|
s.field = "hf"
|
|
s.transponder = "NTAG215"
|
|
assert s.breadcrumb() == "[raw / hf / NTAG215 / 0x]"
|
|
|
|
def test_field_only(self):
|
|
s = RawSession()
|
|
s.field = "lf"
|
|
assert s.breadcrumb() == "[raw / lf / 0x]"
|
|
|
|
|
|
class TestTraceView:
|
|
def test_command_and_response_lines(self):
|
|
out = _plain(render_exchange(bytes([0x30, 0x04]),
|
|
bytes([0x00, 0x01, 0x02, 0x03]),
|
|
protocol="hf14a", is_tty=False))
|
|
assert "Reader → Tag" in out
|
|
assert "Tag → Reader" in out
|
|
assert "30 04" in out # command bytes shown
|
|
|
|
def test_command_only_when_no_response(self):
|
|
out = _plain(render_exchange(bytes([0x26]), None, protocol="hf14a", is_tty=False))
|
|
assert "Reader → Tag" in out
|
|
assert "Tag → Reader" not in out
|
|
|
|
def test_annotation_present_for_known_command(self):
|
|
# 0x30 = 14a READ; the decoder should annotate it (exact text may vary)
|
|
out = _plain(render_exchange(bytes([0x30, 0x04]), None, protocol="hf14a", is_tty=False))
|
|
assert "READ" in out.upper()
|
|
|
|
|
|
class TestCliDispatch:
|
|
def test_rawcli_subcommand_parsed(self):
|
|
args = build_parser().parse_args(["rawcli", "--port", "/dev/ttyACM0"])
|
|
assert args.command == "rawcli" and args.port == "/dev/ttyACM0"
|
|
|
|
def test_no_command_prints_help(self, capsys):
|
|
rc = main([])
|
|
assert rc == 1
|
|
assert "rawcli" in capsys.readouterr().out
|
|
|
|
|
|
class TestParser:
|
|
def test_hex_tokens(self):
|
|
assert parse_token("30") == b"\x30"
|
|
assert parse_token("0x3004") == b"\x30\x04"
|
|
assert parse_bytes("30 04") == b"\x30\x04"
|
|
assert parse_bytes("3004") == b"\x30\x04" # default hex mode
|
|
|
|
def test_binary_tokens(self):
|
|
assert parse_token("0b00110000") == b"\x30"
|
|
assert parse_token("00110000", "bin") == b"\x30"
|
|
|
|
def test_intermixed(self):
|
|
assert parse_bytes("0b00110000 0x04") == b"\x30\x04"
|
|
assert parse_bytes("0x30 00000100", "bin") == b"\x30\x04" # bare token uses mode
|
|
|
|
def test_bad_tokens(self):
|
|
with pytest.raises(ValueError):
|
|
parse_token("303") # odd hex
|
|
with pytest.raises(ValueError):
|
|
parse_token("0b0011") # not a whole byte
|
|
|
|
def test_classify_control(self):
|
|
c = parse_line("identify")
|
|
assert c.kind == "control" and c.name == "identify"
|
|
assert parse_line("help ntag215").kind == "control"
|
|
|
|
def test_classify_call(self):
|
|
c = parse_line("READ(4)")
|
|
assert c.kind == "call" and c.name == "READ" and c.args == ["4"]
|
|
c = parse_line("WRITE(4, 0x00112233)")
|
|
assert c.args == ["4", "0x00112233"]
|
|
assert parse_line("GET_DATA()").kind == "call"
|
|
|
|
def test_classify_raw(self):
|
|
c = parse_line("30 04")
|
|
assert c.kind == "raw" and c.payload == b"\x30\x04"
|
|
|
|
|
|
class TestByteSpace:
|
|
def test_hex_grouping(self):
|
|
assert byte_space("3004", "hex") == "30 04"
|
|
assert byte_space("300", "hex") == "30 0"
|
|
assert byte_space("30 04", "hex") == "30 04" # idempotent
|
|
|
|
def test_binary_grouping(self):
|
|
assert byte_space("0011000000000100", "bin") == "00110000 00000100"
|
|
|
|
def test_prefixed_left_alone(self):
|
|
assert byte_space("0x3004", "hex") == "0x3004" # explicit prefix not mangled
|
|
|
|
|
|
class TestDispatch:
|
|
def test_quit_returns_false(self):
|
|
assert dispatch(RawSession(), "quit") is False
|
|
|
|
def test_help_prints(self, capsys):
|
|
assert dispatch(RawSession(), "help") is True
|
|
assert "rawcli" in capsys.readouterr().out
|
|
|
|
def test_raw_renders_without_device(self, capsys):
|
|
assert dispatch(RawSession(), "30 04") is True
|
|
assert "30 04" in _plain(capsys.readouterr().out)
|
|
|
|
def test_bad_hex_reports(self, capsys):
|
|
assert dispatch(RawSession(), "0b0011") is True
|
|
assert "whole number of bytes" in capsys.readouterr().out
|
|
|
|
|
|
def _mock_device(scan14=None, scan15=None):
|
|
dev = MagicMock()
|
|
dev.hf.iso14a.scan.return_value = scan14 or {"found": False}
|
|
dev.hf.iso15.scan.return_value = scan15 or {"found": False}
|
|
dev.lf.search.return_value = None
|
|
return dev
|
|
|
|
|
|
class TestIdentify:
|
|
def test_no_device(self):
|
|
s = RawSession()
|
|
r = identify(s)
|
|
assert r["found"] is False and s.field is None
|
|
|
|
def test_14a_sets_field_and_name(self):
|
|
s = RawSession(device=_mock_device(
|
|
scan14={"found": True, "sak": 0x08, "uid": b"\x01\x02\x03\x04"}))
|
|
r = identify(s)
|
|
assert r["found"] and s.field == "hf" and s.protocol == "hf14a"
|
|
assert s.transponder == "MIFARE Classic 1K"
|
|
assert r["uid"] == b"\x01\x02\x03\x04"
|
|
|
|
def test_15693_fallback(self):
|
|
s = RawSession(device=_mock_device(scan15={"found": True, "uid": b"\xE0\x04"}))
|
|
r = identify(s)
|
|
assert r["found"] and s.field == "hf" and s.protocol == "hf15"
|
|
assert s.transponder == "ISO15693"
|
|
|
|
def test_name_unknown_sak(self):
|
|
assert name_14a({"sak": 0x99}) == "ISO14443-A (SAK 99)"
|
|
|
|
def test_clear_forgets_tag(self):
|
|
s = RawSession(device=_mock_device(scan14={"found": True, "sak": 0x08}))
|
|
identify(s)
|
|
clear(s)
|
|
assert s.field is None and s.transponder is None and s.protocol == "hf14a"
|
|
|
|
|
|
class TestControlCommands:
|
|
def test_identify_command(self, capsys):
|
|
s = RawSession(device=_mock_device(
|
|
scan14={"found": True, "sak": 0x00, "uid": b"\x04\x11\x22\x33"}))
|
|
dispatch(s, "identify")
|
|
out = capsys.readouterr().out
|
|
assert "MIFARE UL/NTAG" in out and "04112233" in out
|
|
assert s.breadcrumb() == "[raw / hf / MIFARE UL/NTAG / 0x]"
|
|
|
|
def test_transponder_when_none(self, capsys):
|
|
dispatch(RawSession(), "transponder")
|
|
assert "run 'identify'" in capsys.readouterr().out
|
|
|
|
def test_close(self, capsys):
|
|
s = RawSession(device=_mock_device(scan14={"found": True, "sak": 0x08}))
|
|
dispatch(s, "identify")
|
|
dispatch(s, "close")
|
|
assert "closed" in capsys.readouterr().out
|
|
assert s.transponder is None
|
|
|
|
|
|
class TestCatalog:
|
|
def test_type2_builds(self):
|
|
assert TYPE2.get("READ").build("4") == b"\x30\x04"
|
|
assert TYPE2.get("read").build("0x04") == b"\x30\x04" # case + hex arg
|
|
assert TYPE2.get("FAST_READ").build("0", "3") == b"\x3A\x00\x03"
|
|
assert TYPE2.get("GET_VERSION").build() == b"\x60"
|
|
assert TYPE2.get("WRITE").build("4", "01020304") == b"\xA2\x04\x01\x02\x03\x04"
|
|
|
|
def test_iso15_builds(self):
|
|
assert ISO15.get("READ_BLOCK").build("4") == b"\x02\x20\x04"
|
|
assert ISO15.get("INVENTORY").build() == b"\x26\x01\x00"
|
|
|
|
def test_resolver(self):
|
|
s = RawSession()
|
|
assert catalog_for(s) is None # nothing identified
|
|
s.protocol, s.transponder = "hf14a", "NTAG215"
|
|
assert catalog_for(s) is TYPE2
|
|
s.transponder = "MIFARE Classic 1K"
|
|
assert catalog_for(s).name == "MIFARE Classic"
|
|
s.protocol, s.transponder = "hf15", "ISO15693"
|
|
assert catalog_for(s) is ISO15
|
|
|
|
|
|
class TestFunctionCalls:
|
|
def _id(self, sak=0x00):
|
|
return RawSession(device=_mock_device(
|
|
scan14={"found": True, "sak": sak, "uid": b"\x04\x01\x02\x03"}))
|
|
|
|
def test_call_read_sends_correct_bytes(self, capsys):
|
|
s = self._id()
|
|
dispatch(s, "identify")
|
|
s.device.hf.iso14a.raw.return_value = {"data": b"\xAA\xBB\xCC\xDD"}
|
|
capsys.readouterr()
|
|
dispatch(s, "READ(4)")
|
|
out = _plain(capsys.readouterr().out)
|
|
assert "30 04" in out # the built command hit the trace
|
|
s.device.hf.iso14a.raw.assert_called_with(b"\x30\x04")
|
|
|
|
def test_call_without_identify(self, capsys):
|
|
dispatch(RawSession(), "READ(4)")
|
|
assert "identify" in capsys.readouterr().out
|
|
|
|
def test_unknown_call(self, capsys):
|
|
s = self._id(); dispatch(s, "identify"); capsys.readouterr()
|
|
dispatch(s, "FLY(4)")
|
|
assert "unknown command" in capsys.readouterr().out
|
|
|
|
def test_bad_args_shows_usage(self, capsys):
|
|
s = self._id(); dispatch(s, "identify"); capsys.readouterr()
|
|
dispatch(s, "READ()") # missing page arg
|
|
assert "usage: READ(page)" in capsys.readouterr().out
|
|
|
|
def test_help_lists_catalog(self, capsys):
|
|
s = self._id(); dispatch(s, "identify"); capsys.readouterr()
|
|
dispatch(s, "help")
|
|
out = capsys.readouterr().out
|
|
assert "READ(page)" in out and "GET_VERSION" in out
|
|
|
|
def test_help_command_detail(self, capsys):
|
|
s = self._id(); dispatch(s, "identify"); capsys.readouterr()
|
|
dispatch(s, "help READ")
|
|
assert "0x30" in capsys.readouterr().out
|
|
|
|
|
|
class TestTLV:
|
|
# NULL + NDEF (a well-known text record "Hi"/en) + TERMINATOR
|
|
TEXT_REC = bytes.fromhex("D101055402656E4869") # D1 01 05 T 02 'en' 'Hi'
|
|
BLOCK = bytes([0x00, 0x03, len(TEXT_REC)]) + TEXT_REC + bytes([0xFE])
|
|
|
|
def test_parse_structure(self):
|
|
tlvs = parse_tlv(self.BLOCK)
|
|
tags = [t.tag for t in tlvs]
|
|
assert tags == [0x00, 0x03, 0xFE]
|
|
ndef = tlvs[1]
|
|
assert ndef.length == len(self.TEXT_REC) and ndef.value == self.TEXT_REC
|
|
|
|
def test_extended_length(self):
|
|
big = bytes([0x03, 0xFF, 0x01, 0x00]) + b"\x00" * 256 + bytes([0xFE])
|
|
tlvs = parse_tlv(big)
|
|
assert tlvs[0].tag == 0x03 and tlvs[0].length == 256
|
|
|
|
def test_format_multiline(self):
|
|
out = format_tlv(self.BLOCK)
|
|
assert "NULL" in out
|
|
assert "NDEF MESSAGE len=9" in out
|
|
assert "TERMINATOR" in out
|
|
assert "\n" in out # genuinely multi-line
|
|
assert "→" in out # NDEF annotation line present
|
|
|
|
def test_tlv_command(self, capsys):
|
|
dispatch(RawSession(), "tlv " + self.BLOCK.hex())
|
|
assert "NDEF MESSAGE" in capsys.readouterr().out
|
|
|
|
|
|
class TestCompleter:
|
|
def _complete(self, session, text):
|
|
from prompt_toolkit.document import Document
|
|
return list(RawCompleter(session).get_completions(Document(text), None))
|
|
|
|
def test_completes_control_verbs(self):
|
|
cs = self._complete(RawSession(), "ide")
|
|
assert any(c.text == "identify" for c in cs)
|
|
|
|
def test_completes_catalog_commands_with_meta(self):
|
|
s = RawSession()
|
|
s.protocol, s.transponder = "hf14a", "NTAG215" # -> Type 2 catalog
|
|
cs = self._complete(s, "REA")
|
|
texts = [c.text for c in cs]
|
|
assert "READ(" in texts
|
|
read = next(c for c in cs if c.text == "READ(")
|
|
assert "0x30" in read.display_meta_text # tooltip = help
|
|
|
|
def test_no_catalog_before_identify(self):
|
|
cs = self._complete(RawSession(), "REA")
|
|
assert all(c.text != "READ(" for c in cs) # no tag => no catalog commands
|
|
|
|
def test_help_argument_completes_command_names(self):
|
|
s = RawSession()
|
|
s.protocol, s.transponder = "hf14a", "NTAG215"
|
|
cs = self._complete(s, "help GET")
|
|
assert any(c.text == "GET_VERSION" for c in cs)
|