ctms/backend/tests/test_permission_cache.py

"""缓存测试：权限缓存机制验证

测试权限缓存的功能，包括：
- 缓存命中和未命中
- 缓存过期
- 缓存失效
- 并发缓存访问
- 缓存统计
"""

import asyncio
import time
import uuid

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.permission_cache import PermissionCache, get_permission_cache, set_permission_cache
from app.core.project_permissions import get_project_role_permissions, get_member_role


@pytest.mark.asyncio
async def test_cache_hit():
    """测试缓存命中"""
    cache = PermissionCache()

    # 第一次调用：缓存未命中
    cache._project_permissions_cache["test_key"] = ({"test": "data"}, time.time())

    # 第二次调用：缓存命中
    cached_data, timestamp = cache._project_permissions_cache.get("test_key")
    assert cached_data == {"test": "data"}
    assert not cache._is_expired(timestamp, 300)


@pytest.mark.asyncio
async def test_cache_miss():
    """测试缓存未命中"""
    cache = PermissionCache()

    # 缓存中不存在该键
    assert "nonexistent_key" not in cache._project_permissions_cache


@pytest.mark.asyncio
async def test_cache_expiration():
    """测试缓存过期"""
    cache = PermissionCache(default_ttl=1)

    # 添加一个即将过期的缓存项
    cache._project_permissions_cache["test_key"] = ({"test": "data"}, time.time() - 2)

    # 验证缓存已过期
    _, timestamp = cache._project_permissions_cache["test_key"]
    assert cache._is_expired(timestamp, 1)


@pytest.mark.asyncio
async def test_cache_invalidation():
    """测试缓存失效"""
    cache = PermissionCache()
    study_id = uuid.uuid4()

    # 添加缓存
    cache._project_permissions_cache[cache._make_project_cache_key(study_id)] = (
        {"test": "data"},
        time.time(),
    )
    assert len(cache._project_permissions_cache) == 1

    # 失效缓存
    cache.invalidate_project_permissions(study_id)
    assert len(cache._project_permissions_cache) == 0


@pytest.mark.asyncio
async def test_member_role_cache_invalidation():
    """测试成员角色缓存失效"""
    cache = PermissionCache()
    study_id = uuid.uuid4()
    user_id = uuid.uuid4()

    # 添加缓存
    cache._member_role_cache[cache._make_member_cache_key(study_id, user_id)] = ("PM", time.time())
    assert len(cache._member_role_cache) == 1

    # 失效缓存
    cache.invalidate_member_role(study_id, user_id)
    assert len(cache._member_role_cache) == 0


@pytest.mark.asyncio
async def test_invalidate_all_member_roles():
    """测试失效项目中所有成员的角色缓存"""
    cache = PermissionCache()
    study_id = uuid.uuid4()

    # 添加多个成员的缓存
    for i in range(5):
        user_id = uuid.uuid4()
        cache._member_role_cache[cache._make_member_cache_key(study_id, user_id)] = ("PM", time.time())

    assert len(cache._member_role_cache) == 5

    # 失效项目中所有成员的缓存
    cache.invalidate_all_member_roles(study_id)
    assert len(cache._member_role_cache) == 0


@pytest.mark.asyncio
async def test_concurrent_cache_access():
    """测试并发缓存访问"""
    cache = PermissionCache()
    study_id = uuid.uuid4()

    async def add_to_cache(i):
        user_id = uuid.uuid4()
        cache._member_role_cache[cache._make_member_cache_key(study_id, user_id)] = ("PM", time.time())

    # 并发添加缓存
    tasks = [add_to_cache(i) for i in range(10)]
    await asyncio.gather(*tasks)

    # 验证所有缓存都被添加
    assert len(cache._member_role_cache) == 10


@pytest.mark.asyncio
async def test_cache_key_generation():
    """测试缓存键生成"""
    cache = PermissionCache()
    study_id = uuid.uuid4()
    user_id = uuid.uuid4()

    # 验证缓存键格式
    project_key = cache._make_project_cache_key(study_id)
    assert project_key.startswith("project_permissions:")
    assert str(study_id) in project_key

    member_key = cache._make_member_cache_key(study_id, user_id)
    assert member_key.startswith("member_role:")
    assert str(study_id) in member_key
    assert str(user_id) in member_key


@pytest.mark.asyncio
async def test_cache_stats():
    """测试缓存统计"""
    cache = PermissionCache()
    study_id = uuid.uuid4()

    # 添加缓存
    cache._project_permissions_cache[cache._make_project_cache_key(study_id)] = (
        {"test": "data"},
        time.time(),
    )
    cache._member_role_cache[cache._make_member_cache_key(study_id, uuid.uuid4())] = ("PM", time.time())

    # 获取统计信息
    stats = cache.get_cache_stats()
    assert stats["project_permissions_count"] == 1
    assert stats["member_role_count"] == 1
    assert stats["total_count"] == 2


@pytest.mark.asyncio
async def test_clear_all_cache():
    """测试清除所有缓存"""
    cache = PermissionCache()
    study_id = uuid.uuid4()

    # 添加缓存
    cache._project_permissions_cache[cache._make_project_cache_key(study_id)] = (
        {"test": "data"},
        time.time(),
    )
    cache._member_role_cache[cache._make_member_cache_key(study_id, uuid.uuid4())] = ("PM", time.time())

    assert len(cache._project_permissions_cache) == 1
    assert len(cache._member_role_cache) == 1

    # 清除所有缓存
    cache.clear_all()
    assert len(cache._project_permissions_cache) == 0
    assert len(cache._member_role_cache) == 0


@pytest.mark.asyncio
async def test_cache_with_different_ttl(db_session: AsyncSession, study_id: uuid.UUID):
    """测试不同 TTL 的缓存"""
    cache = PermissionCache()

    # 使用不同的 TTL 获取权限
    permissions1 = await cache.get_project_role_permissions(db_session, study_id, ttl=1)
    permissions2 = await cache.get_project_role_permissions(db_session, study_id, ttl=300)

    # 两次调用都应该返回相同的数据
    assert permissions1 == permissions2


@pytest.mark.asyncio
async def test_global_cache_instance():
    """测试全局缓存实例"""
    cache1 = get_permission_cache()
    cache2 = get_permission_cache()

    # 应该是同一个实例
    assert cache1 is cache2


@pytest.mark.asyncio
async def test_set_global_cache_instance():
    """测试设置全局缓存实例"""
    new_cache = PermissionCache()
    set_permission_cache(new_cache)

    cache = get_permission_cache()
    assert cache is new_cache


@pytest.mark.asyncio
async def test_cache_performance_improvement(db_session: AsyncSession, study_id: uuid.UUID):
    """测试缓存的性能改进

    验证缓存确实提高了性能。
    """
    cache = PermissionCache()
    set_permission_cache(cache)
    cache.clear_all()

    # 第一次调用：从数据库查询
    start_time = time.time()
    permissions1 = await cache.get_project_role_permissions(db_session, study_id)
    first_call_time = time.time() - start_time

    # 第二次调用：从缓存获取
    start_time = time.time()
    permissions2 = await cache.get_project_role_permissions(db_session, study_id)
    second_call_time = time.time() - start_time

    # 验证结果相同
    assert permissions1 == permissions2

    # 缓存调用应该快得多
    print(f"\n缓存性能改进：第一次 {first_call_time*1000:.2f}ms，第二次 {second_call_time*1000:.2f}ms")
    assert second_call_time < first_call_time / 2, "缓存性能改进不足"