|
|
"""配置存储测试：MySQL + JSON fallback。"""
|
|
|
import json
|
|
|
from pathlib import Path
|
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
import pytest
|
|
|
|
|
|
from app.config_store import ConfigStore
|
|
|
|
|
|
|
|
|
def _make_session_maker(mock_db):
|
|
|
"""构造正确的 sessionmaker mock，使 with 语句返回 mock_db。"""
|
|
|
session_obj = MagicMock()
|
|
|
session_obj.__enter__ = MagicMock(return_value=mock_db)
|
|
|
session_obj.__exit__ = MagicMock(return_value=False)
|
|
|
|
|
|
session_maker = MagicMock(return_value=session_obj)
|
|
|
return session_maker
|
|
|
|
|
|
|
|
|
class TestConfigStoreGet:
|
|
|
"""ConfigStore.get_config 测试。"""
|
|
|
|
|
|
def test_get_config_returns_mysql_value_on_hit(self, tmp_path):
|
|
|
"""MySQL 命中时返回数据库中的配置。"""
|
|
|
mock_config = MagicMock()
|
|
|
mock_config.config_value = {"futures": {"沪银": "AG2608"}}
|
|
|
|
|
|
mock_db = MagicMock()
|
|
|
mock_db.query.return_value.filter.return_value.first.return_value = mock_config
|
|
|
|
|
|
storage = MagicMock()
|
|
|
storage.check_mysql.return_value = True
|
|
|
|
|
|
store = ConfigStore(
|
|
|
storage_manager=storage,
|
|
|
config_dir=tmp_path,
|
|
|
session_maker=_make_session_maker(mock_db),
|
|
|
)
|
|
|
|
|
|
result = store.get_config("symbols", {"futures": {}})
|
|
|
|
|
|
assert result == {"futures": {"沪银": "AG2608"}}
|
|
|
|
|
|
def test_get_config_reads_json_and_backfills_when_mysql_miss(self, tmp_path):
|
|
|
"""MySQL 未命中时读取 JSON 并回填数据库。"""
|
|
|
json_path = tmp_path / "symbols_config.json"
|
|
|
json_path.write_text(json.dumps({"futures": {"沪银": "AG2608"}}, ensure_ascii=False), encoding="utf-8")
|
|
|
|
|
|
mock_db = MagicMock()
|
|
|
mock_db.query.return_value.filter.return_value.first.return_value = None
|
|
|
|
|
|
storage = MagicMock()
|
|
|
storage.check_mysql.return_value = True
|
|
|
|
|
|
store = ConfigStore(
|
|
|
storage_manager=storage,
|
|
|
config_dir=tmp_path,
|
|
|
session_maker=_make_session_maker(mock_db),
|
|
|
)
|
|
|
|
|
|
result = store.get_config("symbols", {"futures": {}})
|
|
|
|
|
|
assert result == {"futures": {"沪银": "AG2608"}}
|
|
|
mock_db.add.assert_called_once()
|
|
|
mock_db.commit.assert_called_once()
|
|
|
|
|
|
def test_get_config_falls_back_to_json_when_mysql_unavailable(self, tmp_path):
|
|
|
"""MySQL 不可用时直接读取 JSON。"""
|
|
|
json_path = tmp_path / "symbols_config.json"
|
|
|
json_path.write_text(json.dumps({"futures": {"沪银": "AG2608"}}, ensure_ascii=False), encoding="utf-8")
|
|
|
|
|
|
storage = MagicMock()
|
|
|
storage.check_mysql.return_value = False
|
|
|
|
|
|
store = ConfigStore(
|
|
|
storage_manager=storage,
|
|
|
config_dir=tmp_path,
|
|
|
session_maker=None,
|
|
|
)
|
|
|
|
|
|
result = store.get_config("symbols", {"futures": {}})
|
|
|
|
|
|
assert result == {"futures": {"沪银": "AG2608"}}
|
|
|
|
|
|
def test_get_config_returns_fallback_when_no_mysql_and_no_json(self, tmp_path):
|
|
|
"""MySQL 和 JSON 都不存在时返回 fallback。"""
|
|
|
storage = MagicMock()
|
|
|
storage.check_mysql.return_value = False
|
|
|
|
|
|
store = ConfigStore(
|
|
|
storage_manager=storage,
|
|
|
config_dir=tmp_path,
|
|
|
session_maker=None,
|
|
|
)
|
|
|
|
|
|
result = store.get_config("symbols", {"futures": {"默认": "DF2609"}})
|
|
|
|
|
|
assert result == {"futures": {"默认": "DF2609"}}
|
|
|
|
|
|
|
|
|
class TestConfigStoreSet:
|
|
|
"""ConfigStore.set_config 测试。"""
|
|
|
|
|
|
def test_set_config_writes_to_mysql_and_json(self, tmp_path):
|
|
|
"""MySQL 可用时同时写入数据库和 JSON。"""
|
|
|
mock_existing = MagicMock()
|
|
|
|
|
|
mock_db = MagicMock()
|
|
|
mock_db.query.return_value.filter.return_value.first.return_value = mock_existing
|
|
|
|
|
|
storage = MagicMock()
|
|
|
storage.check_mysql.return_value = True
|
|
|
|
|
|
store = ConfigStore(
|
|
|
storage_manager=storage,
|
|
|
config_dir=tmp_path,
|
|
|
session_maker=_make_session_maker(mock_db),
|
|
|
)
|
|
|
|
|
|
result = store.set_config("symbols", {"futures": {"沪银": "AG2608"}})
|
|
|
|
|
|
assert result is True
|
|
|
assert mock_existing.config_value == {"futures": {"沪银": "AG2608"}}
|
|
|
mock_db.commit.assert_called_once()
|
|
|
|
|
|
json_path = tmp_path / "symbols_config.json"
|
|
|
assert json_path.exists()
|
|
|
assert json.loads(json_path.read_text(encoding="utf-8")) == {"futures": {"沪银": "AG2608"}}
|
|
|
|
|
|
def test_set_config_writes_only_json_when_mysql_unavailable(self, tmp_path):
|
|
|
"""MySQL 不可用时只写入 JSON。"""
|
|
|
storage = MagicMock()
|
|
|
storage.check_mysql.return_value = False
|
|
|
|
|
|
store = ConfigStore(
|
|
|
storage_manager=storage,
|
|
|
config_dir=tmp_path,
|
|
|
session_maker=None,
|
|
|
)
|
|
|
|
|
|
result = store.set_config("symbols", {"futures": {"沪银": "AG2608"}})
|
|
|
|
|
|
assert result is True
|
|
|
|
|
|
json_path = tmp_path / "symbols_config.json"
|
|
|
assert json_path.exists()
|
|
|
assert json.loads(json_path.read_text(encoding="utf-8")) == {"futures": {"沪银": "AG2608"}}
|
|
|
|
|
|
|
|
|
class TestConfigMigration:
|
|
|
"""配置迁移测试。"""
|
|
|
|
|
|
def test_migrate_skips_when_config_exists(self, tmp_path):
|
|
|
"""数据库已有配置时跳过迁移。"""
|
|
|
from app.config_migration import migrate_configs_to_mysql
|
|
|
|
|
|
with patch("app.config_migration.get_config_store") as mock_get_store:
|
|
|
store = MagicMock()
|
|
|
store._config_exists_in_mysql.return_value = True
|
|
|
store.storage_manager.check_mysql.return_value = True
|
|
|
store.session_maker = MagicMock()
|
|
|
mock_get_store.return_value = store
|
|
|
|
|
|
result = migrate_configs_to_mysql()
|
|
|
|
|
|
assert result is True
|
|
|
store._load_json.assert_not_called()
|
|
|
|
|
|
def test_migrate_loads_json_and_saves(self, tmp_path):
|
|
|
"""从 JSON 读取并保存到数据库。"""
|
|
|
from app.config_migration import migrate_configs_to_mysql
|
|
|
|
|
|
symbols_path = tmp_path / "symbols_config.json"
|
|
|
symbols_path.write_text(json.dumps({"futures": {"沪银": "AG2608"}}, ensure_ascii=False), encoding="utf-8")
|
|
|
|
|
|
ai_path = tmp_path / "ai_config.json"
|
|
|
ai_path.write_text(json.dumps({"models": []}, ensure_ascii=False), encoding="utf-8")
|
|
|
|
|
|
with patch("app.config_migration.get_config_store") as mock_get_store:
|
|
|
store = MagicMock()
|
|
|
store._config_exists_in_mysql.return_value = False
|
|
|
store.storage_manager.check_mysql.return_value = True
|
|
|
store.session_maker = MagicMock()
|
|
|
store.config_dir = tmp_path
|
|
|
store.config_files = {
|
|
|
"symbols": symbols_path,
|
|
|
"ai": ai_path,
|
|
|
}
|
|
|
store._load_json.side_effect = lambda key, default: json.loads(store.config_files[key].read_text(encoding="utf-8")) if store.config_files[key].exists() else default
|
|
|
store._save_to_mysql.return_value = True
|
|
|
mock_get_store.return_value = store
|
|
|
|
|
|
result = migrate_configs_to_mysql()
|
|
|
|
|
|
assert result is True
|
|
|
assert store._save_to_mysql.call_count == 2
|
