73 lines
3.4 KiB
Python
73 lines
3.4 KiB
Python
# ============================================================
|
||
# 统一工具运行时黄金测试（plan.md §3.3 / 矩阵 111 行「LLM 只有提议权」）
|
||
# 覆盖：未登记意图拒绝（LLM 只有提议权）、已登记意图直通、TOOL 审计溯源、
|
||
# 只读白名单放行、不能通过未登记 API 绕过。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
from server.agent_core.tool_runtime import is_registered, run_tool
|
||
from server.contracts import IntentResult
|
||
from server.state.seed import ensure_flex_seed, seed_world
|
||
|
||
|
||
class _MemStore:
|
||
"""最小内存 store：满足 run_tool 对 data/next_id/save 的依赖。"""
|
||
|
||
def __init__(self, data):
|
||
self.data = data
|
||
|
||
def next_id(self, kind: str) -> int:
|
||
key = f"_c_{kind}"
|
||
self.data[key] = self.data.get(key, 0) + 1
|
||
return self.data[key]
|
||
|
||
def save(self):
|
||
pass
|
||
|
||
|
||
def test_unregistered_intent_denied_with_audit():
|
||
"""未登记意图（LLM 只有提议权）：拒绝执行并写 TOOL 审计（result=DENIED）。"""
|
||
store = _MemStore(seed_world())
|
||
reply = run_tool(store, "t", IntentResult(intent="unknown", params={},
|
||
confidence=1.0, source="LLM"))
|
||
assert "未登记" in reply.text
|
||
audits = [a for a in store.data["auditEvents"] if a["category"] == "TOOL"]
|
||
assert audits and audits[0]["action"] == "tool.denied"
|
||
assert audits[0]["result"] == "DENIED"
|
||
# 未登记意图不得触发任何业务执行（无 schedule.run 等）
|
||
assert not any(a["action"] == "schedule.run" for a in store.data["auditEvents"])
|
||
|
||
|
||
def test_registered_intent_runs_and_audits():
|
||
"""已登记意图（schedule.run P1）：经统一运行时执行并写 tool.run 审计。"""
|
||
world = seed_world()
|
||
ensure_flex_seed(world)
|
||
store = _MemStore(world)
|
||
reply = run_tool(store, "t", IntentResult(intent="schedule.run",
|
||
params={"engine": "RULE", "strategy": "DELIVERY_FIRST"},
|
||
confidence=1.0, source="LLM"))
|
||
assert "排产完成" in reply.text
|
||
audits = [a for a in store.data["auditEvents"] if a["category"] == "TOOL"]
|
||
assert audits and audits[0]["action"] == "tool.run"
|
||
# 业务审计同步落盘（schedule.run ALGO_RUN）
|
||
assert any(a["action"] == "schedule.run" for a in store.data["auditEvents"])
|
||
|
||
|
||
def test_registered_readonly_intent_passes():
|
||
"""只读白名单（master.query 等）放行：仍经统一运行时。"""
|
||
world = seed_world()
|
||
store = _MemStore(world)
|
||
reply = run_tool(store, "t", IntentResult(intent="master.query", params={"entity": "orders"},
|
||
confidence=1.0, source="LLM"))
|
||
assert reply.text, "只读意图应产生回复"
|
||
assert is_registered("master.query") is True
|
||
|
||
|
||
def test_is_registered_matches_power_map_and_whitelist():
|
||
"""登记判定：写入类匹配 _POWER_MAP；只读类在白名单；其余未登记。"""
|
||
assert is_registered("schedule.run") is True # P1 写入类
|
||
assert is_registered("schedule.publish") is True # P2 写入类
|
||
assert is_registered("master.query") is True # 只读白名单
|
||
assert is_registered("knowledge.query") is True # 只读白名单
|
||
assert is_registered("unknown") is False # 未登记
|