buffer_platform/tests/test_dual_write_and_migration.py

"""
数据缓冲平台 - 双写一致性与数据迁移完整性验证测试

本测试通过内存 Fake Redis 与 mock/fake MySQL 验证：
1. save_market_data 成功后 MySQL 与 Redis 数据内容一致；
2. Redis 写入失败不影响 MySQL 持久化；
3. migrate_sqlite_to_mysql 的幂等性与迁移记录数正确性；
4. 使用真实内存数据库做端到端迁移验证时暴露的生产代码缺陷。
"""
import json
from datetime import datetime
from unittest.mock import MagicMock, patch

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

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


class FakeRedis:
    """内存模拟 Redis 客户端，支持 setex / get / delete / keys / ping。"""

    def __init__(self, fail_on_write=False):
        self._store = {}
        self.fail_on_write = fail_on_write

    def ping(self):
        return True

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

    def setex(self, key, seconds, value):
        if self.fail_on_write:
            raise Exception("Redis 写入失败")
        self._store[key] = value

    def delete(self, *keys):
        if self.fail_on_write:
            raise Exception("Redis 删除失败")
        for key in keys:
            self._store.pop(key, None)

    def keys(self, pattern):
        prefix = pattern.replace("*", "")
        return [k for k in self._store.keys() if k.startswith(prefix)]


def _make_mysql_memory():
    """创建内存 MySQL 模拟引擎和会话工厂。"""
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(bind=engine)
    SessionLocal = sessionmaker(bind=engine)
    return engine, SessionLocal


def _create_sqlite_source(tmp_path, market_count=2, timestamp_count=1):
    """创建包含测试数据的 SQLite 源数据库，并返回引擎与记录数。"""
    db_path = tmp_path / "test_source.db"
    engine = create_engine(f"sqlite:///{db_path}")
    Base.metadata.create_all(bind=engine)
    Session = sessionmaker(bind=engine)
    session = Session()
    fetched_at = datetime(2026, 7, 4, 10, 0, 0)
    for i in range(market_count):
        session.add(
            MarketData(
                symbol="AG2606",
                data_type="futures",
                period=f"{(i + 1) * 5}min",
                candles_json='[{"open": 100, "close": 110}]',
                current_price=123.45,
                fetched_at=fetched_at,
                candle_count=1,
            )
        )
    for _ in range(timestamp_count):
        session.add(
            SymbolTimestamp(
                symbol="AG2606",
                data_type="futures",
                last_refresh_at=fetched_at,
                refresh_count=5,
            )
        )
    session.commit()
    session.close()
    return engine, market_count, timestamp_count


def _make_mysql_session(market_data_count=0, symbol_timestamp_count=0):
    """构造模拟 MySQL 会话，支持记录写入和计数查询。"""
    mysql_session = MagicMock()
    mysql_session.__enter__ = MagicMock(return_value=mysql_session)
    mysql_session.__exit__ = MagicMock(return_value=False)

    added_market_data = []
    added_symbol_timestamps = []

    def mock_query(model):
        query = MagicMock()
        if model is MarketData:
            query.count.return_value = market_data_count
        elif model is SymbolTimestamp:
            query.count.return_value = symbol_timestamp_count
        return query

    mysql_session.query.side_effect = mock_query
    mysql_session.add_all.side_effect = lambda records: (
        added_market_data.extend(records)
        if records and isinstance(records[0], MarketData)
        else added_symbol_timestamps.extend(records)
    )

    return mysql_session, added_market_data, added_symbol_timestamps


@pytest.fixture
def sample_market_data():
    """构造与采集脚本格式一致的测试数据。"""
    return {
        "type": "futures",
        "current_price": 123.45,
        "timestamp": datetime(2026, 7, 4, 10, 0, 0).isoformat(),
        "timeframes": {
            "5min": [{"open": 100, "close": 110}],
            "15min": [{"open": 105, "close": 115}],
        },
    }


class TestDualWriteConsistency:
    """双写一致性验证。"""

    def test_save_market_data_writes_consistent_data_to_mysql_and_redis(
        self, sample_market_data
    ):
        """save_market_data 成功后，MySQL 与 Redis 中数据内容一致。"""
        redis_client = FakeRedis()
        mysql_engine, MySQLSessionLocal = _make_mysql_memory()
        manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine)

        result = manager.save_market_data("AG2606", sample_market_data)

        assert result is True

        # 验证 MySQL 持久化数据
        with MySQLSessionLocal() as db:
            records = db.query(MarketData).filter_by(symbol="AG2606").all()
            assert len(records) == 2
            mysql_by_period = {r.period: r for r in records}
            assert "5min" in mysql_by_period
            assert "15min" in mysql_by_period
            assert mysql_by_period["5min"].current_price == 123.45
            assert mysql_by_period["15min"].current_price == 123.45

        # 验证 Redis 缓存数据
        redis_5min = json.loads(redis_client.get("market_data:AG2606:5min"))
        redis_15min = json.loads(redis_client.get("market_data:AG2606:15min"))
        assert redis_5min["current_price"] == 123.45
        assert redis_15min["current_price"] == 123.45
        assert redis_5min["candles"] == [{"open": 100, "close": 110}]
        assert redis_15min["candles"] == [{"open": 105, "close": 115}]

        # 验证 MySQL 与 Redis 内容一致
        assert (
            mysql_by_period["5min"].current_price == redis_5min["current_price"]
        )
        assert (
            mysql_by_period["15min"].current_price == redis_15min["current_price"]
        )
        assert (
            json.loads(mysql_by_period["5min"].candles_json) == redis_5min["candles"]
        )
        assert (
            json.loads(mysql_by_period["15min"].candles_json)
            == redis_15min["candles"]
        )

    def test_redis_write_failure_does_not_affect_mysql_persistence(
        self, sample_market_data
    ):
        """Redis 写入失败时，MySQL 数据持久化不受影响，save_market_data 仍返回 True。"""
        redis_client = FakeRedis(fail_on_write=True)
        mysql_engine, MySQLSessionLocal = _make_mysql_memory()
        manager = StorageManager(redis_client=redis_client, mysql_engine=mysql_engine)

        result = manager.save_market_data("AG2606", sample_market_data)

        assert result is True

        # 验证 MySQL 仍有完整数据
        with MySQLSessionLocal() 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"}

        # 验证 Redis 无缓存写入
        assert redis_client.get("market_data:AG2606:5min") is None
        assert redis_client.get("market_data:AG2606:15min") is None


class TestMigrationIntegrity:
    """数据迁移完整性验证。"""

    def test_migration_copies_expected_record_counts_from_sqlite_to_mysql(
        self, tmp_path
    ):
        """MySQL 为空时，migrate_sqlite_to_mysql 迁移 SQLite 中全部记录。"""
        sqlite_engine, sqlite_market_count, sqlite_timestamp_count = (
            _create_sqlite_source(tmp_path)
        )
        mysql_session, added_market_data, added_symbol_timestamps = (
            _make_mysql_session(market_data_count=0, symbol_timestamp_count=0)
        )

        with patch("app.migration.create_engine", return_value=sqlite_engine):
            with patch(
                "app.migration.MySQLSessionLocal", return_value=mysql_session
            ):
                from app.migration import migrate_sqlite_to_mysql

                result = migrate_sqlite_to_mysql()

        assert result is True
        assert len(added_market_data) == sqlite_market_count
        assert len(added_symbol_timestamps) == sqlite_timestamp_count
        assert added_market_data[0].symbol == "AG2606"
        assert added_market_data[0].period == "5min"
        assert added_symbol_timestamps[0].symbol == "AG2606"
        mysql_session.commit.assert_called_once()

    def test_migration_is_idempotent_when_mysql_has_data(self, tmp_path):
        """MySQL 已有数据时迁移被跳过，保持幂等性，不覆盖已有数据。"""
        sqlite_engine, _, _ = _create_sqlite_source(tmp_path)
        mysql_session, added_market_data, added_symbol_timestamps = (
            _make_mysql_session(market_data_count=1, symbol_timestamp_count=1)
        )

        with patch("app.migration.create_engine", return_value=sqlite_engine):
            with patch(
                "app.migration.MySQLSessionLocal", return_value=mysql_session
            ):
                from app.migration import migrate_sqlite_to_mysql

                result = migrate_sqlite_to_mysql()

        assert result is False
        assert len(added_market_data) == 0
        assert len(added_symbol_timestamps) == 0
        mysql_session.commit.assert_not_called()

    @pytest.mark.xfail(
        reason="app/migration.py 对 detached ORM 实例直接 add_all 不会触发 INSERT，"
        "需使用 make_transient() 或重新构造记录。",
        strict=False,
    )
    def test_migration_end_to_end_with_in_memory_databases(self, tmp_path):
        """使用真实内存数据库验证迁移前后数据条数一致（当前已知缺陷）。"""
        sqlite_engine, sqlite_market_count, sqlite_timestamp_count = (
            _create_sqlite_source(tmp_path)
        )

        mysql_engine, MySQLSessionLocal = _make_mysql_memory()

        with patch("app.migration.create_engine", return_value=sqlite_engine):
            with patch("app.migration.MySQLSessionLocal", MySQLSessionLocal):
                from app.migration import migrate_sqlite_to_mysql

                result = migrate_sqlite_to_mysql()

        assert result is True

        mysql_session = MySQLSessionLocal()
        try:
            assert mysql_session.query(MarketData).count() == sqlite_market_count
            assert (
                mysql_session.query(SymbolTimestamp).count() == sqlite_timestamp_count
            )
        finally:
            mysql_session.close()