AuthBot/tests/test_state.py

"""验证状态持久化与超时定时器测试。"""

import asyncio
import json
import tempfile
import time
from pathlib import Path

from authbot.state import (
    ForeignClickTracker,
    PendingRecord,
    RecentSet,
    VerifyState,
    pending_key,
)


def make_record(group="G1", member="M1", deadline=None, token="tok1", answer=7):
    return PendingRecord(
        group_openid=group,
        member_openid=member,
        deadline=deadline if deadline is not None else time.time() + 600,
        token=token,
        answer=answer,
        created_at=time.time(),
    )


def test_put_get_pop_roundtrip(tmp_path):
    async def scenario():
        async def noop(record):
            pass

        state = VerifyState(tmp_path, on_timeout=noop)
        rec = make_record()
        state.put(rec)
        assert state.get(rec.key) is not None

        # 新实例从磁盘恢复（模拟重启）
        state2 = VerifyState(tmp_path, on_timeout=noop)
        restored = state2.get(rec.key)
        assert restored is not None
        assert restored.token == rec.token
        assert restored.answer == rec.answer
        assert restored.deadline == rec.deadline

        assert state2.pop(rec.key).token == rec.token
        assert state2.get(rec.key) is None
        assert state2.pop(rec.key) is None  # 重复 pop 安全

    asyncio.run(scenario())


def test_pop_persists_deletion(tmp_path):
    async def scenario():
        async def noop(record):
            pass

        state = VerifyState(tmp_path, on_timeout=noop)
        rec = make_record()
        state.put(rec)
        state.pop(rec.key)
        assert VerifyState(tmp_path, on_timeout=noop).get(rec.key) is None

    asyncio.run(scenario())


def test_verified_trust_window(tmp_path):
    async def scenario():
        async def noop(record):
            pass

        state = VerifyState(tmp_path, on_timeout=noop)
        key = pending_key("G1", "M1")
        state.mark_verified(key)

        # trust_days=0：即使刚通过也不免验
        assert state.is_verified_recently(key, 0) is False
        assert state.is_verified_recently(key, 1) is True

        # 手动把通过时间改到 3 天前
        verified = json.loads((tmp_path / "verified.json").read_text("utf-8"))
        verified[key] = time.time() - 3 * 86400
        (tmp_path / "verified.json").write_text(json.dumps(verified), "utf-8")
        state2 = VerifyState(tmp_path, on_timeout=noop)
        assert state2.is_verified_recently(key, 1) is False
        assert state2.is_verified_recently(key, 7) is True

    asyncio.run(scenario())


def test_summary_remaining_seconds(tmp_path):
    async def scenario():
        async def noop(record):
            pass

        state = VerifyState(tmp_path, on_timeout=noop)
        rec = make_record(deadline=time.time() + 120)
        state.put(rec)
        [(got, remain)] = state.summary()
        assert got.key == rec.key
        assert 110 < remain <= 120

    asyncio.run(scenario())


def test_timeout_timer_fires_and_pop_cancels(tmp_path):
    async def scenario():
        fired = []

        async def on_timeout(record):
            fired.append(record.key)

        state = VerifyState(tmp_path, on_timeout=on_timeout)
        # 已到点的：立即触发
        state.put(make_record(member="EXPIRED", deadline=time.time() - 1))
        # 未到点的随后被 pop：不应触发
        state.put(make_record(member="CANCELLED", deadline=time.time() + 5))
        state.pop(pending_key("G1", "CANCELLED"))
        await asyncio.sleep(0.3)
        assert fired == [pending_key("G1", "EXPIRED")]

    asyncio.run(scenario())


def test_restore_rearms_from_disk(tmp_path):
    async def scenario():
        async def noop(record):
            pass

        # 先写入一条已过期与一条未过期的记录
        VerifyState(tmp_path, on_timeout=noop).put(
            make_record(member="OLD", deadline=time.time() - 1)
        )
        VerifyState(tmp_path, on_timeout=noop).put(
            make_record(member="NEW", deadline=time.time() + 600)
        )

        fired = []

        async def on_timeout(record):
            fired.append(record.key)

        state = VerifyState(tmp_path, on_timeout=on_timeout)
        await state.restore()
        await asyncio.sleep(0.3)
        # 只有已过期的触发，未到期的不触发
        assert fired == [pending_key("G1", "OLD")]
        assert state.get(pending_key("G1", "NEW")) is not None

    asyncio.run(scenario())


def test_timeout_callback_exception_isolated(tmp_path):
    """回调抛异常不炸定时器任务本身，也不影响其他记录。"""

    async def scenario():
        async def boom(record):
            raise RuntimeError("boom")

        state = VerifyState(tmp_path, on_timeout=boom)
        state.put(make_record(member="A", deadline=time.time() - 1))
        state.put(make_record(member="B", deadline=time.time() - 1))
        await asyncio.sleep(0.3)  # 两条回调都跑完且异常被记录

    asyncio.run(scenario())


def test_corrupt_state_file_backed_up(tmp_path):
    async def scenario():
        async def noop(record):
            pass

        (tmp_path / "pending.json").write_text("{not json", "utf-8")
        state = VerifyState(tmp_path, on_timeout=noop)
        assert state.all_pending() == []
        assert (tmp_path / "pending.json.bad").exists()

    asyncio.run(scenario())


def test_recent_set():
    s = RecentSet(maxsize=3)
    assert s.add("a") is True
    assert s.add("a") is False
    assert s.add("b") is True
    assert s.add("c") is True
    assert s.add("d") is True  # 挤掉最旧的 a
    assert s.add("a") is True  # a 已被淘汰，重新视为首次


def test_foreign_click_tracker_counts_and_resets():
    t = ForeignClickTracker(limit=3, window=1800.0)
    now = 1000.0
    assert t.register("k", now) == 1
    assert t.register("k", now + 10) == 2
    assert t.register("k", now + 20) == 3  # 到达上限
    t.reset("k")
    assert t.register("k", now + 30) == 1  # 处罚后重新计数


def test_foreign_click_tracker_window_prunes_old():
    t = ForeignClickTracker(limit=3, window=60.0)
    now = 1000.0
    assert t.register("k", now) == 1
    assert t.register("k", now + 61) == 1  # 窗口外的第一次点击不计入
    assert t.register("k", now + 61 + 30) == 2


def test_foreign_click_tracker_keys_independent():
    t = ForeignClickTracker(limit=2)
    assert t.register("a") == 1
    assert t.register("b") == 1
    assert t.register("a") == 2


def test_get_by_token():
    async def scenario():
        async def noop(record):
            pass

        state = VerifyState(Path(tempfile.mkdtemp()), on_timeout=noop)
        rec = make_record(token="tok-xyz")
        state.put(rec)
        assert state.get_by_token("tok-xyz").key == rec.key
        assert state.get_by_token("nope") is None
        state.pop(rec.key)

    asyncio.run(scenario())