Chan/web/tests/test_analyze_contract.py

"""加深 /api/analyze 相关契约 —— mock 行情 + analyze_chan 关键字段快照。"""
from __future__ import annotations

import json
import sys
from pathlib import Path
from unittest.mock import patch

import pandas as pd
import pytest

ROOT = Path(__file__).resolve().parents[2]
WEB_ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
sys.path.insert(0, str(WEB_ROOT))

from tests.helpers import make_ohlcv  # noqa: E402


_CONTRACT_DOC = json.loads(
    (Path(__file__).resolve().parent / "fixtures" / "analyze_contract_keys.json").read_text(encoding="utf-8")
)
CONTRACT_KEYS = (
    _CONTRACT_DOC["required"]
    if isinstance(_CONTRACT_DOC, dict) and "required" in _CONTRACT_DOC
    else _CONTRACT_DOC
)

# analyze_chan 直接返回的对象字段（未序列化前）
ANALYZE_CHAN_KEYS = {
    "klc_list",
    "klu_list",
    "bi_list",
    "seg_list",
    "zs_list",
    "bi_zs_list",
    "bsp_list",
    "fast_bsp_list",
    "klc_fx_info",
    "chan_macd",
    "ema52_dict",
}

CHAN_MACD_SERIALIZED_KEYS = {
    "seg_list",
    "unittf_list",
    "histset_list",
    "high_position_list",
    "high_empty_list",
    "low_position_list",
    "low_empty_list",
    "return_zero_list",
    "cross0_up_list",
    "cross0_down_list",
    "klu_list",
}


def test_analyze_route_registered():
    from app import app

    rules = {r.rule for r in app.url_map.iter_rules()}
    assert "/api/analyze" in rules
    assert "/api/klines/recent" in rules
    assert "/api/chart_metadata" in rules
    assert "/" in rules
    assert "/chan_tv" in rules


def test_klines_recent_returns_tail_only():
    from app import app

    df = make_ohlcv(n=30)
    with patch("api.analyze.get_kl_data", return_value=df):
        client = app.test_client()
        resp = client.get(
            "/api/klines/recent",
            query_string={"symbol": "BTC/USDT:USDT", "timeframe": "5m", "limit": 2},
        )
    assert resp.status_code == 200
    body = resp.get_json()
    assert body.get("partial") is True
    assert body.get("limit") == 2
    assert isinstance(body.get("kline_data"), list)
    assert len(body["kline_data"]) == 2
    assert "bi_list" not in body


def test_contract_keys_stable():
    assert "bi_list" in CONTRACT_KEYS and "seg_list" in CONTRACT_KEYS
    for k in ("kline_data", "macd", "zs_list", "bsp_list", "fast_bsp_list", "chan_macd"):
        assert k in CONTRACT_KEYS


def test_analyze_chan_keys_on_fixture():
    from services.runtime import add_indicators, analyze_chan

    df = add_indicators(make_ohlcv(400))
    result = analyze_chan(df, symbol="TEST/USDT:USDT", timeframe="5m")
    assert set(result.keys()) == ANALYZE_CHAN_KEYS
    assert isinstance(result["bi_list"], list)
    assert isinstance(result["seg_list"], list)
    assert isinstance(result["chan_macd"], dict)
    for k in ("seg_list", "unittf_list", "histset_list"):
        assert k in result["chan_macd"]


def test_serialize_chan_macd_shape():
    from pytz import timezone

    from services.runtime import add_indicators, analyze_chan, serialize_chan_macd_data

    df = add_indicators(make_ohlcv(200))
    result = analyze_chan(df)
    serialized = serialize_chan_macd_data(result["chan_macd"], timezone("Asia/Shanghai"))
    assert set(serialized.keys()) == CHAN_MACD_SERIALIZED_KEYS
    json.dumps(serialized)


def test_analyze_http_contract_with_mocked_kl():
    """Flask 测试客户端：mock get_kl_data，断言响应含契约关键字段。"""
    from app import app
    from services.runtime import add_indicators

    df = add_indicators(make_ohlcv(300))
    df = df.copy()
    if "timestamp" not in df.columns:
        df["timestamp"] = (pd.to_datetime(df["date"]).astype("int64") // 10**6).astype("int64")

    with patch("api.analyze.get_kl_data", return_value=df):
        client = app.test_client()
        resp = client.get(
            "/api/analyze",
            query_string={
                "symbol": "BTC/USDT:USDT",
                "timeframe": "5m",
                "timezone": "Asia/Shanghai",
            },
        )
    assert resp.status_code == 200, resp.data[:500]
    payload = resp.get_json()
    assert payload is not None and "error" not in payload
    missing = [k for k in CONTRACT_KEYS if k not in payload]
    assert not missing, f"missing contract keys: {missing}"