610 lines
21 KiB
Python
610 lines
21 KiB
Python
"""T-04 对话 Tool：意图匹配 / 归属校验 / agent_tool_call 落库 / 图节点注入。
|
||
|
||
单测层：run_tool 直调（sqlite 注入，success/blocked/error 三态与落库字段、
|
||
落库降级）；图集成：FakeLLM 捕获注入的 [工具查询结果] 上下文、降级回复
|
||
带摘要、无会话上下文空转。全链路（TestClient 真路由栈）见 test_chat.py。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import datetime as _dt
|
||
import logging
|
||
|
||
import pytest
|
||
from langchain_core.messages import AIMessage
|
||
from sqlalchemy import text
|
||
|
||
from _ddl import create_sqlite_engine
|
||
|
||
from app.config.settings import settings
|
||
from app.repository.core_ro import CoreReadOnlyRepository
|
||
from app.repository.session_repository import SessionRepository
|
||
from app.service import agent_service, tool_service
|
||
|
||
|
||
class FakeLLM:
|
||
def __init__(self, reply: str = "模拟回复"):
|
||
self.reply = reply
|
||
self.calls: list[list] = []
|
||
|
||
def invoke(self, messages):
|
||
self.calls.append(list(messages))
|
||
return AIMessage(content=self.reply)
|
||
|
||
|
||
class FakeRiskRepo:
|
||
"""合规审计仓储替身（T-04 评审 P1-2）：记录越权双写，避免测试写真库。"""
|
||
|
||
def __init__(self):
|
||
self.audit_logs: list[dict] = []
|
||
self.guard_logs: list[dict] = []
|
||
|
||
def insert_audit_log(self, payload: dict) -> None:
|
||
self.audit_logs.append(payload)
|
||
|
||
def insert_input_guard_log(self, **kwargs) -> None:
|
||
self.guard_logs.append(kwargs)
|
||
|
||
|
||
ACTOR_CUSTOMER = {"actor_id": "CUST-9527", "roles": ["customer"], "token_type": "customer"}
|
||
ACTOR_ADVISOR = {"actor_id": "STAFF-10086", "roles": ["advisor"], "token_type": "staff"}
|
||
ACTOR_RISK = {"actor_id": "STAFF-30001", "roles": ["risk_officer"], "token_type": "staff"}
|
||
ACTOR_ANALYST = {"actor_id": "STAFF-40001", "roles": ["analyst"], "token_type": "staff"}
|
||
|
||
|
||
@pytest.fixture()
|
||
def tool_env(monkeypatch):
|
||
engine = create_sqlite_engine()
|
||
session_repo = SessionRepository(engine=engine)
|
||
core_ro = CoreReadOnlyRepository(engine=engine)
|
||
with engine.begin() as conn:
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_customer (customer_id, display_name, age, occupation, is_active)"
|
||
" VALUES ('CUST-9527', '张三', 45, '工程师', 1)"
|
||
)
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at)"
|
||
" VALUES ('CUST-9527', 'A', '2026-08-01 10:00:00')"
|
||
)
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_product (product_id, product_name, min_risk_code, product_type)"
|
||
" VALUES ('P-001', '稳健一号', 'A', 'fund')"
|
||
)
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_holding (customer_id, product_id, market_value, quantity)"
|
||
" VALUES ('CUST-9527', 'P-001', 50000.00, 100)"
|
||
)
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type, amount,"
|
||
" trade_status, traded_at) VALUES ('TRD-001', 'CUST-9527', 'P-001', 'subscribe',"
|
||
" 10000.00, 'confirmed', :ts)"
|
||
),
|
||
{"ts": _dt.datetime.now()},
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_customer_advisor (advisor_id, customer_id, rel_status)"
|
||
" VALUES ('STAFF-10086', 'CUST-9527', 'active'),"
|
||
" ('STAFF-99999', 'CUST-9527', 'inactive')"
|
||
)
|
||
)
|
||
monkeypatch.setattr(tool_service, "_session_repo", lambda: session_repo)
|
||
monkeypatch.setattr(tool_service, "_core_ro", lambda: core_ro)
|
||
risk_repo = FakeRiskRepo()
|
||
monkeypatch.setattr(tool_service, "_risk_repo", lambda: risk_repo)
|
||
yield {"engine": engine, "session_repo": session_repo, "risk_repo": risk_repo}
|
||
engine.dispose()
|
||
|
||
|
||
def _tool_rows(engine):
|
||
with engine.connect() as conn:
|
||
return [
|
||
dict(r)
|
||
for r in conn.execute(
|
||
text("SELECT * FROM agent_tool_call ORDER BY id")
|
||
).mappings().all()
|
||
]
|
||
|
||
|
||
# ---------- 意图匹配 ----------
|
||
|
||
|
||
def test_match_intent_hits():
|
||
assert tool_service.match_intent("customer", "查一下我的持仓") == "query_holdings"
|
||
assert tool_service.match_intent("customer", "我的风险测评结果是什么") == "query_customer_profile"
|
||
assert tool_service.match_intent("advisor", "看看客户交易记录") == "query_recent_trades"
|
||
|
||
|
||
def test_match_intent_miss_or_agent():
|
||
assert tool_service.match_intent("customer", "你好呀") is None
|
||
assert tool_service.match_intent("risk", "查一下我的持仓") is None # risk 分支归 C1
|
||
assert tool_service.match_intent("analyst", "交易流水") is None
|
||
|
||
|
||
# ---------- run_tool：success / blocked / error 与落库 ----------
|
||
|
||
|
||
def test_run_tool_holdings_success_and_audit(tool_env):
|
||
record = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
session_id="sess-t1",
|
||
trace_id="trace-t1",
|
||
)
|
||
assert record["status"] == "success" and record["error_code"] is None
|
||
assert record["data"]["total_count"] == 1
|
||
assert record["data"]["sum_market_value"] == 50000.0
|
||
assert record["data"]["items"][0]["product_name"] == "稳健一号" # JOIN 生效
|
||
|
||
rows = _tool_rows(tool_env["engine"])
|
||
assert len(rows) == 1
|
||
row = rows[0]
|
||
assert (row["session_id"], row["trace_id"], row["tool_name"], row["status"]) == (
|
||
"sess-t1",
|
||
"trace-t1",
|
||
"query_holdings",
|
||
"success",
|
||
)
|
||
assert row["error_code"] is None and row["latency_ms"] is not None
|
||
assert '"customer_id": "CUST-9527"' in row["tool_input"]
|
||
assert '"total_count": 1' in row["tool_output"]
|
||
|
||
|
||
def test_run_tool_customer_profile(tool_env):
|
||
record = tool_service.run_tool(
|
||
tool_name="query_customer_profile",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
session_id="sess-t2",
|
||
)
|
||
assert record["status"] == "success"
|
||
assert record["data"]["found"] is True
|
||
assert record["data"]["risk_code"] == "A"
|
||
|
||
|
||
def test_run_tool_recent_trades_params(tool_env):
|
||
record = tool_service.run_tool(
|
||
tool_name="query_recent_trades",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
tool_input={"days": 7},
|
||
session_id="sess-t3",
|
||
)
|
||
assert record["status"] == "success"
|
||
assert record["data"]["days"] == 7 and record["data"]["total_count"] == 1
|
||
|
||
|
||
def test_run_tool_blocked_not_owner(tool_env):
|
||
"""customer 查他人 → blocked（AUTH_403_NOT_OWNER，与 deps 同码）+ 留痕。"""
|
||
record = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-1001",
|
||
session_id="sess-b1",
|
||
)
|
||
assert record["status"] == "blocked" and record["error_code"] == "AUTH_403_NOT_OWNER"
|
||
assert record["data"] is None
|
||
row = _tool_rows(tool_env["engine"])[0]
|
||
assert (row["status"], row["error_code"]) == ("blocked", "AUTH_403_NOT_OWNER")
|
||
|
||
|
||
def test_run_tool_blocked_advisor_not_assigned(tool_env):
|
||
record = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="advisor",
|
||
actor=ACTOR_ADVISOR,
|
||
customer_id="CUST-1010",
|
||
session_id="sess-b2",
|
||
)
|
||
assert record["status"] == "blocked" and record["error_code"] == "AUTH_403_NOT_ASSIGNED"
|
||
|
||
|
||
def test_run_tool_blocked_scope(tool_env):
|
||
"""analyst 等未授权角色 fail-closed（AUTH_403_SCOPE）。"""
|
||
record = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="customer",
|
||
actor=ACTOR_ANALYST,
|
||
customer_id="CUST-9527",
|
||
session_id="sess-b3",
|
||
)
|
||
assert record["status"] == "blocked" and record["error_code"] == "AUTH_403_SCOPE"
|
||
|
||
|
||
def test_run_tool_risk_officer_full_access(tool_env):
|
||
"""risk_officer 全量（对齐 assert_customer_access 口径）。"""
|
||
record = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="risk",
|
||
actor=ACTOR_RISK,
|
||
customer_id="CUST-9527",
|
||
session_id="sess-r1",
|
||
)
|
||
assert record["status"] == "success"
|
||
|
||
|
||
def test_run_tool_blocked_no_customer(tool_env):
|
||
record = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="",
|
||
session_id="sess-n1",
|
||
)
|
||
assert record["status"] == "blocked" and record["error_code"] == "TOOL_BLOCKED_NO_CUSTOMER"
|
||
|
||
|
||
def test_run_tool_unknown_tool(tool_env):
|
||
record = tool_service.run_tool(
|
||
tool_name="drop_database",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
session_id="sess-u1",
|
||
)
|
||
assert record["status"] == "blocked" and record["error_code"] == "TOOL_UNKNOWN"
|
||
|
||
|
||
def test_run_tool_error_swallowed_and_audited(tool_env, monkeypatch):
|
||
"""Tool 执行异常 → error 落痕且不向对话链路抛。"""
|
||
from app.tool import core_tools
|
||
|
||
original = core_tools.TOOL_REGISTRY["query_holdings"]["func"]
|
||
|
||
def boom(customer_id, core_ro):
|
||
raise RuntimeError("db exploded")
|
||
|
||
core_tools.TOOL_REGISTRY["query_holdings"]["func"] = boom
|
||
try:
|
||
record = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
session_id="sess-e1",
|
||
)
|
||
finally:
|
||
core_tools.TOOL_REGISTRY["query_holdings"]["func"] = original
|
||
assert record["status"] == "error" and record["error_code"] == "TOOL_ERROR"
|
||
row = _tool_rows(tool_env["engine"])[0]
|
||
assert (row["status"], row["error_code"]) == ("error", "TOOL_ERROR")
|
||
|
||
|
||
def test_run_tool_audit_insert_degrades(tool_env, monkeypatch):
|
||
"""留痕失败降级 warning（不阻塞对话，口径同 T-02 审计降级）。"""
|
||
def broken_repo():
|
||
raise RuntimeError("repo down")
|
||
|
||
monkeypatch.setattr(tool_service, "_session_repo", broken_repo)
|
||
record = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
session_id="sess-d1",
|
||
)
|
||
assert record["status"] == "success" # 对话链路不受留痕故障影响
|
||
|
||
|
||
# ---------- 图集成：tool 节点注入与降级 ----------
|
||
|
||
|
||
@pytest.fixture()
|
||
def fake_llm(monkeypatch):
|
||
llm = FakeLLM()
|
||
monkeypatch.setattr(agent_service, "_llm", llm)
|
||
monkeypatch.setattr(settings, "deepseek_api_key", "test-key")
|
||
yield llm
|
||
agent_service.reset_cache()
|
||
|
||
|
||
def test_graph_tool_result_injected_into_llm(fake_llm, tool_env):
|
||
out = agent_service.chat(
|
||
"customer",
|
||
[],
|
||
"查一下我的持仓",
|
||
session_id="sess-g1",
|
||
trace_id="trace-g1",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
)
|
||
msgs = fake_llm.calls[0]
|
||
assert msgs[1].__class__.__name__ == "SystemMessage"
|
||
assert "[工具查询结果]" in msgs[1].content
|
||
assert "合计市值 50000" in msgs[1].content
|
||
assert '"total_count": 1' in msgs[1].content # 数据 JSON 一并注入
|
||
assert len(_tool_rows(tool_env["engine"])) == 1
|
||
assert out["tool_results"][0]["status"] == "success"
|
||
|
||
|
||
def test_graph_no_intent_skips_tool(fake_llm, tool_env):
|
||
agent_service.chat(
|
||
"customer",
|
||
[],
|
||
"你好",
|
||
session_id="sess-g2",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
)
|
||
tool_msgs = [
|
||
m
|
||
for m in fake_llm.calls[0]
|
||
if m.__class__.__name__ == "SystemMessage" and "[工具查询结果]" in m.content
|
||
]
|
||
assert tool_msgs == []
|
||
assert _tool_rows(tool_env["engine"]) == []
|
||
|
||
|
||
def test_graph_blocked_result_visible_to_llm(fake_llm, tool_env):
|
||
"""归属拒绝以 blocked 结果注入（对话内呈现，非 403）。"""
|
||
agent_service.chat(
|
||
"customer",
|
||
[],
|
||
"查一下我的持仓",
|
||
session_id="sess-g3",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-1001",
|
||
)
|
||
content = fake_llm.calls[0][1].content
|
||
assert "拒绝" in content and "AUTH_403_NOT_OWNER" in content
|
||
|
||
|
||
def test_graph_without_session_context_no_tool(fake_llm, tool_env):
|
||
"""无会话上下文（旧调用方式）Tool 空转——T-07 兼容。"""
|
||
out = agent_service.chat("customer", [], "查一下我的持仓")
|
||
assert out["tool_results"] == []
|
||
assert "[工具查询结果]" not in fake_llm.calls[0][1].content
|
||
assert _tool_rows(tool_env["engine"]) == []
|
||
|
||
|
||
def test_graph_degraded_reply_includes_summary(tool_env, monkeypatch):
|
||
"""无 key 降级：回复携带 Tool 查询摘要（查询不白跑）。"""
|
||
monkeypatch.setattr(settings, "deepseek_api_key", "")
|
||
agent_service.reset_cache()
|
||
out = agent_service.chat(
|
||
"customer",
|
||
[],
|
||
"查一下我的持仓",
|
||
session_id="sess-g4",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
)
|
||
assert "LLM 未配置" in out["reply"]
|
||
assert "合计市值 50000" in out["reply"]
|
||
assert out["has_disclaimer"] is True # 降级回复同样经 guard
|
||
|
||
|
||
# ---------- summarize / context_text ----------
|
||
|
||
|
||
def test_summarize_variants(tool_env):
|
||
ok = tool_service.run_tool(
|
||
tool_name="query_customer_profile",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
session_id="sess-s1",
|
||
)
|
||
text = tool_service.summarize(ok)
|
||
assert "张三" in text and "A" in text
|
||
|
||
blocked = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-1001",
|
||
session_id="sess-s2",
|
||
)
|
||
assert "无权访问" in tool_service.summarize(blocked)
|
||
|
||
not_found = tool_service.run_tool(
|
||
tool_name="query_customer_profile",
|
||
agent_type="risk",
|
||
actor=ACTOR_RISK,
|
||
customer_id="CUST-NOPE",
|
||
session_id="sess-s3",
|
||
)
|
||
assert "未找到" in tool_service.summarize(not_found)
|
||
|
||
|
||
def test_context_text_empty():
|
||
assert tool_service.context_text([]) == ""
|
||
|
||
|
||
# ---------- T-04 评审补测：入参白名单与边界钳制 ----------
|
||
|
||
|
||
def _run_days(tool_env, days):
|
||
return tool_service.run_tool(
|
||
tool_name="query_recent_trades",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
tool_input={"days": days},
|
||
session_id="sess-p1",
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"raw,expected",
|
||
[(0, 1), (-30, 1), (400, 365), (365, 365), (1, 1), ("7", 7)],
|
||
)
|
||
def test_days_clamped(tool_env, raw, expected):
|
||
"""越界天数钳制到 [1,365]，非整数字符串可转换——防 LLM 传参拉爆查询。"""
|
||
record = _run_days(tool_env, raw)
|
||
assert record["status"] == "success"
|
||
assert record["data"]["days"] == expected
|
||
|
||
|
||
@pytest.mark.parametrize("raw", [7.5, "abc", True, None, [7]])
|
||
def test_days_invalid_blocked(tool_env, raw):
|
||
record = _run_days(tool_env, raw)
|
||
assert record["status"] == "blocked" and record["error_code"] == "TOOL_BAD_PARAM"
|
||
|
||
|
||
@pytest.mark.parametrize("extra", [{"foo": 1}, {"customer_id": "CUST-1001"}, {"core_ro": None}])
|
||
def test_unknown_param_blocked(tool_env, extra):
|
||
"""白名单外入参（含注入参数名）一律拒，避免重复关键字 TypeError 被吞。"""
|
||
record = tool_service.run_tool(
|
||
tool_name="query_recent_trades",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
tool_input=extra,
|
||
session_id="sess-p2",
|
||
)
|
||
assert record["status"] == "blocked" and record["error_code"] == "TOOL_BAD_PARAM"
|
||
|
||
|
||
def test_tool_without_params_rejects_any_input(tool_env):
|
||
record = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
tool_input={"days": 7},
|
||
session_id="sess-p3",
|
||
)
|
||
assert record["status"] == "blocked" and record["error_code"] == "TOOL_BAD_PARAM"
|
||
|
||
|
||
# ---------- T-04 评审补测：归属与留痕 ----------
|
||
|
||
|
||
def test_run_tool_blocked_advisor_inactive_rel(tool_env):
|
||
"""rel_status='inactive'（非缺失）同样拒绝。"""
|
||
record = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="advisor",
|
||
actor={"actor_id": "STAFF-99999", "roles": ["advisor"], "token_type": "staff"},
|
||
customer_id="CUST-9527",
|
||
session_id="sess-b5",
|
||
)
|
||
assert record["status"] == "blocked" and record["error_code"] == "AUTH_403_NOT_ASSIGNED"
|
||
|
||
|
||
def test_run_tool_blocked_writes_authz_audit(tool_env):
|
||
"""P1-2：对话内越权进全局鉴权台账（audit_log + input_guard_log 双写）。"""
|
||
record = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-1001",
|
||
session_id="sess-b6",
|
||
)
|
||
assert record["error_code"] == "AUTH_403_NOT_OWNER"
|
||
audit = tool_env["risk_repo"].audit_logs
|
||
guard = tool_env["risk_repo"].guard_logs
|
||
assert len(audit) == 1
|
||
assert audit[0]["decision"] == "forbidden"
|
||
assert audit[0]["customer_id"] == "CUST-1001"
|
||
assert audit[0]["input_summary"]["code"] == "AUTH_403_NOT_OWNER"
|
||
assert len(guard) == 1
|
||
assert guard[0]["agent_type"] == "customer"
|
||
assert guard[0]["raw_excerpt"] == "AUTH_403_NOT_OWNER"
|
||
|
||
|
||
def test_non_authz_block_skips_authz_audit(tool_env):
|
||
"""工具层自身拒绝（TOOL_*）不污染鉴权台账。"""
|
||
tool_service.run_tool(
|
||
tool_name="drop_database",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
session_id="sess-b7",
|
||
)
|
||
assert tool_env["risk_repo"].audit_logs == []
|
||
assert tool_env["risk_repo"].guard_logs == []
|
||
|
||
|
||
def test_audit_degrade_logs_error_with_context(tool_env, monkeypatch, caplog):
|
||
"""P1-1：留痕失败仍降级不阻塞，但必须 error 级留底且带定位字段。"""
|
||
def broken_repo():
|
||
raise RuntimeError("repo down")
|
||
|
||
monkeypatch.setattr(tool_service, "_session_repo", broken_repo)
|
||
with caplog.at_level(logging.ERROR, logger="app.service.tool_service"):
|
||
record = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-9527",
|
||
session_id="sess-d2",
|
||
trace_id="trace-d2",
|
||
)
|
||
assert record["status"] == "success"
|
||
entries = [r.getMessage() for r in caplog.records if "agent_tool_call insert failed" in r.getMessage()]
|
||
assert entries and "sess-d2" in entries[0] and "trace-d2" in entries[0]
|
||
|
||
|
||
def test_authz_audit_failure_does_not_break_tool(tool_env, monkeypatch):
|
||
"""越权留痕自身故障不影响 blocked 语义。"""
|
||
def broken_repo():
|
||
raise RuntimeError("audit repo down")
|
||
|
||
monkeypatch.setattr(tool_service, "_risk_repo", broken_repo)
|
||
record = tool_service.run_tool(
|
||
tool_name="query_holdings",
|
||
agent_type="customer",
|
||
actor=ACTOR_CUSTOMER,
|
||
customer_id="CUST-1001",
|
||
session_id="sess-b8",
|
||
)
|
||
assert record["status"] == "blocked" and record["error_code"] == "AUTH_403_NOT_OWNER"
|
||
|
||
|
||
# ---------- T-04 评审补测：上下文与摘要边界 ----------
|
||
|
||
|
||
def test_context_text_truncates_long_payload():
|
||
big = {"items": [{"name": "x" * 100} for _ in range(50)]}
|
||
record = {"tool_name": "query_holdings", "status": "success", "data": big}
|
||
text = tool_service.context_text([record])
|
||
assert "…(truncated)" in text
|
||
assert len(text) < tool_service._RESULT_CONTEXT_MAX_CHARS + 500
|
||
|
||
|
||
def test_summarize_holdings_shows_first_five_only():
|
||
record = {
|
||
"tool_name": "query_holdings",
|
||
"status": "success",
|
||
"data": {
|
||
"total_count": 25,
|
||
"sum_market_value": 250.0,
|
||
"truncated": False,
|
||
"items": [{"product_name": f"P{i}", "market_value": 10.0} for i in range(25)],
|
||
},
|
||
}
|
||
text = tool_service.summarize(record)
|
||
assert "共 25 笔" in text
|
||
assert text.count("市值") == 6 # 合计 1 次 + 明细 5 条
|
||
|
||
|
||
def test_summarize_holdings_truncated_note():
|
||
record = {
|
||
"tool_name": "query_holdings",
|
||
"status": "success",
|
||
"data": {"total_count": 500, "sum_market_value": 1.0, "truncated": True, "items": []},
|
||
}
|
||
assert "已达拉取上限" in tool_service.summarize(record)
|
||
|
||
|
||
def test_summarize_tool_rejection_not_authz_wording():
|
||
record = {"tool_name": "query_holdings", "status": "blocked", "error_code": "TOOL_BAD_PARAM"}
|
||
text = tool_service.summarize(record)
|
||
assert "无权访问" not in text and "TOOL_BAD_PARAM" in text
|