hookvision-rk/tests/test_rfid_client.py

import time
from types import SimpleNamespace

from hookvision.rfid_client import RFIDClient, _MIN_BACKOFF


class _FakeSock:
    def recv(self, size):
        raise OSError("connection reset")

    def close(self):
        pass


def _make_client(callbacks=None):
    callbacks = callbacks or {}
    config = SimpleNamespace(ip="127.0.0.1", port=1, stability_time=5, stability_count=3)
    client = RFIDClient(config, on_tag_callback=callbacks.get("on_tag"), on_state_changed=callbacks.get("on_state"))
    return client


class TestSessionCloseKeepsTag:
    def test_read_loop_exit_does_not_clear_tag(self):
        events = []
        client = _make_client({"on_tag": lambda tag: events.append(tag)})
        client.running = True
        client.sock = _FakeSock()
        client._read_loop()
        assert client.sock is None
        assert client._backoff == _MIN_BACKOFF
        assert client._backoff == 1
        assert events == []

    def test_check_idle_does_not_reconnect_within_backoff(self):
        events = []
        client = _make_client({"on_tag": lambda tag: events.append(tag)})
        client._last_connect_attempt = time.time()
        client.check_idle()
        assert client.sock is None
        assert events == []


class TestIdleTimeout:
    def test_idle_timeout_still_clears_tag_when_connected(self):
        events = []
        client = _make_client({"on_tag": lambda tag: events.append(tag)})
        client._last_tag = "AABBCC"
        client._last_time = time.time() - 10
        client._last_connect_attempt = time.time()
        # 模拟已连接状态
        client.sock = _FakeSock()
        client.check_idle()
        assert client._last_tag is None
        assert events == [None]

    def test_idle_timeout_does_not_clear_tag_when_disconnected(self):
        events = []
        client = _make_client({"on_tag": lambda tag: events.append(tag)})
        client._last_tag = "AABBCC"
        client._last_time = time.time() - 10
        client._last_connect_attempt = time.time()
        # socket为None（断开状态）
        client.check_idle()
        assert client._last_tag == "AABBCC"
        assert events == []