diff --git a/conftest.py b/conftest.py index abf47c532..b81dda2e5 100644 --- a/conftest.py +++ b/conftest.py @@ -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. diff --git a/nanobot/pairing/store.py b/nanobot/pairing/store.py index 11d5ca787..260b2cb0f 100644 --- a/nanobot/pairing/store.py +++ b/nanobot/pairing/store.py @@ -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, } diff --git a/tests/pairing/test_store.py b/tests/pairing/test_store.py index 928782c1e..0f681bd9b 100644 --- a/tests/pairing/test_store.py +++ b/tests/pairing/test_store.py @@ -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"])