"""缓存测试：权限缓存机制验证 测试权限缓存的功能，包括： - 缓存命中和未命中 - 缓存过期 - 缓存失效 - 并发缓存访问 - 缓存统计 """ 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, "缓存性能改进不足"