32 lines
1.1 KiB
Python
32 lines
1.1 KiB
Python
"""Application-level state and port mapping tests."""
|
|
|
|
from datetime import date
|
|
|
|
from zhixing_server.modules.selection.application.evaluate import EvaluateZhixingB1
|
|
from zhixing_server.modules.selection.domain.models import StockHistory
|
|
from zhixing_server.modules.selection.domain.ports import MarketDataReaderError
|
|
|
|
TARGET = date(2024, 1, 2)
|
|
|
|
|
|
class EmptyReader:
|
|
def load_history(self, ts_code: str, target_trade_date: date) -> StockHistory:
|
|
return StockHistory(ts_code=ts_code, name="", bars=())
|
|
|
|
|
|
class FailingReader:
|
|
def load_history(self, ts_code: str, target_trade_date: date) -> StockHistory:
|
|
raise MarketDataReaderError(f"database unavailable for {ts_code}")
|
|
|
|
|
|
def test_evaluate_maps_reader_error_to_data_error() -> None:
|
|
result = EvaluateZhixingB1(FailingReader()).execute("000001.SZ", TARGET)
|
|
assert result.status == "data_error"
|
|
assert result.signals == ()
|
|
assert "000001.SZ" in (result.reason or "")
|
|
|
|
|
|
def test_evaluate_distinguishes_missing_target_from_reader_error() -> None:
|
|
result = EvaluateZhixingB1(EmptyReader()).execute("000001.SZ", TARGET)
|
|
assert result.status == "missing_target_bar"
|