160 lines
7.3 KiB
Python
160 lines
7.3 KiB
Python
# ============================================================
|
||
# 统一工具运行时 × gateway 直连端点 黄金测试（矩阵 111 行剩余项）
|
||
# 覆盖：
|
||
# - 异步入口 run_tool_async：已登记放行 + TOOL 审计；未登记拒绝（tool.denied/DENIED）；
|
||
# - /api/actions/scenario/apply 经工具运行时：已登记放行 + TOOL 审计；
|
||
# - 拒绝路径保留既有响应契约 {message, refresh}，且不触发业务执行；
|
||
# - 其他直连端点盘点：gateway 内所有 handle_intent 调用点必须经工具运行时门禁。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import re
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from server.agent_core.tool_runtime import run_tool_async
|
||
from server.contracts import IntentResult
|
||
from server.state.seed import ensure_flex_seed, seed_world
|
||
|
||
|
||
class _MemStore:
|
||
"""最小内存 store：满足 run_tool_async 对 data/next_id/save 的依赖。"""
|
||
|
||
def __init__(self, data: dict):
|
||
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) -> None:
|
||
pass
|
||
|
||
|
||
def _tool_audits(world: dict) -> list[dict]:
|
||
return [a for a in world.get("auditEvents", []) if a["category"] == "TOOL"]
|
||
|
||
|
||
# ---------------- 异步入口 run_tool_async ----------------
|
||
def test_run_tool_async_registered_runs_and_writes_tool_audit():
|
||
"""已登记 schedule.run（P1）：async 入口校验放行 + tool.run 审计 + 业务执行。"""
|
||
world = seed_world()
|
||
ensure_flex_seed(world)
|
||
store = _MemStore(world)
|
||
reply = asyncio.run(run_tool_async(
|
||
store, "t",
|
||
IntentResult(intent="schedule.run",
|
||
params={"engine": "RULE", "strategy": "DELIVERY_FIRST"},
|
||
confidence=1.0, source="RULE_FAST")))
|
||
assert "排产完成" in reply.text
|
||
audits = _tool_audits(world)
|
||
assert audits and audits[0]["action"] == "tool.run"
|
||
assert audits[0]["target"]["id"] == "schedule.run"
|
||
assert audits[0]["power"] == "P1"
|
||
assert any(a["action"] == "schedule.run" for a in world["auditEvents"])
|
||
|
||
|
||
def test_run_tool_async_unregistered_denied_with_audit():
|
||
"""未登记意图：async 入口拒绝 + tool.denied（DENIED），业务绝不执行。"""
|
||
world = seed_world()
|
||
store = _MemStore(world)
|
||
reply = asyncio.run(run_tool_async(
|
||
store, "t", IntentResult(intent="unknown", params={},
|
||
confidence=1.0, source="LLM")))
|
||
assert "未登记" in reply.text
|
||
audits = _tool_audits(world)
|
||
assert audits and audits[0]["action"] == "tool.denied"
|
||
assert audits[0]["result"] == "DENIED"
|
||
assert not any(a["action"] == "schedule.run" for a in world["auditEvents"])
|
||
|
||
|
||
# ---------------- HTTP 端点：/api/actions/scenario/apply ----------------
|
||
@pytest.fixture
|
||
def client(monkeypatch, tmp_path):
|
||
import server.gateway.app as gateway_module
|
||
from server.knowledge.preferences import PreferenceStore
|
||
from tests.auth_provider import install_test_auth
|
||
|
||
install_test_auth(monkeypatch, "tenant-o-gateway")
|
||
world = seed_world()
|
||
ensure_flex_seed(world)
|
||
store = _MemStore(world)
|
||
monkeypatch.setattr(gateway_module, "get_store", lambda: store)
|
||
# 偏好仓重定向到临时文件，避免测试写入仓库数据目录
|
||
monkeypatch.setattr(
|
||
"server.knowledge.get_preferences",
|
||
lambda: PreferenceStore(str(tmp_path / "preferences.json")),
|
||
)
|
||
c = TestClient(gateway_module.create_app())
|
||
login = c.post("/api/auth/login", json={"username": "planner", "password": "test"})
|
||
assert login.status_code == 200, login.text
|
||
return c, store
|
||
|
||
|
||
def test_scenario_apply_registered_intent_allowed_with_tool_audit(client):
|
||
"""scenario.apply 经工具运行时：已登记 schedule.run 放行 + TOOL 审计，契约 {message, refresh} 保留。"""
|
||
c, store = client
|
||
resp = c.post("/api/actions/scenario/apply",
|
||
json={"strategy": "DELIVERY_FIRST", "engine": "RULE", "sessionId": "s-web"})
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
assert body["refresh"] is True
|
||
assert "排产完成" in body["message"]
|
||
audits = _tool_audits(store.data)
|
||
assert any(a["action"] == "tool.run" and a["target"]["id"] == "schedule.run" for a in audits)
|
||
|
||
|
||
def test_scenario_apply_honors_runtime_denial_and_preserves_contract(client, monkeypatch):
|
||
"""未登记拒绝路径：运行时返回拒绝 → 路由原样透传，保持 {message, refresh} 契约且不执行业务。"""
|
||
import server.agent_core.tool_runtime as tool_runtime
|
||
from server.contracts import AgentReply
|
||
|
||
denied = AgentReply(text="「unknown」不是已登记的操作（LLM 只有提议权，未登记意图不执行）。")
|
||
|
||
async def fake_run_tool_async(store, session_id, intent, actor="planner"):
|
||
return denied
|
||
|
||
monkeypatch.setattr(tool_runtime, "run_tool_async", fake_run_tool_async)
|
||
c, store = client
|
||
before_business = [a for a in store.data["auditEvents"] if a["action"] == "schedule.run"]
|
||
resp = c.post("/api/actions/scenario/apply",
|
||
json={"strategy": "DELIVERY_FIRST", "engine": "RULE", "sessionId": "s-web"})
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()
|
||
assert body["refresh"] is True
|
||
assert "未登记" in body["message"]
|
||
after_business = [a for a in store.data["auditEvents"] if a["action"] == "schedule.run"]
|
||
assert after_business == before_business # 拒绝路径不触发业务执行
|
||
|
||
|
||
# ---------------- 其他直连端点盘点（source-level golden） ----------------
|
||
def test_gateway_no_other_direct_handle_intent_bypass():
|
||
"""盘点：gateway 内所有 handle_intent 调用点都必须经工具运行时（check_tool/run_tool_async）门禁。"""
|
||
src = (Path(__file__).resolve().parents[2] / "server" / "gateway" / "app.py").read_text(encoding="utf-8")
|
||
# scenario.apply 区段必须走 run_tool_async，不再直连 await handle_intent
|
||
region_start = src.find('@app.post("/api/actions/scenario/apply")')
|
||
region_end = src.find('@app.post("/api/actions/confirm")')
|
||
assert region_start > 0 and region_end > region_start
|
||
region = src[region_start:region_end]
|
||
assert "run_tool_async" in region
|
||
assert "await handle_intent(" not in region
|
||
# 全 gateway：任何调用 handle_intent 的函数体必须同时出现门禁符号（check_tool / run_tool_async）
|
||
defs = [(m.start(), m.end()) for m in re.finditer(r"^(?:async )?def \w+", src, re.M)]
|
||
calls = [m.start() for m in re.finditer(r"handle_intent\(", src)]
|
||
assert calls, "expected handle_intent call sites in gateway"
|
||
for pos in calls:
|
||
line_start = src.rfind("\n", 0, pos) + 1
|
||
if "from server.aps_domain.workflow import" in src[line_start:pos]:
|
||
continue # import 行
|
||
enclosing = [d for d in defs if d[0] < pos]
|
||
assert enclosing, f"handle_intent 调用点不在任何函数内: {pos}"
|
||
idx = defs.index(enclosing[-1])
|
||
end = defs[idx + 1][0] if idx + 1 < len(defs) else len(src)
|
||
body = src[enclosing[-1][0]:end]
|
||
assert "check_tool(" in body or "run_tool_async(" in body, \
|
||
f"handle_intent 调用点绕过工具运行时: {pos}"
|