zhixing-system/zhixing-server/tests/test_selection_http.py

"""HTTP contracts for triggering and querying persisted selection runs."""

from datetime import date
from decimal import Decimal

from fastapi.testclient import TestClient

from zhixing_server.bootstrap.app import create_app
from zhixing_server.modules.selection.application.run import PreparedSelectionRun
from zhixing_server.modules.selection.domain.models import SelectionSignal, ZhixingB1Category
from zhixing_server.modules.selection.domain.runs import (
    SelectionExecutionSource,
    SelectionRerunRequired,
    SelectionRun,
    SelectionRunInProgress,
    SelectionStock,
)
from zhixing_server.modules.selection.infrastructure.postgres_reader import (
    SelectionMarketDataNotReady,
)
from zhixing_server.modules.selection.presentation.http import get_selection_service

TARGET = date(2026, 8, 8)


class FakeSelectionService:
    def __init__(self, run: SelectionRun | None = None) -> None:
        self.run = run
        self.executed = False
        self.mode = "ok"

    def prepare(
        self,
        strategy: str,
        target_trade_date: date,
        *,
        rerun: bool,
    ) -> PreparedSelectionRun:
        if self.mode == "in_progress":
            raise SelectionRunInProgress("already running")
        if self.mode == "rerun_required":
            raise SelectionRerunRequired("confirm rerun")
        if self.mode == "market_data_not_ready":
            raise SelectionMarketDataNotReady("market data is not ready")
        run = self.run or _run("run-http", "running")
        return PreparedSelectionRun(
            run=run,
            source=SelectionExecutionSource(
                market_sync_batch_id="market-run-1",
                target_trade_date=target_trade_date,
                target_count=1,
                valid_count=1,
                coverage=Decimal("1"),
                stocks=(SelectionStock("000001.SZ", "平安银行"),),
            ),
        )

    def execute(self, prepared: PreparedSelectionRun) -> None:
        self.executed = True

    def get_run(self, run_id: str) -> SelectionRun | None:
        return self.run if self.run and self.run.id == run_id else None

    def get_latest(
        self,
        strategy: str,
        target_trade_date: date | None = None,
    ) -> SelectionRun | None:
        if self.run is None:
            return None
        if target_trade_date is not None and self.run.target_trade_date != target_trade_date:
            return None
        return self.run


def _run(run_id: str, status: str) -> SelectionRun:
    signal = SelectionSignal(
        ts_code="000001.SZ",
        name="平安银行",
        target_trade_date=TARGET,
        strategy="zhixing_b1",
        category=ZhixingB1Category.ORIGINAL_B1,
        close=10.5,
        details={"j": 12.0},
    )
    from zhixing_server.modules.selection.domain.runs import SelectionRunItem

    return SelectionRun(
        id=run_id,
        strategy="zhixing_b1",
        target_trade_date=TARGET,
        market_sync_batch_id="market-run-1",
        status=status,  # type: ignore[arg-type]
        target_count=1,
        eligible_count=1,
        evaluated_count=1,
        selected_stock_count=1,
        signal_count=1,
        failed_count=0,
        coverage=Decimal("1"),
        items=(
            SelectionRunItem(
                ts_code="000001.SZ",
                name="平安银行",
                status="selected",
                signal_count=1,
                signals=(signal,),
            ),
        ),
        signals=(signal,),
    )


def _client(service: FakeSelectionService) -> TestClient:
    app = create_app()
    app.dependency_overrides[get_selection_service] = lambda: service
    return TestClient(app)


def test_trigger_returns_accepted_run_and_schedules_execution() -> None:
    service = FakeSelectionService()

    response = _client(service).post(
        "/api/v1/selection/runs",
        json={
            "strategy": "zhixing_b1",
            "target_trade_date": "2026-08-08",
            "rerun": False,
        },
    )

    assert response.status_code == 202
    assert response.json()["status"] == "running"
    assert response.json()["target_trade_date"] == "2026-08-08"
    assert service.executed is True


def test_trigger_requires_explicit_rerun_confirmation() -> None:
    service = FakeSelectionService()
    service.mode = "rerun_required"

    response = _client(service).post(
        "/api/v1/selection/runs",
        json={"strategy": "zhixing_b1", "target_trade_date": "2026-08-08"},
    )

    assert response.status_code == 409
    assert response.json()["detail"]["code"] == "rerun_confirmation_required"


def test_trigger_rejects_a_duplicate_running_request() -> None:
    service = FakeSelectionService()
    service.mode = "in_progress"

    response = _client(service).post(
        "/api/v1/selection/runs",
        json={"strategy": "zhixing_b1", "target_trade_date": "2026-08-08"},
    )

    assert response.status_code == 409
    assert response.json()["detail"]["code"] == "run_in_progress"


def test_trigger_rejects_unqualified_market_data() -> None:
    service = FakeSelectionService()
    service.mode = "market_data_not_ready"

    response = _client(service).post(
        "/api/v1/selection/runs",
        json={"strategy": "zhixing_b1", "target_trade_date": "2026-08-08"},
    )

    assert response.status_code == 422
    assert response.json()["detail"]["code"] == "market_data_not_ready"


def test_query_returns_no_data_without_fabricating_a_result() -> None:
    response = _client(FakeSelectionService()).get(
        "/api/v1/selection/results?strategy=zhixing_b1&target_trade_date=2026-08-08"
    )

    assert response.status_code == 200
    assert response.json()["status"] == "no_data"
    assert response.json()["signals"] == []


def test_query_returns_persisted_signal_details() -> None:
    response = _client(FakeSelectionService(_run("run-http", "success"))).get(
        "/api/v1/selection/results?strategy=zhixing_b1&target_trade_date=2026-08-08"
    )

    assert response.status_code == 200
    body = response.json()
    assert body["run_id"] == "run-http"
    assert body["signal_count"] == 1
    assert body["signals"][0]["category"] == "zhixing_b1_original_b1"
    assert body["signals"][0]["details"] == {"j": 12.0}


def test_run_polling_returns_the_persisted_terminal_result() -> None:
    response = _client(FakeSelectionService(_run("run-http", "success"))).get(
        "/api/v1/selection/runs/run-http"
    )

    assert response.status_code == 200
    assert response.json()["status"] == "success"
    assert response.json()["signals"][0]["category"] == "zhixing_b1_original_b1"