69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
from datetime import UTC, datetime
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
from app.core.fund_contracts import FundQuoteQuery
|
|
from app.service.fund_quote_service import FundQuoteService
|
|
|
|
|
|
class StubAdapter:
|
|
async def fetch_names(self, codes: list[str]) -> dict[str, str]:
|
|
return {code: f"基金-{code}" for code in codes}
|
|
|
|
async def fetch_quotes(self, codes: list[str]) -> dict[str, dict[str, object]]:
|
|
return {code: {"nav": Decimal("1.23"), "daily_change": Decimal("0.82")} for code in codes}
|
|
|
|
async def fetch_history(self, code: str, target_date: object) -> dict[str, object]:
|
|
return {
|
|
"nav": Decimal("1.20"), "nav_date": target_date,
|
|
"daily_change": Decimal("0.10"), "degraded": False,
|
|
}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_service_uses_intraday_quote_during_market_hours() -> None:
|
|
service = FundQuoteService(StubAdapter())
|
|
result = await service.query(
|
|
FundQuoteQuery(fund_codes=("159511",)),
|
|
now=datetime(2026, 9, 9, 2, 0, tzinfo=UTC),
|
|
)
|
|
assert result[0].nav == Decimal("1.23")
|
|
assert result[0].is_intraday is True
|
|
assert result[0].degraded is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_service_uses_history_outside_market_hours() -> None:
|
|
service = FundQuoteService(StubAdapter())
|
|
result = await service.query(
|
|
FundQuoteQuery(fund_codes=("159511",)),
|
|
now=datetime(2026, 9, 9, 8, 0, tzinfo=UTC),
|
|
)
|
|
assert result[0].nav == Decimal("1.20")
|
|
assert result[0].is_intraday is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_service_filters_type_and_rejects_unknown_type() -> None:
|
|
service = FundQuoteService(StubAdapter())
|
|
result = await service.query(FundQuoteQuery(fund_type="债券型", limit=1))
|
|
assert len(result) == 1
|
|
assert result[0].fund_type == "债券型"
|
|
with pytest.raises(ValueError, match="基金类型"):
|
|
await service.query(FundQuoteQuery(fund_type="未知"))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_intraday_missing_live_quote_is_degraded() -> None:
|
|
class NoLive(StubAdapter):
|
|
async def fetch_quotes(self, codes: list[str]) -> dict[str, dict[str, object]]:
|
|
return {}
|
|
|
|
result = await FundQuoteService(NoLive()).query(
|
|
FundQuoteQuery(fund_codes=("159511",)),
|
|
now=datetime(2026, 9, 9, 2, 0, tzinfo=UTC),
|
|
)
|
|
assert result[0].degraded is True
|
|
assert result[0].quote_source == "degraded"
|