fix(pairing): avoid duplicate pending requests

This commit is contained in:
Xubin Ren
2026-09-03 15:45:57 +08:00
parent 41477c2510
commit 54e5c63b7e
3 changed files with 30 additions and 2 deletions
+10
View File
@@ -49,6 +49,16 @@ def _isolate_sessions_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> I
yield
@pytest.fixture(autouse=True)
def _isolate_pairing_store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Keep channel pairing tests out of the user's active pairing store."""
pairing_path = tmp_path / "pairing.json"
monkeypatch.setattr(
"nanobot.pairing.store._store_path",
lambda: pairing_path,
)
@pytest.fixture(scope="session", autouse=True)
def _use_windows_system_ca_for_default_http_clients() -> Iterator[None]:
"""Avoid reparsing certifi's CA bundle for every offline HTTP client.
+7 -2
View File
@@ -115,19 +115,24 @@ def generate_code(
sender_id: str,
ttl: int = _TTL_DEFAULT_S,
) -> str:
"""Create a new pairing code for *sender_id* on *channel*.
"""Return an active pairing code for *sender_id* on *channel*.
Returns the code (e.g. ``"ABCD-EFGH"``).
"""
with _LOCK:
data = _load()
_gc_pending(data)
sender = str(sender_id)
for code, info in data.get("pending", {}).items():
if info["channel"] == channel and str(info["sender_id"]) == sender:
return code
raw = "".join(secrets.choice(_ALPHABET) for _ in range(_CODE_LENGTH))
code = f"{raw[:4]}-{raw[4:]}"
data.setdefault("pending", {})[code] = {
"channel": channel,
"sender_id": str(sender_id),
"sender_id": sender,
"created_at": time.time(),
"expires_at": time.time() + ttl,
}
+13
View File
@@ -32,6 +32,19 @@ class TestGenerateCode:
codes = {store.generate_code("telegram", str(i)) for i in range(20)}
assert len(codes) == 20
def test_reuses_active_code_for_same_sender(self) -> None:
first = store.generate_code("telegram", "123")
assert store.generate_code("telegram", "123") == first
assert len(store.list_pending()) == 1
def test_scopes_reused_codes_to_channel(self) -> None:
telegram = store.generate_code("telegram", "123")
discord = store.generate_code("discord", "123")
assert telegram != discord
assert len(store.list_pending()) == 2
def test_ttl_expiration(self, monkeypatch) -> None:
clock = {"now": 1_000.0}
monkeypatch.setattr(store.time, "time", lambda: clock["now"])