137 lines
5.3 KiB
Python
137 lines
5.3 KiB
Python
# ============================================================
|
||
# round-39 黄金测试：EXTERNAL 真实 HTTP skill 端到端（矩阵 114 剩余项）
|
||
# 覆盖：algo_skill_stub.create_stub_app 起真实 HTTP 服务 → 注册 HTTP skill →
|
||
# flex.schedule.external 走真实 HTTP POST → ALGO_RUN 审计携带
|
||
# traceChainHash/traceCount/traceSummary/traceItems → 可复算 + verify_audit_trace 校验通过。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import re
|
||
import socket
|
||
import threading
|
||
import time
|
||
|
||
import pytest
|
||
|
||
from server.agent_core.evidence import verify_audit_trace
|
||
from server.aps_domain.workflow import handle_intent
|
||
from server.contracts import IntentResult
|
||
from server.state.seed import ensure_flex_seed, seed_world
|
||
|
||
_HEX64 = re.compile(r"^[0-9a-f]{64}$")
|
||
|
||
|
||
class _MemStore:
|
||
"""最小内存 store：满足 _run_flex EXTERNAL 路径对 data/next_id/save 的依赖。"""
|
||
|
||
def __init__(self, data):
|
||
self.data = data
|
||
|
||
def next_id(self, kind: str) -> int:
|
||
key = "_c_" + kind
|
||
self.data[key] = self.data.get(key, 0) + 1
|
||
return self.data[key]
|
||
|
||
def save(self):
|
||
pass
|
||
|
||
|
||
def _free_port() -> int:
|
||
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
|
||
s.bind(("127.0.0.1", 0))
|
||
return s.getsockname()[1]
|
||
|
||
|
||
@pytest.fixture
|
||
def http_stub(tmp_path, monkeypatch):
|
||
"""启动 algo_skill_stub 真实 HTTP 服务，并注册为外部算法 skill（algo.http.stub）。"""
|
||
import httpx
|
||
import uvicorn
|
||
from server.integrations.algo_skill_stub import create_stub_app
|
||
|
||
port = _free_port()
|
||
server = uvicorn.Server(uvicorn.Config(
|
||
create_stub_app(), host="127.0.0.1", port=port, log_level="warning"))
|
||
thread = threading.Thread(target=server.run, daemon=True)
|
||
thread.start()
|
||
try:
|
||
deadline = 100
|
||
while not server.started and deadline > 0:
|
||
time.sleep(0.05)
|
||
deadline -= 1
|
||
with httpx.Client(timeout=5.0) as client: # 真实 HTTP 健康探测
|
||
resp = client.get("http://127.0.0.1:{0}/health".format(port))
|
||
assert resp.status_code == 200 and resp.json().get("ok") is True
|
||
skills_file = tmp_path / "skills.json"
|
||
skills_file.write_text(json.dumps({"skills": [{
|
||
"skill_id": "algo.http.stub",
|
||
"name": "HTTP 算法桩",
|
||
"endpoint": "http://127.0.0.1:{0}".format(port),
|
||
"track": "flex",
|
||
"enabled": True,
|
||
"timeout_sec": 10.0,
|
||
"capabilities": ["flex_schedule"],
|
||
"ragScopes": ["process", "sop"],
|
||
}]}), encoding="utf-8")
|
||
import server.agent_core.skills as sk
|
||
monkeypatch.setenv("APS_SKILLS_PATH", str(skills_file))
|
||
monkeypatch.setattr(sk, "_registry", None) # 下次 get_skills() 按新配置重建
|
||
yield {"port": port, "skills_file": skills_file}
|
||
finally:
|
||
server.should_exit = True
|
||
thread.join(timeout=5.0)
|
||
import server.agent_core.skills as sk
|
||
sk._registry = None # 清理单例，避免污染其他测试
|
||
|
||
|
||
def _run_external_http_once(store) -> None:
|
||
async def _run():
|
||
await handle_intent(
|
||
store, "t",
|
||
IntentResult(intent="flex.schedule",
|
||
params={"skillId": "algo.http.stub", "useExternal": True},
|
||
confidence=1.0, source="RULE_FAST"))
|
||
|
||
asyncio.run(_run())
|
||
|
||
|
||
def _external_audit(store) -> dict:
|
||
runs = [e for e in store.data["auditEvents"]
|
||
if e["category"] == "ALGO_RUN" and e["action"] == "flex.schedule.external"]
|
||
assert runs, "应产生一条 ALGO_RUN flex.schedule.external 审计"
|
||
return runs[-1]
|
||
|
||
|
||
def test_external_http_skill_trace_written_and_recomputable(http_stub):
|
||
"""矩阵 114：EXTERNAL 真实 HTTP skill 端到端 —— trace 写入且同输入可复算。"""
|
||
store = _MemStore(seed_world())
|
||
ensure_flex_seed(store.data)
|
||
_run_external_http_once(store)
|
||
ev = _external_audit(store)
|
||
rat = ev["rationale"]
|
||
assert isinstance(rat["traceChainHash"], str) and _HEX64.match(rat["traceChainHash"]), "traceChainHash 应为 64 位 SHA-256"
|
||
assert rat["traceCount"] == 3, "链应含 schedule-version/run/algorithm 三条"
|
||
assert [it["kind"] for it in rat["traceSummary"]] == ["schedule-version", "run", "algorithm"]
|
||
assert rat["traceSummary"][2]["ref"] == "algo.http.stub", "algorithm 元素引用应为 skillId"
|
||
assert isinstance(rat.get("traceItems"), list) and len(rat["traceItems"]) == 3, "traceItems 应落盘以便精确重算"
|
||
|
||
store2 = _MemStore(seed_world())
|
||
ensure_flex_seed(store2.data)
|
||
_run_external_http_once(store2)
|
||
rat2 = _external_audit(store2)["rationale"]
|
||
assert rat2["traceChainHash"] == rat["traceChainHash"], "同输入 EXTERNAL trace 链应可复算"
|
||
assert rat2["traceSummary"] == rat["traceSummary"]
|
||
|
||
|
||
def test_external_http_skill_trace_passes_verify_audit_trace(http_stub):
|
||
"""EXTERNAL HTTP 端到端产出的 ALGO_RUN 审计应通过 verify_audit_trace 全量校验。"""
|
||
store = _MemStore(seed_world())
|
||
ensure_flex_seed(store.data)
|
||
_run_external_http_once(store)
|
||
result = verify_audit_trace(store.data["auditEvents"])
|
||
assert result["ok"] is True
|
||
assert result["checked"] >= 1
|
||
assert result["broken"] == []
|