"""测试配置与公共Fixtures""" from collections.abc import AsyncGenerator, Generator from contextlib import asynccontextmanager from datetime import datetime, timedelta from typing import Any from unittest.mock import AsyncMock, MagicMock import pytest from fastapi.testclient import TestClient from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker # --------------------------------------------------------------------------- # Mock 结果辅助类 # --------------------------------------------------------------------------- class MockScalarResult: """模拟 SQLAlchemy ScalarResult""" def __init__(self, items: list[Any] | None = None): self._items = items or [] def all(self) -> list[Any]: return list(self._items) def first(self) -> Any: return self._items[0] if self._items else None def one_or_none(self) -> Any: if len(self._items) == 1: return self._items[0] if len(self._items) == 0: return None raise Exception("Multiple rows found") def __iter__(self): return iter(self._items) class MockResult: """模拟 SQLAlchemy Result""" def __init__(self, items: list[Any] | None = None, scalar: Any = None): self._items = items or [] self._scalar = scalar def scalars(self) -> MockScalarResult: return MockScalarResult(self._items) def scalar_one_or_none(self) -> Any: if self._scalar is not None: return self._scalar if len(self._items) == 1: return self._items[0] if len(self._items) == 0: return None return self._items[0] def scalar(self) -> Any: return self._scalar def first(self) -> Any: return self._items[0] if self._items else None def all(self) -> list[Any]: return list(self._items) def make_mock_db(return_items: list[Any] | None = None, return_scalar: Any = None) -> AsyncMock: """创建一个模拟的异步数据库会话。 Args: return_items: execute() 返回的列表数据 return_scalar: execute() 返回的标量数据 """ mock_db = AsyncMock() result = MockResult(items=return_items, scalar=return_scalar) mock_db.execute = AsyncMock(return_value=result) mock_db.flush = AsyncMock() mock_db.commit = AsyncMock() mock_db.rollback = AsyncMock() mock_db.refresh = AsyncMock() mock_db.delete = MagicMock() mock_db.add = MagicMock() return mock_db # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture def db_session() -> Generator[Session, None, None]: """创建测试用数据库会话（内存SQLite）""" engine = create_engine("sqlite:///:memory:") from app.models.analysis import Base as AnalysisBase from app.models.market import Base as MarketBase from app.models.user import Base as UserBase MarketBase.metadata.create_all(engine) AnalysisBase.metadata.create_all(engine) UserBase.metadata.create_all(engine) session_factory = sessionmaker(bind=engine) session = session_factory() yield session session.close() @asynccontextmanager async def _mock_lifespan(app: Any) -> AsyncGenerator[None, None]: """模拟应用生命周期，跳过真实数据库/调度器初始化。""" yield @pytest.fixture def mock_session() -> AsyncMock: """创建一个可配置的 mock 数据库会话""" return make_mock_db() @pytest.fixture def client(mock_session: AsyncMock) -> Generator[TestClient, None, None]: """创建FastAPI测试客户端（mock lifespan + mock get_db）。""" from app.database import get_db from app.main import app app.router.lifespan_context = _mock_lifespan async def _override_get_db() -> AsyncGenerator[Any, None]: yield mock_session app.dependency_overrides[get_db] = _override_get_db with TestClient(app) as c: yield c app.dependency_overrides.clear() @pytest.fixture def client_factory() -> Any: """创建一个可以自定义 mock_session 的客户端工厂""" from app.database import get_db from app.main import app app.router.lifespan_context = _mock_lifespan def _create_client(return_items=None, return_scalar=None): mock_db = make_mock_db(return_items=return_items, return_scalar=return_scalar) async def _override_get_db() -> AsyncGenerator[Any, None]: yield mock_db app.dependency_overrides[get_db] = _override_get_db return TestClient(app), mock_db yield _create_client app.dependency_overrides.clear() @pytest.fixture def mock_akshare() -> Any: """Mock akshare数据采集，返回样本K线数据""" mock = AsyncMock() mock.fetch_symbol_data.return_value = _build_sample_result() return mock @pytest.fixture def sample_kline_data() -> list[dict[str, Any]]: """样本K线数据""" base_date = datetime(2025, 1, 2) data: list[dict[str, Any]] = [] for i in range(100): data.append( { "date": (base_date + timedelta(days=i)).strftime("%Y-%m-%d"), "open": 4000.0 + i * 2, "high": 4050.0 + i * 2, "low": 3980.0 + i * 2, "close": 4020.0 + i * 2, "volume": 10000 + i * 100, "hold": 50000 + i * 50, } ) return data @pytest.fixture def sample_symbols_config() -> list[dict[str, Any]]: """样本品种配置""" return [ { "symbol": "rb2510", "variety": "螺纹钢", "exchange": "SHFE", "enabled": True, "periods": ["1d", "4h", "1h"], }, { "symbol": "i2510", "variety": "铁矿石", "exchange": "DCE", "enabled": True, "periods": ["1d", "4h"], }, { "symbol": "MA510", "variety": "甲醇", "exchange": "CZCE", "enabled": True, "periods": ["1d"], }, ] # --------------------------------------------------------------------------- # 辅助函数 # --------------------------------------------------------------------------- def _build_sample_kline_df() -> Any: """构建样本K线 DataFrame（供 mock_akshare 使用）""" try: import pandas as pd except ImportError: return None base_date = datetime(2025, 1, 2) rows = [] for i in range(100): rows.append( { "date": (base_date + timedelta(days=i)).strftime("%Y%m%d"), "open": 4000.0 + i * 2, "high": 4050.0 + i * 2, "low": 3980.0 + i * 2, "close": 4020.0 + i * 2, "volume": 10000 + i * 100, "hold": 50000 + i * 50, } ) return pd.DataFrame(rows) def _build_sample_result() -> dict[str, Any]: """构建样本采集结果。""" return { "symbol": "rb2510", "type": "futures", "current_price": 4020.0, "timestamp": "2025-01-15T10:00:00", "timeframes": { "daily": [ { "datetime": "2025-01-10", "open": 4000.0, "high": 4050.0, "low": 3980.0, "close": 4020.0, "volume": 10000, } ], }, } def make_mock_user( user_id: int = 1, username: str = "admin", role: str = "admin", email: str = "admin@test.com", ) -> Any: """创建一个模拟用户对象。""" class MockUser: def __init__(self) -> None: self.id = user_id self.username = username self.role = role self.email = email self.is_active = True self.created_at = datetime(2025, 1, 1) self.last_login = datetime(2025, 1, 15) return MockUser()