绑定页支持贴卡倒计时过期重扫

talkingq-url/tests/test_binding_session_expiry.py

@@ -0,0 +1,155 @@
+from datetime import datetime, timedelta
+
+import pytest
+
+from banban.dao.binding import SESSION_STATUS_COMPLETED, SESSION_STATUS_EXPIRED
+from banban.service.binding import BindingService
+
+
+class FakeSession:
+    def __init__(self):
+        self.commits = 0
+        self.rollbacks = 0
+        self.closed = False
+
+    async def commit(self):
+        self.commits += 1
+
+    async def rollback(self):
+        self.rollbacks += 1
+
+    async def close(self):
+        self.closed = True
+
+
+@pytest.mark.asyncio
+async def test_finalize_nfc_bind_expires_late_card_without_binding(monkeypatch):
+    service = BindingService()
+    db_session = FakeSession()
+    expired_session = {
+        "id": 42,
+        "bind_token": "bind-token-expired",
+        "device_id": "TalkingQ_device001",
+        "target_child_id": 7,
+        "initiator_user_id": 99,
+        "expires_at": datetime.utcnow() - timedelta(seconds=1),
+    }
+    marked_statuses = []
+    activated_cards = []
+    completed_sessions = []
+
+    async def fake_get_session():
+        return db_session
+
+    async def fake_get_latest_pending_session_by_device(self, device_id):
+        assert device_id == "TalkingQ_device001"
+        return expired_session
+
+    async def fake_mark_session_status(self, session_id, status):
+        marked_statuses.append((session_id, status))
+
+    async def fake_activate_card(**kwargs):
+        activated_cards.append(kwargs)
+
+    async def fake_complete_nfc_bind(self, **kwargs):
+        completed_sessions.append(kwargs)
+
+    monkeypatch.setattr(service, "get_session", fake_get_session)
+    monkeypatch.setattr(
+        "banban.service.binding.BindingDAO.get_latest_pending_session_by_device",
+        fake_get_latest_pending_session_by_device,
+    )
+    monkeypatch.setattr("banban.service.binding.BindingDAO.mark_session_status", fake_mark_session_status)
+    monkeypatch.setattr("banban.service.binding.card_service.activate_card", fake_activate_card)
+    monkeypatch.setattr("banban.service.binding.BindingDAO.complete_nfc_bind", fake_complete_nfc_bind)
+
+    result = await service.finalize_nfc_bind(
+        device_id="TalkingQ_device001",
+        card_uuid="CARD_UUID_LATE",
+    )
+
+    assert result == {
+        "device_id": "TalkingQ_device001",
+        "bind_token": "bind-token-expired",
+        "status": SESSION_STATUS_EXPIRED,
+    }
+    assert marked_statuses == [(42, SESSION_STATUS_EXPIRED)]
+    assert activated_cards == []
+    assert completed_sessions == []
+    assert db_session.commits == 1
+    assert db_session.rollbacks == 0
+    assert db_session.closed is True
+
+
+@pytest.mark.asyncio
+async def test_finalize_nfc_bind_completes_before_expiry(monkeypatch):
+    service = BindingService()
+    db_session = FakeSession()
+    pending_session = {
+        "id": 43,
+        "bind_token": "bind-token-active",
+        "device_id": "TalkingQ_device001",
+        "target_child_id": 7,
+        "initiator_user_id": 99,
+        "expires_at": datetime.utcnow() + timedelta(minutes=10),
+    }
+    marked_statuses = []
+    activated_cards = []
+    completed_sessions = []
+
+    async def fake_get_session():
+        return db_session
+
+    async def fake_get_latest_pending_session_by_device(self, device_id):
+        assert device_id == "TalkingQ_device001"
+        return pending_session
+
+    async def fake_mark_session_status(self, session_id, status):
+        marked_statuses.append((session_id, status))
+
+    async def fake_activate_card(**kwargs):
+        activated_cards.append(kwargs)
+
+    async def fake_complete_nfc_bind(self, **kwargs):
+        completed_sessions.append(kwargs)
+
+    monkeypatch.setattr(service, "get_session", fake_get_session)
+    monkeypatch.setattr(
+        "banban.service.binding.BindingDAO.get_latest_pending_session_by_device",
+        fake_get_latest_pending_session_by_device,
+    )
+    monkeypatch.setattr("banban.service.binding.BindingDAO.mark_session_status", fake_mark_session_status)
+    monkeypatch.setattr("banban.service.binding.card_service.activate_card", fake_activate_card)
+    monkeypatch.setattr("banban.service.binding.BindingDAO.complete_nfc_bind", fake_complete_nfc_bind)
+
+    result = await service.finalize_nfc_bind(
+        device_id="TalkingQ_device001",
+        card_uuid="CARD_UUID_ON_TIME",
+    )
+
+    assert result == {
+        "device_id": "TalkingQ_device001",
+        "bind_token": "bind-token-active",
+        "status": SESSION_STATUS_COMPLETED,
+        "child_id": 7,
+        "card_uuid": "CARD_UUID_ON_TIME",
+    }
+    assert marked_statuses == []
+    assert activated_cards == [
+        {
+            "card_uuid": "CARD_UUID_ON_TIME",
+            "device_id": "TalkingQ_device001",
+            "db_session": db_session,
+        }
+    ]
+    assert completed_sessions == [
+        {
+            "session_id": 43,
+            "device_id": "TalkingQ_device001",
+            "child_id": 7,
+            "user_id": 99,
+        }
+    ]
+    assert db_session.commits == 1
+    assert db_session.rollbacks == 0
+    assert db_session.closed is True