diff --git a/tests/integration/test_customer_service_trend_chart_persistence.py b/tests/integration/test_customer_service_trend_chart_persistence.py new file mode 100644 index 0000000..5e58485 --- /dev/null +++ b/tests/integration/test_customer_service_trend_chart_persistence.py @@ -0,0 +1,144 @@ +"""`W29-d` · 图表/事项码数据**落库**的端到端验证（走真实 HTTP + Worker）。 + +## 为什么这个文件必须存在 + +`W29` 的图表与 `W28` 的事项码，此前的全部验证都是**进程内直调** +（`_w29_e2e_check.py` 直接构造 `ToolExecutor`；金标探针直接调 Agent）—— +**从来没有走过「HTTP 受理 → Worker 执行 → 落库」这条真实链路**。 + +后果是具体可查的：库里 `tool_calls LIKE '%trend_chart%'` 命中 **0 条**、 +`'%svc_topic%'` 也是 **0 条**（最近一条真实 run 停在 2026-09-21 14:46，早于这两批改动）。 + +而**前端取图完全依赖这条路**：前端读的是 `result.tool_calls.data.trend_chart`， +它由 `agent_persistence_service` 写进 `conversation_message.tool_calls` 这个 JSON 列。 +**这条路不通，图就是「代码里有、页面上没有」。** + +本文件补的就是这一段： + +1. 走真实 `POST /api/v1/agent-runs` → `WorkerRuntime` 执行 → 落库； +2. 断言 **HTTP 出参**里 `result.tool_calls.data.trend_chart` 存在（= 前端取数路径本身）； +3. 断言 **库里** `conversation_message.tool_calls` 同样含 `trend_chart` 与 `svc_topic`。 + +## 为什么断言数据、而不是断言像素 + +组件渲染已由 `_w29_chart_render_check.mjs`（jsdom）覆盖（11/11）。 +本文件管的是它的**上游**：数据有没有被写进库、有没有出到接口。 +分开的理由是**失败时的定位完全不同** —— +「图不显示」既可能是数据没落库（本文件），也可能是渲染出错（那个文件）， +混在一个测试里只会得到一句"图表测试失败"。 +""" + +from __future__ import annotations + +import json +from uuid import uuid4 + +import httpx +import pytest +from sqlalchemy import delete, select + +from app.api.dependencies.auth import build_request_context +from app.core.contracts import RequestContext +from app.infrastructure.db import SessionFactory +from app.main import create_app +from app.model.conversation import ConversationMessage +from app.model.platform import AgentRun +from app.service.agent.bootstrap import get_agent_factory +from app.worker.runtime import WorkerRuntime + +#: 能确定性命中 `E6` 出口的问句：含行情词 + 6 位产品代码 ⇒ **不调模型**、快。 +TREND_QUESTION = "159382今天净值走势" + + +@pytest.mark.integration +async def test_trend_chart_and_service_topic_reach_db_and_api() -> None: + """走势问句走完整链路后，`trend_chart` 与 `svc_topic` 必须同时出现在**出参与库里**。""" + context = RequestContext( + user_id="1", + trace_id=str(uuid4()), + roles=("customer",), + customer_ids=("1",), + data_scope="self", + portal="api", + # `fund:quote:read` 是 `query_fund_trend` 的 `required_permission`（`bootstrap.py`）。 + permissions=("agent:run", "fund:quote:read"), + ) + app = create_app() + app.dependency_overrides[build_request_context] = lambda: context + session_id = f"w29d-chart-{uuid4()}" + run_id: str | None = None + async def identity(value: object) -> RequestContext: + # ⚠️ `WorkerRuntime.restore_context` 会 `await self.resolve_identity(...)`， + # 所以这里**必须**是协程函数 —— 传同步 lambda 会得到 + # `TypeError: object RequestContext can't be used in 'await' expression`。 + del value + return context + + runtime = WorkerRuntime(get_agent_factory(), resolve_identity=identity) + + try: + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + accepted = await client.post("/api/v1/agent-runs", json={ + "agent_type": "customer_service", + "message": TREND_QUESTION, + "session_id": session_id, + "idempotency_key": str(uuid4()), + }) + assert accepted.status_code == 202, accepted.text + run_id = accepted.json()["data"]["run_id"] + + assert await runtime.dispatch_one(run_id=run_id), "run 未派发成功" + await runtime.execute(run_id) + + snapshot = (await client.get(f"/api/v1/agent-runs/{run_id}")).json()["data"] + + assert snapshot["status"] == "succeeded", snapshot + result = snapshot["result"] or {} + + # ① **前端取数路径**：`result.tool_calls.data.trend_chart` + # 前端之所以读这一层而不是 `result.data`：后者只对 `financial_nl2sql` 的 run 暴露。 + stored = result.get("tool_calls") + assert isinstance(stored, dict), f"tool_calls 形状异常：{type(stored).__name__}" + data = stored.get("data") + assert isinstance(data, dict), f"tool_calls.data 未落库：{list(stored)}" + + assert "trend_chart" in data, f"data 里没有 trend_chart，实际键：{sorted(data)}" + chart = data["trend_chart"] + assert chart["kind"] == "interval_change" + assert chart["source"] == "fin_nav_history" + assert chart["intervals"], "区间数据为空 —— 图会画不出来" + # `INV-8` 的落库侧前提：这些数字必须在正文里出现过（正文侧由 jsdom 测试断言）。 + for interval in chart["intervals"]: + assert str(interval["change_pct"]) in (result.get("content") or ""), ( + f"图数据里的涨跌 {interval['change_pct']} 未出现在答复正文：{result.get('content')}" + ) + + # ② `W28` 的事项码同批落库（此前同样是「代码里有、库里没有」） + assert data.get("svc_topic"), "svc_topic 未落库" + assert data.get("svc_owner") and data.get("svc_retention") + + # ③ 库里确实有（不只是出参拼出来的） + async with SessionFactory() as session: + message = (await session.execute( + select(ConversationMessage) + .where( + ConversationMessage.session_id == session_id, + ConversationMessage.role == "assistant", + ) + .order_by(ConversationMessage.id.desc()) + .limit(1) + )).scalars().first() + assert message is not None, "本次会话没有落库的助手消息" + raw = json.dumps(message.tool_calls, ensure_ascii=False) + assert "trend_chart" in raw, "库里的 tool_calls 没有 trend_chart" + assert "svc_topic" in raw, "库里的 tool_calls 没有 svc_topic" + finally: + # 只删本次 session 产生的行（`session_id` 带 uuid，不会误伤历史数据）。 + async with SessionFactory() as session, session.begin(): + await session.execute( + delete(ConversationMessage).where(ConversationMessage.session_id == session_id) + ) + if run_id: + await session.execute(delete(AgentRun).where(AgentRun.run_id == run_id)) diff --git a/tests/integration/test_financial_nl2sql_http_wiring.py b/tests/integration/test_financial_nl2sql_http_wiring.py new file mode 100644 index 0000000..197ee2b --- /dev/null +++ b/tests/integration/test_financial_nl2sql_http_wiring.py @@ -0,0 +1,105 @@ +"""`W29-d` · NL2SQL 接线后的 **HTTP 链路**验证（此前只测到服务层）。 + +## 为什么需要它 + +`W29` 把 `agent_tools/financial_nl2sql:financial_query` 发布出去之后，验证只做到两层： + +- **服务层直调**：`FinancialNL2SQLService().query(...)`； +- **工具层**：自建 `ToolExecutor` + **手写** + `configured_tools={"financial_query": ("query_financial_data",)}`。 + +这两层都**绕过了**真实链路最关键的一段：`agent_tools` 白名单是**从库里装配**的 +（`AgentFactory` 读 `config_release` 的生效快照）。手写 `configured_tools` 等于 +**把要验证的东西自己填好了** —— 它能证明"工具逻辑对"，但**证明不了"接线生效"**。 + +库里的证据也印证了这一点：`agent_run` 表里 `agent_type='financial_nl2sql'` **0 条**。 + +本文件走完整链路：HTTP 受理 → Worker 从**生效配置**装配白名单 → 工具执行 → 落库 → 读出参。 +""" + +from __future__ import annotations + +from uuid import uuid4 + +import httpx +import pytest +from sqlalchemy import delete + +from app.api.dependencies.auth import build_request_context +from app.core.contracts import RequestContext +from app.infrastructure.db import SessionFactory +from app.main import create_app +from app.model.conversation import ConversationMessage +from app.model.platform import AgentRun +from app.service.agent.bootstrap import get_agent_factory +from app.worker.runtime import WorkerRuntime + +WRITE_WORDS = ("INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "TRUNCATE", "GRANT", "REVOKE") + + +@pytest.mark.integration +async def test_financial_nl2sql_serves_over_http_after_wiring() -> None: + """接线后，`financial_nl2sql` 必须能在**真实 HTTP 链路**上出结果。 + + 断言的是"端到端可用"，不是"工具逻辑正确"（后者由 `test_financial_nl2sql_*.py` 覆盖）。 + 两者会分别失败，所以分开测 —— 否则"NL2SQL 坏了"这句话定位不到是哪一层。 + """ + context = RequestContext( + user_id="1", + trace_id=str(uuid4()), + roles=("operator",), + data_scope="all", + portal="api", + permissions=("agent:run", "financial:nl2sql:read"), + ) + app = create_app() + app.dependency_overrides[build_request_context] = lambda: context + session_id = f"w29d-nl2sql-{uuid4()}" + run_id: str | None = None + + async def identity(value: object) -> RequestContext: + # 必须是协程函数：`restore_context` 会 `await` 它。 + del value + return context + + runtime = WorkerRuntime(get_agent_factory(), resolve_identity=identity) + try: + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://test" + ) as client: + accepted = await client.post("/api/v1/agent-runs", json={ + "agent_type": "financial_nl2sql", + "message": "查询近30天净值", + "session_id": session_id, + "idempotency_key": str(uuid4()), + }) + assert accepted.status_code == 202, accepted.text + run_id = accepted.json()["data"]["run_id"] + assert await runtime.dispatch_one(run_id=run_id), "run 未派发成功" + await runtime.execute(run_id) + snapshot = (await client.get(f"/api/v1/agent-runs/{run_id}")).json()["data"] + + assert snapshot["status"] == "succeeded", snapshot + result = snapshot["result"] or {} + + # ① 出参：`RunQueryService` **只**对 `financial_nl2sql` 暴露 `data` 与 `sql` + assert isinstance(result.get("data"), dict), f"出参缺 data：{sorted(result)}" + assert isinstance(result.get("sql"), str), f"出参缺 sql：{sorted(result)}" + + # ② 只读自证（工具声明 `read_only=True`，这里验它真的没产出写语句） + sql = result["sql"] + upper = sql.upper() + assert upper.startswith("SELECT"), f"只读工具产出了非 SELECT：{sql}" + assert not any(word in upper for word in WRITE_WORDS), sql + assert "*" not in sql, f"禁止 SELECT *：{sql}" + assert "LIMIT" in upper, f"缺 LIMIT：{sql}" + + # ③ 真的**执行**了，而不是只编译完就返回 —— `data.total` 只有执行后才填 + assert "total" in result["data"], f"查询未真正执行：{result['data']}" + finally: + async with SessionFactory() as session, session.begin(): + await session.execute( + delete(ConversationMessage).where(ConversationMessage.session_id == session_id) + ) + if run_id: + await session.execute(delete(AgentRun).where(AgentRun.run_id == run_id))