LQAgentPlatform/utils_test/Redis_Test/redis_ttl_bug_test.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Redis TTL不同步Bug单元测试

验证问题：store_file_info函数只更新meta的TTL，不更新content的TTL，
导致两个键的过期时间不同步，造成间歇性缓存失败。
"""

import asyncio
import json
import time
import pytest
from typing import Dict, Any

# 导入被测试的模块
import sys
from pathlib import Path
root_dir = Path(__file__).parent.parent
sys.path.append(str(root_dir))

from foundation.utils.redis_utils import store_file_info, get_file_info
from foundation.infrastructure.cache.redis_connection import RedisConnectionFactory


class TestRedisTTLSync:
    """Redis TTL同步问题测试类"""

    @pytest.fixture
    async def redis_store(self):
        """Redis连接fixture"""
        redis_store = await RedisConnectionFactory.get_redis_store()
        yield redis_store
        # 测试后清理
        await RedisConnectionFactory.close_all()

    @pytest.fixture
    async def test_file_data(self):
        """测试文件数据fixture"""
        file_id = "test_ttl_bug_file_123"
        file_info = {
            "file_name": "测试文档.pdf",
            "file_size": 1024,
            "file_content": "这是一个测试文件内容，用于验证TTL不同步问题。" * 50,
            "callback_task_id": "original_task_id",
            "upload_time": int(time.time())
        }
        return file_id, file_info

    async def test_ttl_sync_bug_reproduction(self, redis_store, test_file_data):
        """
        测试1: 复现TTL不同步Bug

        1. 存储完整文件信息
        2. 更新callback_task_id（触发TTL不同步）
        3. 等待原始content过期
        4. 验证获取文件内容失败
        """
        file_id, file_info = test_file_data

        print("\n=== 开始测试TTL不同步Bug复现 ===")

        # 步骤1: 存储完整文件信息，TTL=5秒（便于测试）
        print("1. 存储完整文件信息...")
        success = await store_file_info(file_id, file_info, expire_seconds=5)
        assert success, "存储文件信息失败"

        # 验证初始状态
        print("2. 验证初始状态...")
        retrieved_info = await get_file_info(file_id, include_content=True)
        assert retrieved_info is not None, "初始获取文件信息失败"
        assert retrieved_info['file_content'] is not None, "初始文件内容为空"
        print(f"   [OK] 文件内容长度: {len(retrieved_info['file_content'])}")

        # 检查键的TTL
        meta_ttl = await redis_store.ttl(f"meta:{file_id}")
        content_ttl = await redis_store.ttl(f"content:{file_id}")
        print(f"   [OK] 初始TTL - meta: {meta_ttl}s, content: {content_ttl}s")
        assert abs(meta_ttl - content_ttl) <= 1, "初始TTL应该基本同步"

        # 步骤3: 等待3秒，让TTL减少一些
        print("3. 等待3秒...")
        await asyncio.sleep(3)

        # 步骤4: 更新callback_task_id（触发Bug）
        print("4. 更新callback_task_id（触发TTL不同步Bug）...")
        update_success = await store_file_info(
            file_id,
            {'callback_task_id': 'updated_task_id'},
            expire_seconds=10  # 设置更长的TTL
        )
        assert update_success, "更新callback_task_id失败"

        # 步骤5: 检查TTL不同步问题
        print("5. 检查TTL不同步问题...")
        new_meta_ttl = await redis_store.ttl(f"meta:{file_id}")
        new_content_ttl = await redis_store.ttl(f"content:{file_id}")
        print(f"   [OK] 更新后TTL - meta: {new_meta_ttl}s, content: {new_content_ttl}s")

        # 验证TTL不同步：meta的TTL应该重置为10秒，content应该继续倒计时（约2秒）
        assert new_meta_ttl > 8, f"meta的TTL应该被重置为约10秒，实际为{new_meta_ttl}"
        assert new_content_ttl < 5, f"content的TTL应该继续倒计时，实际为{new_content_ttl}"
        assert abs(new_meta_ttl - new_content_ttl) > 3, "TTL应该明显不同步"

        print("   [PASS] TTL不同步Bug已复现！")

        # 步骤6: 等待content过期
        print("6. 等待content过期...")
        await asyncio.sleep(new_content_ttl + 1)

        # 步骤7: 验证获取文件内容失败
        print("7. 验证content过期后获取失败...")
        final_info = await get_file_info(file_id, include_content=True)

        # 根据当前bug逻辑，这里应该返回None（因为content缺失）
        assert final_info is None, f"content过期后应该返回None，实际返回: {final_info}"

        print("   [PASS] Bug验证成功：content过期后无法获取文件信息")

        # 清理测试数据
        await redis_store.delete(f"meta:{file_id}", f"content:{file_id}")

    async def test_only_meta_retrieval_works(self, redis_store, test_file_data):
        """
        测试2: 验证不包含content的获取仍然正常工作

        即使content过期，meta信息应该仍然可以获取
        """
        file_id, file_info = test_file_data

        print("\n=== 测试meta单独获取功能 ===")

        # 存储文件信息
        await store_file_info(file_id, file_info, expire_seconds=3)

        # 更新callback_task_id
        await store_file_info(file_id, {'callback_task_id': 'updated'}, expire_seconds=10)

        # 等待content过期
        await asyncio.sleep(4)

        # 测试不包含content的获取
        meta_only_info = await get_file_info(file_id, include_content=False)

        assert meta_only_info is not None, "meta信息应该可以正常获取"
        assert meta_only_info['callback_task_id'] == 'updated', "callback_task_id应该被正确更新"
        assert 'file_content' not in meta_only_info, "不应该包含file_content"

        print("   [PASS] meta信息单独获取正常")

        # 清理
        await redis_store.delete(f"meta:{file_id}")

    async def test_current_bug_detection_in_logs(self, redis_store, test_file_data):
        """
        测试3: 模拟日志中观察到的实际场景

        复现日志中的时间序列：
        - 11:41:53 文件信息存储
        - 11:42:01 获取失败（8秒后）
        """
        file_id, file_info = test_file_data

        print("\n=== 模拟日志中的实际场景 ===")

        # 步骤1: 模拟文件上传（11:41:53）
        print("1. 模拟文件上传（11:41:53）...")
        upload_time = time.time()
        await store_file_info(file_id, file_info, expire_seconds=10)  # 10秒TTL

        # 步骤2: 模拟一些处理时间
        await asyncio.sleep(2)

        # 步骤3: 模拟callback_task_id更新（可能发生在中间某个时间点）
        print("3. 模拟callback_task_id更新...")
        await store_file_info(file_id, {'callback_task_id': 'updated_task'}, expire_seconds=15)

        # 步骤4: 等待到11:42:01（约8秒后）
        elapsed_time = time.time() - upload_time
        wait_time = 8 - elapsed_time
        if wait_time > 0:
            print(f"4. 等待到11:42:01（还需要{wait_time:.1f}秒）...")
            await asyncio.sleep(wait_time)

        # 步骤5: 验证获取失败（如日志所示）
        print("5. 验证11:42:01时的获取失败...")
        failed_info = await get_file_info(file_id, include_content=True)

        assert failed_info is None, "模拟11:42:01时应该获取失败"
        print("   [PASS] 成功复现日志中的失败场景")

        # 清理
        await redis_store.delete(f"meta:{file_id}")


async def run_ttl_bug_tests():
    """运行TTL Bug测试"""
    print("[TEST] 开始Redis TTL不同步Bug测试")

    test_instance = TestRedisTTLSync()

    try:
        # 获取redis连接
        redis_store = await RedisConnectionFactory.get_redis_store()

        # 准备测试数据
        file_id = "test_ttl_bug_file_123"
        file_info = {
            "file_name": "测试文档.pdf",
            "file_size": 1024,
            "file_content": "这是一个测试文件内容，用于验证TTL不同步问题。" * 50,
            "callback_task_id": "original_task_id",
            "upload_time": int(time.time())
        }
        test_file_data = (file_id, file_info)

        # 运行测试
        await test_instance.test_ttl_sync_bug_reproduction(redis_store, test_file_data)
        await test_instance.test_only_meta_retrieval_works(redis_store, test_file_data)
        await test_instance.test_current_bug_detection_in_logs(redis_store, test_file_data)

        print("\n[PASS] 所有TTL Bug测试通过！")

    except Exception as e:
        print(f"\n[FAIL] 测试失败: {e}")
        import traceback
        traceback.print_exc()

    finally:
        await RedisConnectionFactory.close_all()


if __name__ == "__main__":
    # 直接运行测试
    asyncio.run(run_ttl_bug_tests())