vee
142 lines
4.1 KiB
Python
142 lines
4.1 KiB
Python
from __future__ import annotations
|
|
|
|
import time
|
|
from dataclasses import dataclass
|
|
|
|
from cognitive_agent.infrastructure.nfc.reader import NfcReaderService
|
|
|
|
|
|
@dataclass
|
|
class FakePort:
|
|
device: str
|
|
description: str = ""
|
|
hwid: str = ""
|
|
|
|
|
|
class FakeReader:
|
|
def __init__(self, port: str) -> None:
|
|
self.port = port
|
|
self.closed = False
|
|
self.sam_configured = False
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
def get_firmware_version(self) -> tuple[int, int]:
|
|
return (1, 6)
|
|
|
|
def sam_configuration(self) -> bool:
|
|
self.sam_configured = True
|
|
return True
|
|
|
|
def read_uid(self) -> bytes:
|
|
return bytes.fromhex("04A1B2C3")
|
|
|
|
|
|
def wait_until(predicate, timeout: float = 1.0) -> bool:
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
if predicate():
|
|
return True
|
|
time.sleep(0.01)
|
|
return predicate()
|
|
|
|
|
|
def test_resolve_port_prefers_a_single_ch340_reader() -> None:
|
|
ports = [
|
|
FakePort("COM3", "Bluetooth serial"),
|
|
FakePort("COM5", "USB-SERIAL CH340", "USB VID:PID=1A86:7523"),
|
|
]
|
|
|
|
assert NfcReaderService.resolve_port("auto", ports) == "COM5"
|
|
assert NfcReaderService.resolve_port("COM8", ports) == "COM8"
|
|
|
|
|
|
def test_reader_service_publishes_reader_and_card_state() -> None:
|
|
created: list[FakeReader] = []
|
|
|
|
def create_reader(port: str) -> FakeReader:
|
|
reader = FakeReader(port)
|
|
created.append(reader)
|
|
return reader
|
|
|
|
service = NfcReaderService(
|
|
reader_factory=create_reader,
|
|
ports_provider=lambda: [FakePort("COM5", "PCR532 CH340")],
|
|
retry_seconds=0.01,
|
|
poll_seconds=0.01,
|
|
mqtt_enabled=False,
|
|
)
|
|
service.start()
|
|
try:
|
|
assert wait_until(lambda: service.snapshot()["card"] is not None)
|
|
snapshot = service.snapshot()
|
|
assert snapshot["reader"] == {
|
|
"online": True,
|
|
"state": "ready",
|
|
"port": "COM5",
|
|
"firmware": "1.6",
|
|
"error": None,
|
|
"source": "serial",
|
|
}
|
|
assert snapshot["card"]["uid"] == "04A1B2C3"
|
|
assert snapshot["card"]["action"] == "present"
|
|
assert created[0].sam_configured is True
|
|
finally:
|
|
service.close()
|
|
assert created[0].closed is True
|
|
|
|
|
|
def test_disabled_reader_is_non_blocking_and_reported() -> None:
|
|
service = NfcReaderService(enabled=False, mqtt_enabled=False)
|
|
service.start()
|
|
|
|
snapshot = service.snapshot()
|
|
assert snapshot["reader"]["online"] is False
|
|
assert snapshot["reader"]["state"] == "disabled"
|
|
assert snapshot["card"] is None
|
|
|
|
|
|
def test_existing_mqtt_bridge_is_reused_when_it_owns_the_reader() -> None:
|
|
service = NfcReaderService(enabled=True, mqtt_enabled=False)
|
|
|
|
service._handle_bridge_status(
|
|
{"online": True, "state": "ready", "port": "COM5", "firmware": "1.6"}
|
|
)
|
|
service._handle_bridge_card(
|
|
{"action": "present", "uid": "1a53dd95"}
|
|
)
|
|
service._handle_bridge_profiles(
|
|
{
|
|
"profiles": [
|
|
{
|
|
"id": "ambient-light",
|
|
"name": "第三方氛围灯",
|
|
"cardUid": "1a53dd95",
|
|
"actions": [
|
|
{"id": "red", "slot": 7, "name": "红色", "learned": True}
|
|
],
|
|
}
|
|
]
|
|
}
|
|
)
|
|
service._handle_local_bridge_status({"addresses": ["172.20.10.8"]})
|
|
|
|
snapshot = service.snapshot()
|
|
assert snapshot["reader"] == {
|
|
"online": True,
|
|
"state": "ready",
|
|
"port": "COM5",
|
|
"firmware": "1.6",
|
|
"error": None,
|
|
"source": "mqtt_bridge",
|
|
}
|
|
assert snapshot["card"]["uid"] == "1A53DD95"
|
|
assert snapshot["legacy_profiles"][0]["name"] == "第三方氛围灯"
|
|
assert snapshot["legacy_profiles"][0]["cardUid"] == "1A53DD95"
|
|
assert snapshot["legacy_profiles"][0]["actions"][0]["slot"] == 7
|
|
assert snapshot["legacy_bridge_addresses"] == ["172.20.10.8"]
|
|
|
|
service._handle_bridge_card({"action": "removed", "uid": "1A53DD95"})
|
|
assert service.snapshot()["card"] is None
|