张胜宇
da1d14cb31
- 分支内容对齐 qyqy_develop 3e24033，树完全一致（同步后 git diff 为空） - 覆盖本轮 W27 交付：L0 表层判定层 + 出口 E6 行情 + 收益过滤槽位白名单 + 免责声明分档 + 金标扩容至 55 条（全绿）+ 配置版本 244 已发布 - 新增 app/core/exit_codes.py、app/service/fund_trend_service.py 及 3 个测试文件 - 新增 开发文档\D3.9-客服Agent智能路由与行情出口设计-2026-09-21.md - 基线：e239eb7（2026-09-17 品牌口径统一快照），本提交为其直接后继
240 lines
9.1 KiB
Python
240 lines
9.1 KiB
Python
from datetime import date
|
||
from decimal import Decimal
|
||
from types import SimpleNamespace
|
||
from typing import Any
|
||
|
||
import pytest
|
||
|
||
from app.core.contracts import RequestContext
|
||
from app.core.fund_contracts import FundTrendQuery
|
||
from app.model.fund import FundNavHistory
|
||
from app.service import fund_trend_service
|
||
from app.service.fund_trend_service import (
|
||
LISTED_STATUS,
|
||
NAV_SOURCE,
|
||
query_fund_trend_tool,
|
||
summarize_trend,
|
||
)
|
||
|
||
|
||
def _series(days: int, *, start: str = "1.0000", step: str = "0.0100") -> list[dict[str, str]]:
|
||
points: list[dict[str, str]] = []
|
||
nav = Decimal(start)
|
||
day = date(2026, 1, 2)
|
||
for _ in range(days):
|
||
points.append({"nav_date": day.isoformat(), "nav": str(nav)})
|
||
nav += Decimal(step)
|
||
day = date.fromordinal(day.toordinal() + 1)
|
||
return points
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 一、纯函数：`summarize_trend`
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
def test_summarize_trend_reports_no_series_for_empty_input() -> None:
|
||
"""**没有净值序列是一种结论，不是一次失败**（`INV-7` 靠它说"无可奉告"）。"""
|
||
assert summarize_trend([]) == {
|
||
"found": False, "reason": "no_nav_series", "source": NAV_SOURCE,
|
||
}
|
||
|
||
|
||
def test_summarize_trend_skips_unparsable_points() -> None:
|
||
"""脏行（日期解析不了 / 值是 `--`）**跳过**，不能让一条脏数据毁掉整条序列。"""
|
||
result = summarize_trend([
|
||
{"nav_date": "not-a-date", "nav": "1.0"},
|
||
{"nav_date": "2026-03-02", "nav": "--"},
|
||
{"nav_date": "2026-03-02", "nav": "1.5000"},
|
||
])
|
||
assert result["found"] is True
|
||
assert result["series_points"] == 1
|
||
assert result["latest_nav"] == "1.5000"
|
||
|
||
|
||
def test_summarize_trend_orders_points_regardless_of_input_order() -> None:
|
||
"""入参顺序不限：**最新一档取日期最大的那条**，不是数组的最后一个。"""
|
||
result = summarize_trend([
|
||
{"nav_date": "2026-03-05", "nav": "1.2000"},
|
||
{"nav_date": "2026-03-01", "nav": "1.0000"},
|
||
{"nav_date": "2026-03-03", "nav": "1.1000"},
|
||
])
|
||
assert result["latest_nav_date"] == "2026-03-05"
|
||
assert result["from_date"] == "2026-03-01"
|
||
|
||
|
||
def test_summarize_trend_computes_intervals_high_and_low() -> None:
|
||
result = summarize_trend(_series(6))
|
||
assert result["found"] is True
|
||
assert result["source"] == NAV_SOURCE
|
||
names = [row["name"] for row in result["intervals"]]
|
||
# 只有 6 个点：够到 5 日窗口，够不到 20/60/120 —— 不够的窗口**不编**。
|
||
assert names == ["近 5 个净值日"]
|
||
row = result["intervals"][0]
|
||
assert row["trading_days"] == 4
|
||
assert row["start_nav"] == "1.0100"
|
||
assert row["end_nav"] == "1.0500"
|
||
assert row["change_pct"] == "3.96"
|
||
assert result["high"] == {"nav": "1.0500", "nav_date": "2026-01-07"}
|
||
assert result["low"] == {"nav": "1.0000", "nav_date": "2026-01-02"}
|
||
|
||
|
||
def test_summarize_trend_change_pct_is_signed_for_down_moves() -> None:
|
||
result = summarize_trend(_series(6, step="-0.0100"))
|
||
# 5 日窗口：起点 0.9900 → 终点 0.9500 ⇒ (0.95-0.99)/0.99 = -4.04%
|
||
assert result["intervals"][0]["change_pct"] == "-4.04"
|
||
|
||
|
||
def test_listed_status_matches_public_product_service() -> None:
|
||
"""`LISTED_STATUS` 与 `public_product_service` **必须逐字相同**。
|
||
|
||
本模块**刻意不 import** 那个常量（会形成
|
||
`public_product_service → admin_service → agent.bootstrap → 本模块` 的循环导入，
|
||
实测 `partially initialized module`）。于是"两处同值"这件事只能靠测试钉住 ——
|
||
否则哪天有人改了其中一处，行情出口会静默查不到任何产品（返回"无公开净值序列"），
|
||
而那看起来像数据问题、不像代码问题。
|
||
"""
|
||
from app.service.public_product_service import LISTED_STATUS as public_listed_status
|
||
|
||
assert LISTED_STATUS == public_listed_status
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 二、工具处理器：`query_fund_trend_tool`（用假 session 走真实 `_resolve_product`）
|
||
# --------------------------------------------------------------------------- #
|
||
|
||
class _FakeScalars:
|
||
def __init__(self, items: list[Any]) -> None:
|
||
self._items = items
|
||
|
||
def all(self) -> list[Any]:
|
||
return list(self._items)
|
||
|
||
def first(self) -> Any:
|
||
return self._items[0] if self._items else None
|
||
|
||
|
||
class _FakeResult:
|
||
def __init__(self, items: list[Any]) -> None:
|
||
self._items = items
|
||
|
||
def scalars(self) -> _FakeScalars:
|
||
return _FakeScalars(self._items)
|
||
|
||
|
||
class _FakeSession:
|
||
"""按**实体类型**分发假结果，并记录产品查询次数。
|
||
|
||
`_resolve_product` 的产品查询最多两次（先精确、后 `contains` 模糊），
|
||
用计数器模拟这两步，而不是去复刻 SQL 语义。
|
||
"""
|
||
|
||
def __init__(self, *, products: list[list[Any]], navs: list[Any]) -> None:
|
||
self._products = list(products)
|
||
self._navs = list(navs)
|
||
self.product_queries = 0
|
||
|
||
async def execute(self, statement: Any) -> _FakeResult:
|
||
entity = statement.column_descriptions[0]["entity"]
|
||
if entity is FundNavHistory:
|
||
return _FakeResult(self._navs)
|
||
self.product_queries += 1
|
||
index = min(self.product_queries, len(self._products) + 1) - 1
|
||
if index >= len(self._products):
|
||
return _FakeResult([])
|
||
return _FakeResult(self._products[index])
|
||
|
||
|
||
def _patch_session(monkeypatch: pytest.MonkeyPatch, session: _FakeSession) -> None:
|
||
class _Factory:
|
||
def __call__(self) -> "_Factory": # pragma: no cover - 仅用于类型直觉
|
||
return self
|
||
|
||
async def __aenter__(self) -> _FakeSession:
|
||
return session
|
||
|
||
async def __aexit__(self, *args: object) -> None:
|
||
return None
|
||
|
||
monkeypatch.setattr(fund_trend_service, "SessionFactory", _Factory())
|
||
|
||
|
||
def _product(**overrides: Any) -> SimpleNamespace:
|
||
base = dict(
|
||
id=1, product_code="159382", product_name="创业板人工智能ETF南方",
|
||
product_category="ETF", risk_level="R4", exchange_code="SZSE",
|
||
status=LISTED_STATUS,
|
||
)
|
||
base.update(overrides)
|
||
return SimpleNamespace(**base)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_query_fund_trend_tool_returns_series_and_keeps_risk_level_as_text(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""`risk_level` 必须**原样透传字符串**（`'R4'`）。
|
||
|
||
回归钉子：`fin_product.risk_level` 列存的是 `R1`—`R5` 这种**字符串**，不是整数。
|
||
本工具最初写成 `int(product.risk_level)`，冒烟直接抛
|
||
`ValueError: invalid literal for int() with base 10: 'R4'` —— 且只在**真的有产品**
|
||
的路径上才炸（查不到产品的路径不碰这个字段），属于"最像成功的那种失败"。
|
||
口径与 `public_product_service._view()` 一致：原样透传，不在这一层做数值转换。
|
||
"""
|
||
session = _FakeSession(
|
||
products=[[_product()]],
|
||
navs=[
|
||
SimpleNamespace(nav_date=date(2026, 3, 2), nav=Decimal("1.0000")),
|
||
SimpleNamespace(nav_date=date(2026, 3, 3), nav=Decimal("1.5000")),
|
||
],
|
||
)
|
||
_patch_session(monkeypatch, session)
|
||
result = await query_fund_trend_tool(
|
||
FundTrendQuery(fund_code="159382"), RequestContext(user_id="1", trace_id="t")
|
||
)
|
||
assert result["found"] is True
|
||
assert result["fund_code"] == "159382"
|
||
assert result["fund_name"] == "创业板人工智能ETF南方"
|
||
assert result["risk_level"] == "R4"
|
||
assert result["source"] == NAV_SOURCE
|
||
assert result["latest_nav"] == "1.5000"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_query_fund_trend_tool_reports_missing_product(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""查不到产品时**如实返回 `found=False`**，不抛异常、不猜一只相近的。"""
|
||
session = _FakeSession(products=[[]], navs=[])
|
||
_patch_session(monkeypatch, session)
|
||
result = await query_fund_trend_tool(
|
||
FundTrendQuery(fund_name="南方稳健增利债券A"),
|
||
RequestContext(user_id="1", trace_id="t"),
|
||
)
|
||
assert result == {"found": False, "reason": "product_not_found", "source": NAV_SOURCE}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_query_fund_trend_tool_reports_ambiguous_name(
|
||
monkeypatch: pytest.MonkeyPatch,
|
||
) -> None:
|
||
"""名称命中多只时返回 `ambiguous` —— **替客户挑一只**才是这里最坏的行为（`INV-2`）。"""
|
||
session = _FakeSession(
|
||
products=[
|
||
[],
|
||
[_product(id=1, product_code="159700"), _product(id=2, product_code="159701")],
|
||
],
|
||
navs=[],
|
||
)
|
||
_patch_session(monkeypatch, session)
|
||
result = await query_fund_trend_tool(
|
||
FundTrendQuery(fund_name="科创债ETF南方"), RequestContext(user_id="1", trace_id="t")
|
||
)
|
||
assert result["found"] is False
|
||
assert result["reason"] == "ambiguous"
|
||
|
||
|
||
def test_trend_query_requires_at_least_one_entity() -> None:
|
||
"""契约层就挡住"两个实体都不给"的调用（否则工具会去扫全表）。"""
|
||
with pytest.raises(ValueError):
|
||
FundTrendQuery()
|