xiaoxia-saas/tests/unit/test_memory_state_store.py

"""MemoryStateStore 单元测试 - 微信 OAuth state 存储

覆盖：正常存取、一次性消费、过期清理、并发安全、空 state 处理。
"""

from __future__ import annotations

import time
from threading import Thread

import pytest


class TestMemoryStateStore:
    def test_put_and_verify_success(self):
        """正常存入并校验成功"""
        from packages.application.auth.wechat_oauth_service import MemoryStateStore

        store = MemoryStateStore()
        store.put("test_state_123")
        assert store.verify_and_consume("test_state_123") is True

    def test_verify_nonexistent_state_fails(self):
        """不存在的 state 校验失败"""
        from packages.application.auth.wechat_oauth_service import MemoryStateStore

        store = MemoryStateStore()
        assert store.verify_and_consume("nonexistent") is False

    def test_state_single_use(self):
        """state 只能消费一次（防重放）"""
        from packages.application.auth.wechat_oauth_service import MemoryStateStore

        store = MemoryStateStore()
        store.put("single_use_state")
        assert store.verify_and_consume("single_use_state") is True
        assert store.verify_and_consume("single_use_state") is False

    def test_empty_state_rejected(self):
        """空字符串 state 校验失败"""
        from packages.application.auth.wechat_oauth_service import MemoryStateStore

        store = MemoryStateStore()
        store.put("")
        # 空字符串作为 key 技术上可以存，但业务层应该拒绝
        # 这里验证 store 本身行为一致性
        assert store.verify_and_consume("") is True  # 存入了就能通过一次
        assert store.verify_and_consume("") is False  # 消费后就没了

    def test_expired_state_cleaned(self):
        """过期 state 会被清理，校验失败"""
        from packages.application.auth.wechat_oauth_service import MemoryStateStore

        # TTL 设为 0.01 秒，快速过期
        store = MemoryStateStore(ttl_seconds=0.01)
        store.put("expire_me")
        time.sleep(0.02)
        assert store.verify_and_consume("expire_me") is False

    def test_multiple_states_independent(self):
        """多个 state 互不影响"""
        from packages.application.auth.wechat_oauth_service import MemoryStateStore

        store = MemoryStateStore()
        store.put("state_a")
        store.put("state_b")
        store.put("state_c")

        # 消费 b
        assert store.verify_and_consume("state_b") is True
        assert store.verify_and_consume("state_b") is False

        # a 和 c 仍然有效
        assert store.verify_and_consume("state_a") is True
        assert store.verify_and_consume("state_c") is True

    def test_clean_expired_doesnt_touch_valid(self):
        """过期清理不影响未过期的 state"""
        from packages.application.auth.wechat_oauth_service import MemoryStateStore

        store = MemoryStateStore(ttl_seconds=10)
        store.put("valid_state")

        # 手动触发清理（通过 verify 触发内部 clean_expired）
        # 由于所有 state 都没过期，清理不影响
        assert store.verify_and_consume("valid_state") is True

    def test_thread_safety_concurrent_put(self):
        """并发写入不丢数据"""
        from packages.application.auth.wechat_oauth_service import MemoryStateStore

        store = MemoryStateStore(ttl_seconds=60)
        states = [f"state_{i}" for i in range(100)]

        def put_states(states_list):
            for s in states_list:
                store.put(s)

        threads = [Thread(target=put_states, args=(states[i * 20 : (i + 1) * 20],)) for i in range(5)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        # 每个 state 都能消费一次
        for s in states:
            assert store.verify_and_consume(s) is True

    def test_thread_safety_concurrent_consume(self):
        """并发消费同一个 state 只有一个能成功"""
        from packages.application.auth.wechat_oauth_service import MemoryStateStore

        store = MemoryStateStore()
        store.put("contested_state")

        results = []

        def try_consume():
            results.append(store.verify_and_consume("contested_state"))

        threads = [Thread(target=try_consume) for _ in range(10)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()

        # 只有一个成功，其余失败
        assert sum(1 for r in results if r) == 1
        assert sum(1 for r in results if not r) == 9

    def test_default_ttl_is_10_minutes(self):
        """默认 TTL 是 600 秒（10分钟）"""
        from packages.application.auth.wechat_oauth_service import (
            STATE_TTL_SECONDS,
            MemoryStateStore,
        )

        assert STATE_TTL_SECONDS == 600
        store = MemoryStateStore()
        # 验证默认值生效：存入后立即验证应该通过
        store.put("default_ttl_test")
        assert store.verify_and_consume("default_ttl_test") is True