pm3py/tests/test_sim_medium.py

"""Tests for pm3py.sim.medium — Medium ABC and SoftwareMedium."""
import asyncio
import pytest

from pm3py.sim.frame import RFFrame
from pm3py.sim.medium import SoftwareMedium
from pm3py.sim.transponder import Transponder


class StubTransponder(Transponder):
    """Minimal transponder that echoes back a fixed response."""

    def __init__(self, response: RFFrame | None = None):
        super().__init__()
        self._response = response
        self._state = "IDLE"
        self._powered = False
        self._received: list[RFFrame] = []

    async def power_on(self) -> None:
        self._powered = True
        self._state = "READY"

    async def power_off(self) -> None:
        self._powered = False
        self._state = "OFF"

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

    @property
    def state(self) -> str:
        return self._state


class TestSoftwareMediumAttachDetach:
    """Test transponder attachment and detachment."""

    def test_attach_returns_unique_ids(self):
        medium = SoftwareMedium()
        t1 = StubTransponder()
        t2 = StubTransponder()
        id1 = asyncio.get_event_loop().run_until_complete(medium.attach(t1))
        id2 = asyncio.get_event_loop().run_until_complete(medium.attach(t2))
        assert id1 != id2

    def test_attach_powers_on_transponder(self):
        medium = SoftwareMedium()
        t = StubTransponder()
        assert not t._powered
        asyncio.get_event_loop().run_until_complete(medium.attach(t))
        assert t._powered

    def test_detach_powers_off_transponder(self):
        medium = SoftwareMedium()
        t = StubTransponder()
        loop = asyncio.get_event_loop()
        tid = loop.run_until_complete(medium.attach(t))
        loop.run_until_complete(medium.detach(tid))
        assert not t._powered

    def test_detach_nonexistent_id_raises(self):
        medium = SoftwareMedium()
        with pytest.raises(KeyError):
            asyncio.get_event_loop().run_until_complete(medium.detach(999))


class TestSoftwareMediumSingleTransponder:
    """Test basic transmit/receive with one transponder."""

    def test_reader_transmit_delivers_frame_to_transponder(self):
        medium = SoftwareMedium()
        t = StubTransponder(response=RFFrame.from_hex("AABB"))
        loop = asyncio.get_event_loop()
        loop.run_until_complete(medium.attach(t))

        cmd = RFFrame.from_hex("26")  # REQA
        loop.run_until_complete(medium.transmit_reader(cmd))

        assert len(t._received) == 1
        assert t._received[0].data == b"\x26"

    def test_reader_receives_transponder_response(self):
        medium = SoftwareMedium()
        response = RFFrame.from_hex("4400")
        t = StubTransponder(response=response)
        loop = asyncio.get_event_loop()
        loop.run_until_complete(medium.attach(t))

        cmd = RFFrame.from_hex("26")
        loop.run_until_complete(medium.transmit_reader(cmd))
        result = loop.run_until_complete(medium.receive_reader())

        assert result is not None
        assert result.data == b"\x44\x00"

    def test_no_transponders_returns_none(self):
        medium = SoftwareMedium()
        loop = asyncio.get_event_loop()
        cmd = RFFrame.from_hex("26")
        loop.run_until_complete(medium.transmit_reader(cmd))
        result = loop.run_until_complete(medium.receive_reader())
        assert result is None

    def test_transponder_returning_none_means_no_response(self):
        medium = SoftwareMedium()
        t = StubTransponder(response=None)  # stays quiet
        loop = asyncio.get_event_loop()
        loop.run_until_complete(medium.attach(t))

        cmd = RFFrame.from_hex("26")
        loop.run_until_complete(medium.transmit_reader(cmd))
        result = loop.run_until_complete(medium.receive_reader())
        assert result is None


class TestSoftwareMediumCollision:
    """Test multi-transponder collision detection."""

    def test_two_identical_responses_no_collision(self):
        medium = SoftwareMedium()
        t1 = StubTransponder(response=RFFrame.from_hex("A5"))
        t2 = StubTransponder(response=RFFrame.from_hex("A5"))
        loop = asyncio.get_event_loop()
        loop.run_until_complete(medium.attach(t1))
        loop.run_until_complete(medium.attach(t2))

        loop.run_until_complete(medium.transmit_reader(RFFrame.from_hex("26")))
        result = loop.run_until_complete(medium.receive_reader())

        assert result is not None
        assert result.data == b"\xA5"
        assert not result.has_collision

    def test_two_different_responses_detect_collision(self):
        medium = SoftwareMedium()
        t1 = StubTransponder(response=RFFrame.from_hex("A0"))  # 10100000
        t2 = StubTransponder(response=RFFrame.from_hex("A5"))  # 10100101
        loop = asyncio.get_event_loop()
        loop.run_until_complete(medium.attach(t1))
        loop.run_until_complete(medium.attach(t2))

        loop.run_until_complete(medium.transmit_reader(RFFrame.from_hex("26")))
        result = loop.run_until_complete(medium.receive_reader())

        assert result is not None
        assert result.has_collision
        assert len(result.collision_positions) == 2

    def test_three_transponders_one_quiet(self):
        """One transponder stays quiet — only two respond."""
        medium = SoftwareMedium()
        t1 = StubTransponder(response=RFFrame.from_hex("AB"))
        t2 = StubTransponder(response=None)  # quiet
        t3 = StubTransponder(response=RFFrame.from_hex("AB"))
        loop = asyncio.get_event_loop()
        loop.run_until_complete(medium.attach(t1))
        loop.run_until_complete(medium.attach(t2))
        loop.run_until_complete(medium.attach(t3))

        loop.run_until_complete(medium.transmit_reader(RFFrame.from_hex("26")))
        result = loop.run_until_complete(medium.receive_reader())

        assert result is not None
        assert result.data == b"\xAB"
        assert not result.has_collision

    def test_collision_bits_default_to_one(self):
        medium = SoftwareMedium()
        t1 = StubTransponder(response=RFFrame.from_bytes(b"\x00"))
        t2 = StubTransponder(response=RFFrame.from_bytes(b"\xFF"))
        loop = asyncio.get_event_loop()
        loop.run_until_complete(medium.attach(t1))
        loop.run_until_complete(medium.attach(t2))

        loop.run_until_complete(medium.transmit_reader(RFFrame.from_hex("26")))
        result = loop.run_until_complete(medium.receive_reader())

        # All 8 bits collide, all should be 1
        assert result.data == b"\xFF"


class TestSoftwareMediumDynamicInjection:
    """Test attach/detach mid-protocol."""

    def test_detach_mid_protocol(self):
        medium = SoftwareMedium()
        t1 = StubTransponder(response=RFFrame.from_hex("AA"))
        t2 = StubTransponder(response=RFFrame.from_hex("BB"))
        loop = asyncio.get_event_loop()
        loop.run_until_complete(medium.attach(t1))
        tid2 = loop.run_until_complete(medium.attach(t2))

        # First round: collision
        loop.run_until_complete(medium.transmit_reader(RFFrame.from_hex("26")))
        r1 = loop.run_until_complete(medium.receive_reader())
        assert r1.has_collision

        # Detach t2
        loop.run_until_complete(medium.detach(tid2))

        # Second round: no collision
        loop.run_until_complete(medium.transmit_reader(RFFrame.from_hex("26")))
        r2 = loop.run_until_complete(medium.receive_reader())
        assert not r2.has_collision
        assert r2.data == b"\xAA"

    def test_attach_mid_protocol(self):
        medium = SoftwareMedium()
        t1 = StubTransponder(response=RFFrame.from_hex("AA"))
        loop = asyncio.get_event_loop()
        loop.run_until_complete(medium.attach(t1))

        # First round: single tag
        loop.run_until_complete(medium.transmit_reader(RFFrame.from_hex("26")))
        r1 = loop.run_until_complete(medium.receive_reader())
        assert not r1.has_collision

        # Inject new tag
        t2 = StubTransponder(response=RFFrame.from_hex("BB"))
        loop.run_until_complete(medium.attach(t2))

        # Second round: collision
        loop.run_until_complete(medium.transmit_reader(RFFrame.from_hex("26")))
        r2 = loop.run_until_complete(medium.receive_reader())
        assert r2.has_collision