sanguo_vnpy_v2/tests/api/test_factor_routes.py

"""Tests for factor (投研) endpoints (S2): /factor/list, /ic-summary, /report."""
import pytest
from fastapi.testclient import TestClient

from sanguo_api.app import create_app
from sanguo_api.routes import set_orchestrator
from sanguo_api.auth import hash_password


class FakeReport:
    """Stand-in for FactorReport."""

    def __init__(self, ic_summary: dict, report_paths: dict, tears_paths: dict | None = None):
        self.ic_summary = ic_summary
        self.report_paths = report_paths
        self.tears_paths = tears_paths or {}


class FakeOrch:
    def __init__(self, raw):
        self._raw = raw

    def get_raw_result(self, task_id):
        return self._raw


@pytest.fixture(scope="module")
def client() -> TestClient:
    app = create_app(
        db_path="/tmp/test_fc_routes.db",
        auth_config={
            "username": "admin",
            "password_hash": hash_password("admin"),
            "jwt_secret": "test-secret",
            "expire_minutes": 60,
        },
        max_workers=1,
    )
    set_orchestrator(FakeOrch(FakeReport(
        ic_summary={"ma5": {"status": "success", "ic": {
            "1D": {"mean": -0.12, "std": 0.5, "icir": -0.24, "t_stat": -1.1, "count": 49},
        }}},
        report_paths={"ma5": "/tmp/__definitely_absent_ma5.html"},
    )))
    return TestClient(app)


@pytest.fixture(scope="module")
def token(client) -> str:
    return client.post("/api/v1/auth/login", json={"username": "admin", "password": "admin"}).json()["token"]


def test_factor_list_shape(client, token):
    r = client.get("/api/v1/factor/list", headers={"Authorization": f"Bearer {token}"})
    assert r.status_code == 200
    assert isinstance(r.json()["factors"], list)


def test_ic_summary(client, token):
    r = client.get("/api/v1/task/t/ic-summary", headers={"Authorization": f"Bearer {token}"})
    assert r.status_code == 200
    ic = r.json()["ic_summary"]
    assert "ma5" in ic
    assert ic["ma5"]["ic"]["1D"]["mean"] == -0.12


def test_report_bad_token_401(client):
    r = client.get("/api/v1/task/t/report/ma5?token=bad")
    assert r.status_code == 401


def test_report_file_absent_404(client, token):
    r = client.get(f"/api/v1/task/t/report/ma5?token={token}")
    assert r.status_code == 404


# —— tears JSON 端点(方案A) ——

def test_tears_json_no_token_401(client):
    r = client.get("/api/v1/task/t/tears/ma5")
    assert r.status_code == 401


def test_tears_json_absent_404(client, token):
    """fixture 的 FakeReport 无 tears_paths → 404(历史任务语义)."""
    r = client.get("/api/v1/task/t/tears/ma5", headers={"Authorization": f"Bearer {token}"})
    assert r.status_code == 404


def test_tears_json_served(client, token, tmp_path):
    """tears_paths 指向真实 JSON 文件 → 200 + application/json."""
    import json
    p = tmp_path / "ma5_tears.json"
    p.write_text(json.dumps({"factor": "ma5", "periods": {"1D": {"ic_mean": 0.06}}}),
                 encoding="utf-8")
    set_orchestrator(FakeOrch(FakeReport(
        ic_summary={"ma5": {"status": "success", "ic": {}}},
        report_paths={"ma5": "/tmp/__definitely_absent_ma5.html"},
        tears_paths={"ma5": str(p)},
    )))
    r = client.get("/api/v1/task/t/tears/ma5", headers={"Authorization": f"Bearer {token}"})
    assert r.status_code == 200
    assert r.headers["content-type"].startswith("application/json")
    assert r.json()["factor"] == "ma5"


def test_universe_pools(client, token, monkeypatch):
    """/factor/universe/pools → 池清单(依赖 monkeypatch,不连真库)."""
    import sanguo_factor.universe_pools as uni

    monkeypatch.setattr(uni, "list_pools",
                        lambda: [{"key": "000300", "name": "沪深300", "count": 300}])
    r = client.get("/api/v1/factor/universe/pools", headers={"Authorization": f"Bearer {token}"})
    assert r.status_code == 200
    assert r.json() == [{"key": "000300", "name": "沪深300", "count": 300}]


def test_universe_pool_and_search(client, token, monkeypatch):
    """/factor/universe/pool/{key} 与 /search → [{code,name}]."""
    import sanguo_factor.universe_pools as uni

    monkeypatch.setattr(uni, "pool_stocks",
                        lambda key: [{"code": "600519", "name": "贵州茅台"}] if key == "000300" else [])
    monkeypatch.setattr(uni, "search_stocks",
                        lambda q, limit=10: [{"code": "600519", "name": "贵州茅台"}] if "茅台" in q else [])
    h = {"Authorization": f"Bearer {token}"}
    r1 = client.get("/api/v1/factor/universe/pool/000300", headers=h)
    assert r1.status_code == 200 and r1.json()[0]["code"] == "600519"
    r2 = client.get("/api/v1/factor/universe/pool/999999", headers=h)
    assert r2.status_code == 200 and r2.json() == []
    r3 = client.get("/api/v1/factor/universe/search", params={"q": "茅台"}, headers=h)
    assert r3.status_code == 200 and len(r3.json()) == 1


# —— 因子进度心跳(_read_factor_progress) ——

def test_read_factor_progress_roundtrip(tmp_path, monkeypatch):
    import json, time
    from sanguo_api import routes as rt

    p = tmp_path / "factor_abc.progress"
    p.write_text(json.dumps({"stage": "analyze", "detail": "因子 1/2", "ts": time.time() - 5}),
                 encoding="utf-8")
    monkeypatch.setattr(rt, "_FACTOR_PROGRESS_DIR", str(tmp_path))
    d = rt._read_factor_progress("factor_abc")
    assert d is not None and d["stage"] == "analyze" and 4 <= d["age"] <= 15


def test_read_factor_progress_bad_input(tmp_path, monkeypatch):
    from sanguo_api import routes as rt

    # 非 factor 任务恒 None
    assert rt._read_factor_progress("cta_123") is None
    monkeypatch.setattr(rt, "_FACTOR_PROGRESS_DIR", str(tmp_path))
    # 文件不存在 → None
    assert rt._read_factor_progress("factor_missing") is None
    # 损坏 JSON → None(不影响 status 主链路)
    (tmp_path / "factor_bad.progress").write_text("{not json", encoding="utf-8")
    assert rt._read_factor_progress("factor_bad") is None