buffer_platform/tests/test_redis_cache_scenarios.py

"""
Redis 缓存命中/未命中场景验证测试

覆盖：
1. 调用 save_market_data 写入数据后，Redis 与 MySQL 同时存在一致数据
2. 首次调用 get_market_data（Redis 未命中）从 MySQL 读取并回填 Redis
3. 再次调用 get_market_data（Redis 命中）直接返回缓存数据，不访问 MySQL
4. 手动删除 Redis key 后，再次读取回源 MySQL 并重新回填 Redis
"""
import json
import logging
from datetime import datetime
from unittest.mock import MagicMock

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

from app.models import Base, MarketData
from app.storage_manager import StorageManager

# test_main_lifespan.py 中的 fixture 会 patch app.main.Base.metadata.create_all 且未正确恢复，
# 导致后续依赖 create_all 的测试失败。这里保留原始方法引用，必要时恢复。
_original_create_all = Base.metadata.create_all


class FakeRedis:
    """轻量级 Redis 替身，仅支持本测试所需的方法。"""

    def __init__(self):
        self._store = {}

    def ping(self):
        return True

    def get(self, key):
        return self._store.get(key)

    def setex(self, key, ttl, value):
        self._store[key] = value

    def delete(self, *keys):
        for key in keys:
            self._store.pop(key, None)

    def keys(self, pattern):
        # 仅支持以 * 结尾的通配模式
        prefix = pattern.rstrip("*")
        return [k for k in self._store if k.startswith(prefix)]

    def exists(self, key):
        return key in self._store


@pytest.fixture
def mysql_engine():
    """提供已创建表结构的内存 SQLite 引擎，作为 MySQL 替身。"""
    # 防御：修复上游 fixture 对 Base.metadata.create_all 的 patch 泄漏
    if isinstance(Base.metadata.create_all, MagicMock):
        Base.metadata.create_all = _original_create_all

    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(bind=engine)

    # 确认表已创建
    with engine.connect() as conn:
        result = conn.execute(
            text("SELECT name FROM sqlite_master WHERE type='table' AND name='market_data'")
        )
        assert result.fetchone() is not None

    yield engine
    engine.dispose()


@pytest.fixture
def fake_redis():
    return FakeRedis()


@pytest.fixture
def sample_data():
    fetched_at = datetime(2026, 7, 4, 10, 0, 0)
    return {
        "type": "futures",
        "current_price": 123.45,
        "timestamp": fetched_at.isoformat(),
        "timeframes": {
            "5min": [{"datetime": "2026-07-04T09:55:00", "open": 100, "close": 110}],
            "15min": [{"datetime": "2026-07-04T09:45:00", "open": 105, "close": 115}],
        },
    }


def _seed_mysql_market_data(engine, symbol="AG2606", data_type="futures"):
    """在 MySQL 替身中写入行情数据。"""
    fetched_at = datetime(2026, 7, 4, 10, 0, 0)
    Session = sessionmaker(bind=engine)
    with Session() as db:
        db.add(
            MarketData(
                symbol=symbol,
                data_type=data_type,
                period="5min",
                candles_json=json.dumps(
                    [{"datetime": "2026-07-04T09:55:00", "open": 100, "close": 110}]
                ),
                current_price=123.45,
                fetched_at=fetched_at,
                candle_count=1,
            )
        )
        db.commit()


class TestSaveMarketData:
    """save_market_data 写入与缓存回填场景。"""

    def test_save_market_data_writes_mysql_and_backfills_redis(
        self, fake_redis, mysql_engine, sample_data
    ):
        """写入后 MySQL 与 Redis 均保存一致数据。"""
        manager = StorageManager(
            redis_client=fake_redis, mysql_engine=mysql_engine
        )

        assert manager.save_market_data("AG2606", sample_data) is True

        # MySQL 中应写入两条记录
        Session = sessionmaker(bind=mysql_engine)
        with Session() as db:
            records = db.query(MarketData).filter_by(symbol="AG2606").all()
            assert len(records) == 2
            periods = {r.period for r in records}
            assert periods == {"5min", "15min"}
            for r in records:
                assert r.current_price == 123.45

        # Redis 中应回填对应缓存键
        assert fake_redis.exists("market_data:AG2606:5min")
        assert fake_redis.exists("market_data:AG2606:15min")

        cached_5min = json.loads(fake_redis.get("market_data:AG2606:5min"))
        assert cached_5min["current_price"] == 123.45
        assert cached_5min["candles"] == sample_data["timeframes"]["5min"]


class TestRedisCacheHitMiss:
    """Redis 缓存命中/未命中读取场景。"""

    def test_first_read_miss_falls_back_to_mysql_and_backfills_redis(
        self, fake_redis, mysql_engine
    ):
        """Redis 未命中时从 MySQL 读取并回填 Redis。"""
        _seed_mysql_market_data(mysql_engine)
        manager = StorageManager(
            redis_client=fake_redis, mysql_engine=mysql_engine
        )

        assert fake_redis.get("market_data:AG2606:5min") is None

        result = manager.get_market_data("AG2606", "futures", "5min")

        expected = {
            "current_price": 123.45,
            "timestamp": datetime(2026, 7, 4, 10, 0, 0).isoformat(),
            "candles": [{"datetime": "2026-07-04T09:55:00", "open": 100, "close": 110}],
        }
        assert result == expected

        # Redis 应已被回填
        assert fake_redis.exists("market_data:AG2606:5min")
        cached = json.loads(fake_redis.get("market_data:AG2606:5min"))
        assert cached == expected

    def test_second_read_hit_returns_from_redis_without_mysql_access(
        self, fake_redis, mysql_engine
    ):
        """Redis 命中时直接返回缓存，不再访问 MySQL。"""
        _seed_mysql_market_data(mysql_engine)
        manager = StorageManager(
            redis_client=fake_redis, mysql_engine=mysql_engine
        )

        first = manager.get_market_data("AG2606", "futures", "5min")
        assert fake_redis.exists("market_data:AG2606:5min")

        # 将 MySQL 引擎替换为会抛异常的 mock，以证明第二次读取不会访问 MySQL
        bad_engine = MagicMock()
        bad_engine.connect.side_effect = Exception("MySQL should not be accessed")
        manager.mysql_engine = bad_engine

        second = manager.get_market_data("AG2606", "futures", "5min")

        assert second == first
        bad_engine.connect.assert_not_called()

    def test_after_redis_key_deleted_falls_back_to_mysql_again(
        self, fake_redis, mysql_engine
    ):
        """手动删除 Redis key 后，读取回源 MySQL 并重新回填 Redis。"""
        _seed_mysql_market_data(mysql_engine)
        manager = StorageManager(
            redis_client=fake_redis, mysql_engine=mysql_engine
        )

        first = manager.get_market_data("AG2606", "futures", "5min")
        assert fake_redis.exists("market_data:AG2606:5min")

        # 模拟 TTL 过期或手动失效
        fake_redis.delete("market_data:AG2606:5min")
        assert fake_redis.get("market_data:AG2606:5min") is None

        second = manager.get_market_data("AG2606", "futures", "5min")

        assert second == first
        assert fake_redis.exists("market_data:AG2606:5min")

    def test_cache_hit_data_matches_mysql_data(
        self, fake_redis, mysql_engine
    ):
        """缓存命中时返回的数据应与 MySQL 中的数据一致。"""
        _seed_mysql_market_data(mysql_engine)
        manager = StorageManager(
            redis_client=fake_redis, mysql_engine=mysql_engine
        )

        # 首次读取触发回填
        from_redis = manager.get_market_data("AG2606", "futures", "5min")

        # 直接从 MySQL 读取原始记录作为基准
        Session = sessionmaker(bind=mysql_engine)
        with Session() as db:
            record = (
                db.query(MarketData)
                .filter_by(symbol="AG2606", data_type="futures", period="5min")
                .first()
            )
            assert record is not None
            assert from_redis["current_price"] == record.current_price
            assert from_redis["candles"] == json.loads(record.candles_json)


class TestCacheLogMessages:
    """验证缓存命中/未命中的日志输出。"""

    @pytest.mark.xfail(
        reason="当前 StorageManager 未输出 'Redis 缓存命中' 日志，仅记录行为断言",
        strict=True,
    )
    def test_cache_hit_logs_redis_hit(self, fake_redis, mysql_engine, caplog):
        """Redis 命中时日志应包含 'Redis 缓存命中'。"""
        _seed_mysql_market_data(mysql_engine)
        manager = StorageManager(
            redis_client=fake_redis, mysql_engine=mysql_engine
        )
        manager.get_market_data("AG2606", "futures", "5min")  # 回填

        with caplog.at_level(logging.INFO, logger="app.storage_manager"):
            manager.get_market_data("AG2606", "futures", "5min")  # 命中

        assert "Redis 缓存命中" in caplog.text

    @pytest.mark.xfail(
        reason="当前 StorageManager 未输出 'Redis 未命中，从 MySQL 读取' 日志，仅记录行为断言",
        strict=True,
    )
    def test_cache_miss_logs_redis_miss_and_mysql_read(
        self, fake_redis, mysql_engine, caplog
    ):
        """Redis 未命中并回源 MySQL 时日志应包含 'Redis 未命中，从 MySQL 读取'。"""
        _seed_mysql_market_data(mysql_engine)
        manager = StorageManager(
            redis_client=fake_redis, mysql_engine=mysql_engine
        )

        with caplog.at_level(logging.INFO, logger="app.storage_manager"):
            manager.get_market_data("AG2606", "futures", "5min")

        assert "Redis 未命中，从 MySQL 读取" in caplog.text