buffer_platform/tests/test_analysis_storage.py

"""Analysis 存储迁移测试：Redis 缓存 + MySQL/SQLite 降级。"""
import json
from datetime import datetime
from unittest.mock import MagicMock, patch

import pytest

from app.storage_manager import StorageManager


class TestStorageManagerCache:
    """StorageManager 通用缓存方法测试。"""

    def test_cache_get_returns_redis_value_on_hit(self):
        """Redis 命中时返回反序列化数据。"""
        redis_client = MagicMock()
        redis_client.get.return_value = json.dumps({"symbol": "AG2606", "score": 85})
        manager = StorageManager(redis_client=redis_client, mysql_engine=MagicMock())

        result = manager.cache_get("analysis:ai_analysis_cache:1")

        assert result == {"symbol": "AG2606", "score": 85}
        redis_client.get.assert_called_once_with("analysis:ai_analysis_cache:1")

    def test_cache_get_returns_none_when_redis_unavailable(self):
        """Redis 不可用时返回 None。"""
        manager = StorageManager(redis_client=None, mysql_engine=MagicMock())

        result = manager.cache_get("analysis:ai_analysis_cache:1")

        assert result is None

    def test_cache_get_returns_none_on_miss(self):
        """Redis 未命中时返回 None。"""
        redis_client = MagicMock()
        redis_client.get.return_value = None
        manager = StorageManager(redis_client=redis_client, mysql_engine=MagicMock())

        result = manager.cache_get("analysis:ai_analysis_cache:1")

        assert result is None

    def test_cache_set_writes_to_redis_with_ttl(self):
        """cache_set 使用 TTL 写入 Redis。"""
        redis_client = MagicMock()
        manager = StorageManager(redis_client=redis_client, mysql_engine=MagicMock())
        value = {"symbol": "AG2606", "score": 85}

        manager.cache_set("analysis:ai_analysis_cache:1", value, ttl=300)

        redis_client.setex.assert_called_once_with(
            "analysis:ai_analysis_cache:1", 300, json.dumps(value, ensure_ascii=False)
        )

    def test_cache_set_returns_false_when_redis_unavailable(self):
        """Redis 不可用时 cache_set 返回 False。"""
        manager = StorageManager(redis_client=None, mysql_engine=MagicMock())

        result = manager.cache_set("analysis:ai_analysis_cache:1", {"score": 85}, ttl=300)

        assert result is False

    def test_cache_delete_removes_redis_key(self):
        """cache_delete 删除 Redis 键。"""
        redis_client = MagicMock()
        manager = StorageManager(redis_client=redis_client, mysql_engine=MagicMock())

        manager.cache_delete("analysis:ai_analysis_cache:1")

        redis_client.delete.assert_called_once_with("analysis:ai_analysis_cache:1")


class TestAnalysisDbSwitching:
    """analysis_db 动态 Session 切换测试。"""

    def test_get_analysis_db_returns_mysql_session_when_mysql_available(self):
        """MySQL 可用时返回 MySQL Session。"""
        from app.analysis_db import get_analysis_db

        mysql_engine = MagicMock()
        mock_session = MagicMock()
        mock_sessionmaker = MagicMock(return_value=mock_session)
        mock_session.__enter__ = MagicMock(return_value=mock_session)
        mock_session.__exit__ = MagicMock(return_value=False)

        with patch("app.analysis_db.MySQLSessionLocal", mock_sessionmaker):
            with patch("app.analysis_db.get_storage_manager") as mock_sm:
                sm = MagicMock()
                sm.check_mysql.return_value = True
                sm.mysql_engine = mysql_engine
                mock_sm.return_value = sm

                db = next(get_analysis_db())

                assert db is mock_session

    def test_get_analysis_db_returns_sqlite_session_when_mysql_unavailable(self):
        """MySQL 不可用时返回 SQLite Session。"""
        from app.analysis_db import get_analysis_db

        sqlite_engine = MagicMock()
        mock_session = MagicMock()
        mock_sessionmaker = MagicMock(return_value=mock_session)
        mock_session.__enter__ = MagicMock(return_value=mock_session)
        mock_session.__exit__ = MagicMock(return_value=False)

        with patch("app.analysis_db.AnalysisSessionLocal", mock_sessionmaker):
            with patch("app.analysis_db.get_storage_manager") as mock_sm:
                sm = MagicMock()
                sm.check_mysql.return_value = False
                mock_sm.return_value = sm

                db = next(get_analysis_db())

                assert db is mock_session


class TestAnalysisMigration:
    """Analysis 数据迁移测试。"""

    def test_migrate_skips_when_mysql_has_data(self):
        """MySQL 已有数据时跳过迁移。"""
        from app.analysis_migration import migrate_analysis_sqlite_to_mysql

        with patch("app.analysis_migration.mysql_engine", MagicMock()):
            with patch("app.analysis_migration._count_mysql_analysis_records") as mock_count:
                mock_count.return_value = 10

                result = migrate_analysis_sqlite_to_mysql()

                assert result is True
                mock_count.assert_called_once()

    def test_migrate_moves_data_from_sqlite_to_mysql(self):
        """SQLite 数据正确迁移到 MySQL。"""
        from app.analysis_migration import migrate_analysis_sqlite_to_mysql

        with patch("app.analysis_migration.mysql_engine", MagicMock()):
            with patch("app.analysis_migration._count_mysql_analysis_records") as mock_count:
                mock_count.return_value = 0
                with patch("app.analysis_migration._read_sqlite_analysis_data") as mock_read:
                    mock_read.return_value = {
                        "ai_analysis_cache": [{"id": 1, "symbol": "AG2606"}]
                    }
                    with patch("app.analysis_migration._write_mysql_analysis_data") as mock_write:
                        mock_write.return_value = True

                        result = migrate_analysis_sqlite_to_mysql()

                        assert result is True
                        mock_write.assert_called_once_with({"ai_analysis_cache": [{"id": 1, "symbol": "AG2606"}]})