buffer_platform/buffer_platform_v2/tests/conftest.py

"""测试配置与公共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()