pm3py/tests/test_sim_reader.py

"""Tests for pm3py.sim.reader — Reader ABC, ScriptedReader, InteractiveReader."""
import asyncio
import pytest

from pm3py.sim.frame import RFFrame
from pm3py.sim.medium import SoftwareMedium
from pm3py.sim.reader import Reader, ScriptedReader, InteractiveReader, ReaderStep, StepResult
from pm3py.sim.transponder import Transponder


class EchoTransponder(Transponder):
    """Returns the received frame back as response."""

    async def power_on(self) -> None:
        pass

    async def power_off(self) -> None:
        pass

    async def handle_frame(self, frame: RFFrame) -> RFFrame | None:
        return frame

    @property
    def state(self) -> str:
        return "ECHO"


class FixedTransponder(Transponder):
    """Returns a fixed response to any frame."""

    def __init__(self, response: RFFrame):
        super().__init__()
        self._response = response

    async def power_on(self) -> None:
        pass

    async def power_off(self) -> None:
        pass

    async def handle_frame(self, frame: RFFrame) -> RFFrame | None:
        return self._response

    @property
    def state(self) -> str:
        return "FIXED"


class SimpleReader(Reader):
    """Minimal reader implementation for testing the base class."""

    async def run(self) -> dict:
        resp = await self.transceive(RFFrame.from_hex("26"))
        return {"response": resp}


class TestReaderBase:
    """Test Reader ABC transceive."""

    def test_transceive_sends_and_receives(self):
        loop = asyncio.get_event_loop()
        medium = SoftwareMedium()
        loop.run_until_complete(medium.attach(EchoTransponder()))
        reader = SimpleReader(medium)

        result = loop.run_until_complete(reader.run())
        assert result["response"] is not None
        assert result["response"].data == b"\x26"

    def test_transceive_no_transponder_returns_none(self):
        loop = asyncio.get_event_loop()
        medium = SoftwareMedium()
        reader = SimpleReader(medium)

        result = loop.run_until_complete(reader.run())
        assert result["response"] is None


class TestScriptedReader:
    """Test scripted reader sequence execution."""

    def test_run_script_all_pass(self):
        loop = asyncio.get_event_loop()
        medium = SoftwareMedium()
        response = RFFrame.from_hex("4400")
        loop.run_until_complete(medium.attach(FixedTransponder(response)))

        steps = [
            ReaderStep(
                frame=RFFrame.from_hex("26"),
                description="Send REQA",
                check=lambda r: r is not None and r.data == b"\x44\x00",
            ),
            ReaderStep(
                frame=RFFrame.from_hex("9320"),
                description="Send ANTICOL",
                check=lambda r: r is not None,
            ),
        ]

        scripted = ScriptedReader(medium)
        results = loop.run_until_complete(scripted.run_script(steps))

        assert len(results) == 2
        assert all(r.passed for r in results)

    def test_run_script_check_fails(self):
        loop = asyncio.get_event_loop()
        medium = SoftwareMedium()
        response = RFFrame.from_hex("4400")
        loop.run_until_complete(medium.attach(FixedTransponder(response)))

        steps = [
            ReaderStep(
                frame=RFFrame.from_hex("26"),
                description="Send REQA",
                check=lambda r: r is not None and r.data == b"\xFF\xFF",  # wrong
            ),
        ]

        scripted = ScriptedReader(medium)
        results = loop.run_until_complete(scripted.run_script(steps))

        assert len(results) == 1
        assert not results[0].passed

    def test_stop_on_fail(self):
        loop = asyncio.get_event_loop()
        medium = SoftwareMedium()
        response = RFFrame.from_hex("4400")
        loop.run_until_complete(medium.attach(FixedTransponder(response)))

        steps = [
            ReaderStep(
                frame=RFFrame.from_hex("26"),
                description="Fail here",
                check=lambda r: False,
                stop_on_fail=True,
            ),
            ReaderStep(
                frame=RFFrame.from_hex("9320"),
                description="Should not run",
            ),
        ]

        scripted = ScriptedReader(medium)
        results = loop.run_until_complete(scripted.run_script(steps))

        assert len(results) == 1  # stopped after first failure

    def test_no_check_means_pass(self):
        loop = asyncio.get_event_loop()
        medium = SoftwareMedium()
        loop.run_until_complete(medium.attach(FixedTransponder(RFFrame.from_hex("00"))))

        steps = [
            ReaderStep(frame=RFFrame.from_hex("26"), description="No check"),
        ]

        scripted = ScriptedReader(medium)
        results = loop.run_until_complete(scripted.run_script(steps))

        assert results[0].passed


class TestInteractiveReader:
    """Test interactive reader step-through."""

    def test_send_and_receive(self):
        loop = asyncio.get_event_loop()
        medium = SoftwareMedium()
        loop.run_until_complete(medium.attach(EchoTransponder()))

        interactive = InteractiveReader(medium)
        resp = loop.run_until_complete(interactive.send(RFFrame.from_hex("26")))

        assert resp is not None
        assert resp.data == b"\x26"

    def test_send_hex_convenience(self):
        loop = asyncio.get_event_loop()
        medium = SoftwareMedium()
        loop.run_until_complete(medium.attach(EchoTransponder()))

        interactive = InteractiveReader(medium)
        resp = loop.run_until_complete(interactive.send_hex("A5"))

        assert resp is not None
        assert resp.data == b"\xA5"

    def test_history_tracks_exchanges(self):
        loop = asyncio.get_event_loop()
        medium = SoftwareMedium()
        loop.run_until_complete(medium.attach(EchoTransponder()))

        interactive = InteractiveReader(medium)
        loop.run_until_complete(interactive.send_hex("26"))
        loop.run_until_complete(interactive.send_hex("9320"))

        assert len(interactive.history) == 2
        assert interactive.history[0][0].data == b"\x26"
        assert interactive.history[1][0].data == b"\x93\x20"

    def test_dump_trace_returns_string(self):
        loop = asyncio.get_event_loop()
        medium = SoftwareMedium()
        loop.run_until_complete(medium.attach(EchoTransponder()))

        interactive = InteractiveReader(medium)
        loop.run_until_complete(interactive.send_hex("26"))

        trace = interactive.dump_trace()
        assert isinstance(trace, str)
        assert len(trace) > 0