yuxuanhui
273 lines
8.4 KiB
Python
273 lines
8.4 KiB
Python
"""HTTP contracts for triggering and querying persisted selection runs."""
|
|
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
import zhixing_server.modules.selection.presentation.http as selection_http
|
|
from zhixing_server.bootstrap.app import create_app
|
|
from zhixing_server.bootstrap.config import Settings
|
|
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,
|
|
SelectionResultQuery,
|
|
SelectionRun,
|
|
SelectionRunInProgress,
|
|
SelectionStock,
|
|
)
|
|
from zhixing_server.modules.selection.infrastructure.postgres_reader import (
|
|
SelectionMarketDataNotReady,
|
|
)
|
|
from zhixing_server.modules.selection.presentation.http import (
|
|
get_selection_postgres_pool,
|
|
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"
|
|
self.last_query: SelectionResultQuery | None = None
|
|
|
|
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,
|
|
*,
|
|
query: SelectionResultQuery | None = None,
|
|
) -> SelectionRun | None:
|
|
self.last_query = query
|
|
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,
|
|
*,
|
|
query: SelectionResultQuery | None = None,
|
|
) -> SelectionRun | None:
|
|
self.last_query = query
|
|
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_selection_dependency_reuses_one_bounded_pool(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(selection_http, "_SELECTION_POOL_CACHE", {})
|
|
settings = Settings(database_url="postgresql://test", selection_max_workers=4)
|
|
|
|
first = get_selection_postgres_pool(settings)
|
|
second = get_selection_postgres_pool(settings)
|
|
|
|
assert first is second
|
|
assert first.max_connections == 6
|
|
|
|
|
|
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["page"] == 1
|
|
assert body["page_size"] == 10
|
|
assert body["signals_total"] == 1
|
|
assert body["signals"][0]["category"] == "zhixing_b1_original_b1"
|
|
assert body["signals"][0]["details"] == {"j": 12.0}
|
|
|
|
|
|
def test_query_forwards_pagination_and_filters() -> None:
|
|
service = FakeSelectionService(_run("run-http", "success"))
|
|
|
|
response = _client(service).get(
|
|
"/api/v1/selection/results",
|
|
params={
|
|
"strategy": "zhixing_b1",
|
|
"page": 2,
|
|
"page_size": 5,
|
|
"search": " 平安银行 ",
|
|
"category": "original",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert service.last_query == SelectionResultQuery(
|
|
page=2,
|
|
page_size=5,
|
|
search="平安银行",
|
|
category="original",
|
|
)
|
|
assert response.json()["page"] == 2
|
|
assert response.json()["page_size"] == 5
|
|
|
|
|
|
def test_query_rejects_invalid_page_size() -> None:
|
|
response = _client(FakeSelectionService()).get(
|
|
"/api/v1/selection/results",
|
|
params={"strategy": "zhixing_b1", "page_size": 101},
|
|
)
|
|
|
|
assert response.status_code == 422
|
|
|
|
|
|
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"
|