mirror of
https://github.com/HKUDS/nanobot.git
synced 2026-05-19 16:12:30 +00:00
f8e7e50759
- Add os.fsync with Windows-compatible directory flush in pairing store - Increase pairing code length from 6 -> 8 characters for higher entropy - Remove SystemExit on empty allowFrom; empty list now defers to pairing - Update is_allowed docstring to document pairing fallback semantics - Propagate is_dm to Matrix (direct rooms) and Slack (im channels) - Slack _is_allowed now checks pairing store for DM allowlist mode - Fix /pairing revoke to accept optional channel argument - Move inline import time to module top-level - Add WebSocket comment explaining is_dm=True assumption - Add comprehensive tests for store and BaseChannel pairing integration - Fix existing tests that expected empty allowFrom to hard-exit Refs #3774
100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
import time
|
|
|
|
import pytest
|
|
|
|
from nanobot.pairing import store
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _tmp_store(tmp_path, monkeypatch):
|
|
path = tmp_path / "pairing.json"
|
|
monkeypatch.setattr(store, "_store_path", lambda: path)
|
|
|
|
|
|
class TestGenerateCode:
|
|
def test_format(self) -> None:
|
|
code = store.generate_code("telegram", "123")
|
|
assert len(code) == 9 # 4 + 1 + 4
|
|
assert code[4] == "-"
|
|
assert code.replace("-", "").isalnum()
|
|
assert code.replace("-", "").isupper()
|
|
|
|
def test_uniqueness(self) -> None:
|
|
codes = {store.generate_code("telegram", str(i)) for i in range(20)}
|
|
assert len(codes) == 20
|
|
|
|
def test_ttl_expiration(self) -> None:
|
|
code = store.generate_code("telegram", "123", ttl=1)
|
|
assert store.approve_code(code) is not None
|
|
|
|
code2 = store.generate_code("telegram", "456", ttl=0)
|
|
time.sleep(0.1)
|
|
assert store.approve_code(code2) is None
|
|
|
|
|
|
class TestApproveDeny:
|
|
def test_approve_moves_to_approved(self) -> None:
|
|
code = store.generate_code("telegram", "123")
|
|
assert store.is_approved("telegram", "123") is False
|
|
|
|
result = store.approve_code(code)
|
|
assert result == ("telegram", "123")
|
|
assert store.is_approved("telegram", "123") is True
|
|
assert store.get_approved("telegram") == ["123"]
|
|
|
|
def test_deny_removes_pending(self) -> None:
|
|
code = store.generate_code("telegram", "123")
|
|
assert store.deny_code(code) is True
|
|
assert store.approve_code(code) is None
|
|
|
|
def test_deny_unknown_returns_false(self) -> None:
|
|
assert store.deny_code("UNKNOWN") is False
|
|
|
|
def test_approve_expired_returns_none(self) -> None:
|
|
code = store.generate_code("telegram", "123", ttl=0)
|
|
time.sleep(0.1)
|
|
assert store.approve_code(code) is None
|
|
|
|
|
|
class TestRevoke:
|
|
def test_revoke_removes_sender(self) -> None:
|
|
code = store.generate_code("telegram", "123")
|
|
store.approve_code(code)
|
|
assert store.is_approved("telegram", "123") is True
|
|
|
|
assert store.revoke("telegram", "123") is True
|
|
assert store.is_approved("telegram", "123") is False
|
|
assert store.get_approved("telegram") == []
|
|
|
|
def test_revoke_unknown_returns_false(self) -> None:
|
|
assert store.revoke("telegram", "999") is False
|
|
|
|
|
|
class TestListPending:
|
|
def test_empty(self) -> None:
|
|
assert store.list_pending() == []
|
|
|
|
def test_shows_pending(self) -> None:
|
|
store.generate_code("telegram", "123")
|
|
store.generate_code("discord", "456")
|
|
pending = store.list_pending()
|
|
assert len(pending) == 2
|
|
channels = {p["channel"] for p in pending}
|
|
assert channels == {"telegram", "discord"}
|
|
|
|
def test_expired_not_listed(self) -> None:
|
|
store.generate_code("telegram", "123", ttl=0)
|
|
time.sleep(0.1)
|
|
assert store.list_pending() == []
|
|
|
|
|
|
class TestStoreDurability:
|
|
def test_corruption_recovery(self, tmp_path, monkeypatch) -> None:
|
|
path = tmp_path / "pairing.json"
|
|
path.write_text("not json{", encoding="utf-8")
|
|
monkeypatch.setattr(store, "_store_path", lambda: path)
|
|
|
|
# Should recover gracefully and act as empty store
|
|
assert store.list_pending() == []
|
|
assert store.is_approved("telegram", "123") is False
|