支持已绑定设备追加 NFC 卡

- 新增已绑定设备追加卡会话和小程序入口

- 允许同设备多张 active 卡,保留 card_uuid 唯一并补齐老库迁移

- 收紧未知卡自动绑定和跨设备卡复用

验证:talkingq-url/.venv/bin/python -m pytest tests;compileall;npm run build:weapp
This commit is contained in:
stu2not
2026-06-04 09:55:35 +08:00
parent 482fbf351f
commit 82c15b6abc
18 changed files with 646 additions and 92 deletions

View File

@@ -2,8 +2,8 @@ from datetime import datetime, timedelta
import pytest
from banban.dao.binding import SESSION_STATUS_COMPLETED, SESSION_STATUS_EXPIRED
from banban.service.binding import BindingService
from banban.dao.binding import BIND_MODE_CARD, BIND_MODE_DEVICE, SESSION_STATUS_COMPLETED, SESSION_STATUS_EXPIRED
from banban.service.binding import BindingError, BindingService
class FakeSession:
@@ -22,6 +22,80 @@ class FakeSession:
self.closed = True
@pytest.mark.asyncio
async def test_start_card_bind_requires_existing_authorized_binding_and_sends_nfc_command(monkeypatch):
service = BindingService()
db_session = FakeSession()
started_binds = []
sent_commands = []
async def fake_get_session():
return db_session
async def fake_get_by_device(self, device_id, user_id):
assert device_id == "TalkingQ_device001"
assert user_id == 99
return {
"device_id": "TalkingQ_device001",
"child_id": 7,
"owner_user_id": 99,
}
async def fake_start_bind(self, user_id, device_id, child_id, bind_mode):
started_binds.append((user_id, device_id, child_id, bind_mode))
return "bind-token-card", datetime.utcnow() + timedelta(minutes=10)
async def fake_send_bind_nfc_command(device_id):
sent_commands.append(device_id)
monkeypatch.setattr(service, "get_session", fake_get_session)
monkeypatch.setattr("banban.service.binding.BindingDAO.get_by_device", fake_get_by_device)
monkeypatch.setattr("banban.service.binding.BindingDAO.start_bind", fake_start_bind)
monkeypatch.setattr(service, "_send_bind_nfc_command", fake_send_bind_nfc_command)
bind_token, expires_at = await service.start_card_bind(user_id=99, device_id="TalkingQ_device001")
assert bind_token == "bind-token-card"
assert expires_at > datetime.utcnow()
assert started_binds == [(99, "TalkingQ_device001", 7, BIND_MODE_CARD)]
assert sent_commands == ["TalkingQ_device001"]
assert db_session.commits == 1
assert db_session.closed is True
@pytest.mark.asyncio
async def test_start_card_bind_rejects_family_member_who_is_not_owner(monkeypatch):
service = BindingService()
db_session = FakeSession()
started_binds = []
async def fake_get_session():
return db_session
async def fake_get_by_device(self, device_id, user_id):
assert device_id == "TalkingQ_device001"
assert user_id == 88
return {
"device_id": "TalkingQ_device001",
"child_id": 7,
"owner_user_id": 99,
}
async def fake_start_bind(self, *args, **kwargs):
started_binds.append((args, kwargs))
monkeypatch.setattr(service, "get_session", fake_get_session)
monkeypatch.setattr("banban.service.binding.BindingDAO.get_by_device", fake_get_by_device)
monkeypatch.setattr("banban.service.binding.BindingDAO.start_bind", fake_start_bind)
with pytest.raises(BindingError, match="only the device owner can add cards"):
await service.start_card_bind(user_id=88, device_id="TalkingQ_device001")
assert started_binds == []
assert db_session.commits == 0
assert db_session.closed is True
@pytest.mark.asyncio
async def test_finalize_nfc_bind_expires_late_card_without_binding(monkeypatch):
service = BindingService()
@@ -92,6 +166,7 @@ async def test_finalize_nfc_bind_completes_before_expiry(monkeypatch):
"target_child_id": 7,
"initiator_user_id": 99,
"expires_at": datetime.utcnow() + timedelta(minutes=10),
"bind_mode": BIND_MODE_DEVICE,
}
marked_statuses = []
activated_cards = []
@@ -110,6 +185,11 @@ async def test_finalize_nfc_bind_completes_before_expiry(monkeypatch):
async def fake_activate_card(**kwargs):
activated_cards.append(kwargs)
class FakeCard:
card_uuid = kwargs["card_uuid"]
return FakeCard()
async def fake_complete_nfc_bind(self, **kwargs):
completed_sessions.append(kwargs)
@@ -148,8 +228,86 @@ async def test_finalize_nfc_bind_completes_before_expiry(monkeypatch):
"device_id": "TalkingQ_device001",
"child_id": 7,
"user_id": 99,
"card_uuid": "CARD_UUID_ON_TIME",
}
]
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_adds_card_without_rebinding_device(monkeypatch):
service = BindingService()
db_session = FakeSession()
pending_session = {
"id": 44,
"bind_token": "bind-token-card",
"device_id": "TalkingQ_device001",
"target_child_id": 7,
"initiator_user_id": 99,
"expires_at": datetime.utcnow() + timedelta(minutes=10),
"bind_mode": BIND_MODE_CARD,
}
activated_cards = []
completed_card_sessions = []
completed_device_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):
raise AssertionError(f"unexpected status mark: {session_id}, {status}")
async def fake_activate_card(**kwargs):
activated_cards.append(kwargs)
class FakeCard:
card_uuid = kwargs["card_uuid"]
return FakeCard()
async def fake_complete_card_bind(self, **kwargs):
completed_card_sessions.append(kwargs)
async def fake_complete_nfc_bind(self, **kwargs):
completed_device_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_card_bind", fake_complete_card_bind)
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_EXTRA",
)
assert result == {
"device_id": "TalkingQ_device001",
"bind_token": "bind-token-card",
"status": SESSION_STATUS_COMPLETED,
"child_id": 7,
"card_uuid": "CARD_UUID_EXTRA",
}
assert activated_cards == [
{
"card_uuid": "CARD_UUID_EXTRA",
"device_id": "TalkingQ_device001",
"db_session": db_session,
}
]
assert completed_card_sessions == [{"session_id": 44, "card_uuid": "CARD_UUID_EXTRA"}]
assert completed_device_sessions == []
assert db_session.commits == 1
assert db_session.rollbacks == 0
assert db_session.closed is True