pm3py/tests/test_sim_table_compiler.py

"""Tests for table compiler and sim session."""
import asyncio
import struct
import pytest
from unittest.mock import AsyncMock

from pm3py.sim.frame import RFFrame
from pm3py.transponders.hf.iso14443a.base import Tag14443A_3, Tag14443A_4, _compute_bcc
from pm3py.transponders.hf.iso14443a.nxp.mifare_classic import MifareClassicTag
from pm3py.transponders.hf.iso15693.base import Tag15693
from pm3py.transponders.hf.iso14443a.nxp.desfire import DesfireTag
from pm3py.sim.table_compiler import (
    TableEntry, ResponseTable, TableCompiler,
    EML_READ, EML_WRITE, MATCH_PREFIX,
)
from pm3py.transponders.hf.iso15693.nxp.nxp_icode import NxpIcodeTag
from pm3py.transponders.hf.iso15693.nxp.icode_slix2 import IcodeSlix2Tag


def run(coro):
    return asyncio.get_event_loop().run_until_complete(coro)


# ---------------------------------------------------------------------------
# TableEntry
# ---------------------------------------------------------------------------

class TestTableEntry:
    def test_serialize_size(self):
        entry = TableEntry(match=b"\x26", response=b"\x04\x00")
        raw = entry.serialize()
        assert len(raw) == 120

    def test_serialize_roundtrip(self):
        entry = TableEntry(
            match=b"\x26",
            match_mode=0,
            response=b"\x04\x00",
        )
        raw = entry.serialize()
        assert len(raw) == 120

        parsed = TableEntry.deserialize(raw)
        assert parsed.match == b"\x26"
        assert parsed.response == b"\x04\x00"
        assert parsed.match_mode == 0

    def test_eml_fields_roundtrip(self):
        entry = TableEntry(
            match=b"\x30\x04",
            response=b"\x00",
            eml_action=1,
            eml_offset=64,
            eml_len=16,
            eml_resp_insert=1,
        )
        raw = entry.serialize()
        parsed = TableEntry.deserialize(raw)
        assert parsed.eml_action == 1
        assert parsed.eml_offset == 64
        assert parsed.eml_len == 16
        assert parsed.eml_resp_insert == 1

    def test_group_fields_roundtrip(self):
        entry = TableEntry(
            match=b"\x60\x00",
            response=b"\xAB\xCD\xEF\x01",
            group=1,
            activate_groups=0x00000006,
            deactivate_groups=0x00000001,
            set_auth=1,
            clear_auth=0,
        )
        raw = entry.serialize()
        parsed = TableEntry.deserialize(raw)
        assert parsed.group == 1
        assert parsed.activate_groups == 0x00000006
        assert parsed.deactivate_groups == 0x00000001
        assert parsed.set_auth == 1
        assert parsed.clear_auth == 0

    def test_consume_flag(self):
        entry = TableEntry(match=b"\x60\x00", response=b"\xAB\xCD\xEF\x01", flags=0x02)
        raw = entry.serialize()
        parsed = TableEntry.deserialize(raw)
        assert parsed.flags & 0x02

    def test_exact_match(self):
        entry = TableEntry(match=b"\x26", response=b"\x04\x00")
        assert entry.matches(b"\x26")
        assert not entry.matches(b"\x52")

    def test_prefix_match(self):
        entry = TableEntry(match=b"\x93\x20", match_mode=1, response=b"\x01\x02\x03\x04\x04")
        assert entry.matches(b"\x93\x20")
        assert entry.matches(b"\x93\x20\xFF\xFF")
        assert not entry.matches(b"\x95\x20")

    def test_crc_flag(self):
        entry = TableEntry(match=b"\x26", response=b"\x08", response_flags=0x01)
        assert entry.response_flags & 0x01  # CRC append flag

    def test_stateful_flag(self):
        entry = TableEntry(match=b"\x60\x00", response=b"\xAB\xCD\xEF\x01", flags=0x02)
        assert entry.flags & 0x02  # consume after use


# ---------------------------------------------------------------------------
# ResponseTable
# ---------------------------------------------------------------------------

class TestResponseTable:
    def test_serialize_multiple_entries(self):
        table = ResponseTable(entries=[
            TableEntry(match=b"\x26", response=b"\x04\x00"),
            TableEntry(match=b"\x52", response=b"\x04\x00"),
        ])
        raw = table.serialize()
        assert len(raw) == 240  # 2 * 120

    def test_from_trace(self):
        from pm3py.sim.replay import TraceEntry
        trace = [
            TraceEntry(direction="reader", frame=RFFrame.from_hex("26")),
            TraceEntry(direction="tag", frame=RFFrame.from_hex("0400")),
            TraceEntry(direction="reader", frame=RFFrame.from_hex("9320")),
            TraceEntry(direction="tag", frame=RFFrame.from_hex("0102030404")),
        ]
        table = ResponseTable.from_trace(trace)
        assert len(table.entries) == 2
        assert table.entries[0].match == b"\x26"
        assert table.entries[0].response == b"\x04\x00"

    def test_lookup(self):
        table = ResponseTable(entries=[
            TableEntry(match=b"\x26", response=b"\x04\x00"),
            TableEntry(match=b"\x52", response=b"\x04\x00"),
        ])
        entry = table.lookup(b"\x26")
        assert entry is not None
        assert entry.response == b"\x04\x00"

        assert table.lookup(b"\xFF") is None

    def test_overlay_replaces_matching(self):
        base = ResponseTable(entries=[
            TableEntry(match=b"\x26", response=b"\x04\x00"),
        ])
        overlay = ResponseTable(entries=[
            TableEntry(match=b"\x26", response=b"\x44\x00"),  # different response
        ])
        base.overlay(overlay)
        entry = base.lookup(b"\x26")
        assert entry.response == b"\x44\x00"


# ---------------------------------------------------------------------------
# TableCompiler — 14443-A
# ---------------------------------------------------------------------------

class TestTableCompiler14a:
    def test_compile_4byte_uid_tag(self):
        tag = Tag14443A_3(uid=b"\x01\x02\x03\x04", atqa=b"\x04\x00", sak=0x08)
        table = TableCompiler.compile_14a(tag)

        # Should have: REQA, WUPA, ANTICOL CL1, SELECT CL1
        assert len(table.entries) >= 4

        # REQA → ATQA
        reqa = table.lookup(b"\x26")
        assert reqa is not None
        assert reqa.response == b"\x04\x00"

        # WUPA → ATQA
        wupa = table.lookup(b"\x52")
        assert wupa is not None
        assert wupa.response == b"\x04\x00"

    def test_compile_7byte_uid_tag(self):
        tag = Tag14443A_3(uid=b"\x01\x02\x03\x04\x05\x06\x07", sak=0x08)
        table = TableCompiler.compile_14a(tag)

        # Should have: REQA, WUPA, ANTICOL CL1, SELECT CL1, ANTICOL CL2, SELECT CL2
        assert len(table.entries) >= 6

    def test_compile_part4_includes_rats(self):
        tag = Tag14443A_4(uid=b"\x01\x02\x03\x04", sak=0x20,
                          ats=b"\x05\x78\x80\x70\x02")
        table = TableCompiler.compile_14a(tag)

        # Should have RATS entry
        rats = table.lookup(b"\xE0")
        assert rats is not None
        assert rats.response == b"\x05\x78\x80\x70\x02"

    def test_compile_includes_hlta(self):
        tag = Tag14443A_3(uid=b"\x01\x02\x03\x04")
        table = TableCompiler.compile_14a(tag)

        # HLTA should NOT be in table (no response per spec)
        hlta = table.lookup(b"\x50\x00")
        assert hlta is None


class TestTableCompiler15693:
    def test_compile_15693_has_inventory(self):
        tag = Tag15693(uid=b"\xE0\x04\x01\x02\x03\x04\x05\x06", dsfid=0x00)
        table = TableCompiler.compile_15693(tag)

        assert len(table.entries) >= 1  # at least inventory

    def test_compile_15693_has_system_info(self):
        tag = Tag15693(uid=b"\xE0\x04\x01\x02\x03\x04\x05\x06",
                       dsfid=0x42, block_size=4, num_blocks=28)
        table = TableCompiler.compile_15693(tag)

        # System info should be present (unaddressed)
        sysinfo = table.lookup(bytes([0x02, 0x2B]))
        assert sysinfo is not None


class TestTableCompilerMifare:
    def test_compile_mifare_has_auth_entries(self):
        tag = MifareClassicTag(uid=b"\x01\x02\x03\x04", size="1k")
        table = TableCompiler.compile_mifare(tag)

        # Should have anticollision + auth entries for each sector
        assert len(table.entries) >= 4  # at least REQA, WUPA, ANTICOL, SELECT

    def test_compile_mifare_auth_is_stateful(self):
        """Auth entries should be marked stateful (consumed after use)."""
        tag = MifareClassicTag(uid=b"\x01\x02\x03\x04", size="1k")
        table = TableCompiler.compile_mifare(tag)

        # Find an auth entry (AUTH_A for block 0 = 0x60 0x00)
        auth_entries = [e for e in table.entries if len(e.match) >= 1 and e.match[0] == 0x60]
        if auth_entries:
            # Auth nonce response should be stateful
            assert auth_entries[0].flags & 0x02


# ---------------------------------------------------------------------------
# SimSession (mock transport)
# ---------------------------------------------------------------------------

class TestSimSession:
    def test_upload_table(self):
        from pm3py.sim.sim_session import SimSession
        t = AsyncMock()
        t.send_ng.return_value = AsyncMock()
        session = SimSession(t)

        table = ResponseTable(entries=[
            TableEntry(match=b"\x26", response=b"\x04\x00"),
        ])
        run(session.upload_table(table))

        t.send_ng.assert_called()
        # Verify the serialized data was sent
        call_args = t.send_ng.call_args
        assert len(call_args[0][1]) == 120  # one entry

    def test_upload_large_table_chunks(self):
        """Tables larger than 512 bytes should be chunked."""
        from pm3py.sim.sim_session import SimSession
        t = AsyncMock()
        t.send_ng.return_value = AsyncMock()
        session = SimSession(t)

        # 5 entries = 600 bytes > 512
        entries = [TableEntry(match=bytes([i]), response=bytes([i+1])) for i in range(5)]
        table = ResponseTable(entries=entries)
        run(session.upload_table(table))

        # Should have been called multiple times
        assert t.send_ng.call_count >= 2


# ---------------------------------------------------------------------------
# TableCompiler — NXP ICODE
# ---------------------------------------------------------------------------

class TestTableCompilerNxpIcode:
    def test_compile_has_get_random(self):
        tag = NxpIcodeTag(uid=b"\xE0\x04\x01\x02\x03\x04\x05\x06")
        table = TableCompiler.compile_nxp_icode(tag)
        entry = table.lookup(bytes([0x02, 0xB2, 0x04]))
        assert entry is not None
        assert len(entry.response) == 3  # flags + 2 random bytes

    def test_compile_has_get_nxp_system_info(self):
        tag = NxpIcodeTag(uid=b"\xE0\x04\x01\x02\x03\x04\x05\x06")
        table = TableCompiler.compile_nxp_icode(tag)
        entry = table.lookup(bytes([0x02, 0xAB, 0x04]))
        assert entry is not None
        assert len(entry.response) == 8  # flags + PP + PP_cond + lock + feature_flags(4)

    def test_compile_has_read_signature(self):
        sig = bytes(range(32))
        tag = NxpIcodeTag(uid=b"\xE0\x04\x01\x02\x03\x04\x05\x06", signature=sig)
        table = TableCompiler.compile_nxp_icode(tag)
        entry = table.lookup(bytes([0x02, 0xBD, 0x04]))
        assert entry is not None
        assert len(entry.response) == 33  # flags + 32-byte signature
        assert entry.response[1:] == sig

    def test_compile_no_config_register_entries(self):
        """0xA1/0xA2 are FAST INVENTORY READ / SET EAS, not config registers."""
        tag = NxpIcodeTag(uid=b"\xE0\x04\x01\x02\x03\x04\x05\x06")
        table = TableCompiler.compile_nxp_icode(tag)
        # Should NOT have entries for 0xA1 or 0xA2
        assert table.lookup(bytes([0x02, 0xA1, 0x04, 0x00])) is None
        assert table.lookup(bytes([0x02, 0xA2, 0x04, 0x00])) is None


# ---------------------------------------------------------------------------
# TableCompiler — SLIX2
# ---------------------------------------------------------------------------

class TestTableCompilerSlix2:
    def test_compile_has_set_password(self):
        tag = IcodeSlix2Tag(
            uid=b"\xE0\x04\x02\x01\x03\x04\x05\x06",
            read_password=0x12345678,
        )
        table = TableCompiler.compile_slix2(tag)
        fallback = [e for e in table.entries if
                    len(e.match) >= 2 and e.match[1] == 0xB3 and
                    e.match_mode == MATCH_PREFIX]
        assert len(fallback) > 0
        assert fallback[-1].response == bytes([0x01, 0x0F])

    def test_compile_set_password_sets_auth(self):
        tag = IcodeSlix2Tag(
            uid=b"\xE0\x04\x02\x01\x03\x04\x05\x06",
            read_password=0x12345678,
        )
        table = TableCompiler.compile_slix2(tag)
        auth_entries = [e for e in table.entries if e.set_auth != 0]
        assert len(auth_entries) > 0

    def test_compile_includes_nxp_base_entries(self):
        """SLIX2 table includes NXP base entries (GET_RANDOM, READ_SIGNATURE, etc.)."""
        tag = IcodeSlix2Tag(uid=b"\xE0\x04\x02\x01\x03\x04\x05\x06")
        table = TableCompiler.compile_slix2(tag)
        # Should have GET_RANDOM from nxp_icode compiler
        assert table.lookup(bytes([0x02, 0xB2, 0x04])) is not None
        # Should have GET_NXP_SYSTEM_INFO
        assert table.lookup(bytes([0x02, 0xAB, 0x04])) is not None
        # Should have READ_SIGNATURE
        assert table.lookup(bytes([0x02, 0xBD, 0x04])) is not None

    def test_compile_enable_privacy_entry(self):
        """SLIX2 table has ENABLE_PRIVACY (0xBA) entries when privacy password set."""
        tag = IcodeSlix2Tag(
            uid=b"\xE0\x04\x02\x01\x03\x04\x05\x06",
            privacy_password=0xAABBCCDD,
        )
        table = TableCompiler.compile_slix2(tag)
        # Should have ENABLE_PRIVACY entries (correct + fallback)
        enable_entries = [e for e in table.entries if
                         len(e.match) >= 2 and e.match[1] == 0xBA]
        assert len(enable_entries) >= 2  # correct password + fallback

    def test_compile_privacy_groups(self):
        """ENABLE_PRIVACY correct password entry activates a group."""
        tag = IcodeSlix2Tag(
            uid=b"\xE0\x04\x02\x01\x03\x04\x05\x06",
            privacy_password=0xAABBCCDD,
        )
        table = TableCompiler.compile_slix2(tag)
        privacy_entries = [e for e in table.entries if e.activate_groups != 0]
        assert len(privacy_entries) > 0

    def test_compile_eas_afi_password(self):
        """EAS/AFI password (pwd_id 0x10) generates a table entry."""
        tag = IcodeSlix2Tag(
            uid=b"\xE0\x04\x02\x01\x03\x04\x05\x06",
            eas_afi_password=0xAABBCCDD,
        )
        table = TableCompiler.compile_slix2(tag)
        auth_entries = [e for e in table.entries if e.set_auth == 0x10]
        assert len(auth_entries) > 0


# ---------------------------------------------------------------------------
# TableCompiler — EAS ALARM
# ---------------------------------------------------------------------------

class TestTableCompilerEasAlarm:
    def test_slix2_has_eas_alarm_entry(self):
        """SLIX2 table includes EAS ALARM (0xA5) entry."""
        tag = IcodeSlix2Tag(uid=b"\xE0\x04\x02\x01\x03\x04\x05\x06")
        tag.set_eas(True)
        table = TableCompiler.compile_slix2(tag)
        # EAS ALARM should match 0x02 0xA5
        entry = table.lookup(bytes([0x02, 0xA5]))
        assert entry is not None
        assert entry.response[0] == 0x00  # success
        assert len(entry.response) >= 33  # flags + 32-byte EAS sequence

    def test_eas_alarm_disabled_returns_none(self):
        """EAS ALARM entry not present when EAS disabled."""
        tag = IcodeSlix2Tag(uid=b"\xE0\x04\x02\x01\x03\x04\x05\x06")
        # EAS not set
        table = TableCompiler.compile_slix2(tag)
        entry = table.lookup(bytes([0x02, 0xA5]))
        # Should either be absent or return error
        assert entry is None or entry.response[0] & 0x01


# ---------------------------------------------------------------------------
# TableCompiler — ICODE 3
# ---------------------------------------------------------------------------

class TestTableCompilerIcode3:
    def test_compile_icode3_exists(self):
        from pm3py.transponders.hf.iso15693.nxp.icode3 import Icode3Tag
        tag = Icode3Tag()
        table = TableCompiler.compile_icode3(tag)
        assert len(table.entries) > 0

    def test_compile_icode3_includes_slix2_entries(self):
        """ICODE 3 table includes SLIX2 base entries."""
        from pm3py.transponders.hf.iso15693.nxp.icode3 import Icode3Tag
        tag = Icode3Tag()
        table = TableCompiler.compile_icode3(tag)
        # Should have GET_RANDOM, GET_NXP_SYSTEM_INFO, READ_SIGNATURE
        assert table.lookup(bytes([0x02, 0xB2])) is not None
        assert table.lookup(bytes([0x02, 0xAB])) is not None
        assert table.lookup(bytes([0x02, 0xBD])) is not None

    def test_compile_icode3_system_info_has_feature_flags(self):
        """ICODE 3 GET NXP SYSTEM INFO has proper feature flags."""
        from pm3py.transponders.hf.iso15693.nxp.icode3 import Icode3Tag
        tag = Icode3Tag()
        table = TableCompiler.compile_icode3(tag)
        entry = table.lookup(bytes([0x02, 0xAB]))
        assert entry is not None
        feature_flags = struct.unpack_from("<I", entry.response, 4)[0]
        assert feature_flags & (1 << 1)   # COUNTER
        assert feature_flags & (1 << 12)  # PRIVACY
        assert feature_flags & (1 << 13)  # DESTROY

    def test_compile_icode3_config_password(self):
        """ICODE 3 table includes config password (pwd_id 0x20)."""
        from pm3py.transponders.hf.iso15693.nxp.icode3 import Icode3Tag
        tag = Icode3Tag(config_password=0xAABBCCDD)
        table = TableCompiler.compile_icode3(tag)
        auth_entries = [e for e in table.entries if e.set_auth != 0]
        # Should have entries for config password (0x20)
        pwd_ids = set()
        for e in table.entries:
            if e.set_auth != 0 and len(e.match) >= 3:
                pwd_ids.add(e.match[2])  # pwd_id byte
        assert 0x20 in pwd_ids

    def test_session_dispatches_icode3(self):
        """SimSession uses compile_icode3 for Icode3Tag."""
        from pm3py.transponders.hf.iso15693.nxp.icode3 import Icode3Tag
        from pm3py.sim.sim_session import SimSession
        from unittest.mock import MagicMock

        tag = Icode3Tag()
        mock_ser = MagicMock()
        mock_ser.in_waiting = 0
        mock_ser.read.return_value = b""

        session = SimSession(port=mock_ser)
        session.start_15693(tag, compile=True, trace=False)
        calls = mock_ser.write.call_args_list
        assert len(calls) >= 2


# ---------------------------------------------------------------------------
# TableCompiler — ICODE DNA
# ---------------------------------------------------------------------------

class TestTableCompilerDna:
    def test_compile_icode_dna_exists(self):
        from pm3py.transponders.hf.iso15693.nxp.icode_dna import IcodeDnaTag
        tag = IcodeDnaTag()
        table = TableCompiler.compile_icode_dna(tag)
        assert len(table.entries) > 0

    def test_compile_includes_base_entries(self):
        """DNA table includes GET_RANDOM, READ_SIGNATURE from NXP base."""
        from pm3py.transponders.hf.iso15693.nxp.icode_dna import IcodeDnaTag
        tag = IcodeDnaTag()
        table = TableCompiler.compile_icode_dna(tag)
        assert table.lookup(bytes([0x02, 0xB2])) is not None  # GET_RANDOM
        assert table.lookup(bytes([0x02, 0xBD])) is not None  # READ_SIGNATURE

    def test_compile_has_dna_feature_flags(self):
        """GET NXP SYSTEM INFO has DNA-specific feature flags."""
        from pm3py.transponders.hf.iso15693.nxp.icode_dna import IcodeDnaTag
        tag = IcodeDnaTag()
        table = TableCompiler.compile_icode_dna(tag)
        entry = table.lookup(bytes([0x02, 0xAB]))
        assert entry is not None
        feature_flags = struct.unpack_from("<I", entry.response, 4)[0]
        assert feature_flags & (1 << 1)   # COUNTER
        assert feature_flags & (1 << 3)   # EAS_PP
        assert feature_flags & (1 << 4)   # AFI_PP
        assert feature_flags & (1 << 12)  # PRIVACY
        assert feature_flags & (1 << 13)  # DESTROY

    def test_compile_has_eas(self):
        """DNA table includes EAS entries."""
        from pm3py.transponders.hf.iso15693.nxp.icode_dna import IcodeDnaTag
        tag = IcodeDnaTag()
        tag.set_eas(True)
        table = TableCompiler.compile_icode_dna(tag)
        entry = table.lookup(bytes([0x02, 0xA5]))  # EAS ALARM
        assert entry is not None

    def test_session_dispatches_dna(self):
        """SimSession uses compile_icode_dna for IcodeDnaTag."""
        from pm3py.transponders.hf.iso15693.nxp.icode_dna import IcodeDnaTag
        from pm3py.sim.sim_session import SimSession
        from unittest.mock import MagicMock

        tag = IcodeDnaTag()
        mock_ser = MagicMock()
        mock_ser.in_waiting = 0
        mock_ser.read.return_value = b""

        session = SimSession(port=mock_ser)
        session.start_15693(tag, compile=True, trace=False)
        calls = mock_ser.write.call_args_list
        assert len(calls) >= 2


# ---------------------------------------------------------------------------
# SimSession — table compile on start_15693
# ---------------------------------------------------------------------------

class TestSimSessionTableCompile:
    def test_start_15693_compiles_nxp_table(self):
        """SimSession uses compile_icode3 for Icode3Tag."""
        from pm3py.transponders.hf.iso15693.nxp.icode3 import Icode3Tag
        from pm3py.sim.sim_session import SimSession
        from unittest.mock import MagicMock

        tag = Icode3Tag()
        mock_ser = MagicMock()
        mock_ser.in_waiting = 0
        mock_ser.read.return_value = b""

        session = SimSession(port=mock_ser)
        session.start_15693(tag, compile=True, trace=False)

        calls = mock_ser.write.call_args_list
        assert len(calls) >= 2  # SIM start + EML sync + table upload


# ---------------------------------------------------------------------------
# SimSession — table compile on start_15693
# ---------------------------------------------------------------------------

class TestSimSessionTableCompile:
    def test_start_15693_compiles_nxp_table(self):
        """start_15693 with compile=True should upload table entries."""
        from pm3py.sim.sim_session import SimSession
        from unittest.mock import MagicMock

        tag = NxpIcodeTag(uid=b"\xE0\x04\x01\x02\x03\x04\x05\x06")

        mock_ser = MagicMock()
        mock_ser.in_waiting = 0
        mock_ser.read.return_value = b""

        session = SimSession(port=mock_ser)
        session.start_15693(tag, compile=True, trace=False)

        # Should have written: SIM command + EML sync + table upload
        calls = mock_ser.write.call_args_list
        assert len(calls) >= 2

    def test_start_15693_no_compile_default(self):
        """start_15693 without compile should not upload table."""
        from pm3py.sim.sim_session import SimSession, CMD_SIM_TABLE_UPLOAD
        from unittest.mock import MagicMock

        tag = NxpIcodeTag(uid=b"\xE0\x04\x01\x02\x03\x04\x05\x06")

        mock_ser = MagicMock()
        mock_ser.in_waiting = 0
        mock_ser.read.return_value = b""

        session = SimSession(port=mock_ser)
        session.start_15693(tag, compile=False, trace=False)

        # Check no frame contains CMD_SIM_TABLE_UPLOAD
        for call in mock_ser.write.call_args_list:
            frame = call[0][0]
            if len(frame) >= 10:
                cmd_in_frame = struct.unpack_from("<H", frame, 6)[0]
                assert cmd_in_frame != CMD_SIM_TABLE_UPLOAD