|
|
"""
|
|
|
数据缓冲平台 - 双写一致性与数据迁移完整性验证测试
|
|
|
|
|
|
本测试通过内存 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()
|
|
|
|
|
|
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()
|
