81 lines
3.4 KiB
Python
81 lines
3.4 KiB
Python
# ============================================================
|
||
# round-16b 黄金测试：EXTERNAL 引擎（flex.schedule.external）接入 trace_chain
|
||
# 覆盖：ALGO_RUN 审计携带 traceChainHash（64 位 hex）、traceCount=3、
|
||
# evidence 含 skill/run/schedule-version 引用、可复算。
|
||
# ============================================================
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import re
|
||
|
||
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 对 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 _run_external_once() -> tuple[dict, _MemStore]:
|
||
store = _MemStore(seed_world())
|
||
ensure_flex_seed(store.data)
|
||
|
||
async def _run():
|
||
await handle_intent(
|
||
store, "t",
|
||
IntentResult(intent="flex.schedule",
|
||
params={"sortMode": "EXTERNAL", "skillId": "algo.stub"},
|
||
confidence=1.0, source="RULE_FAST"))
|
||
|
||
asyncio.run(_run())
|
||
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]["rationale"], store
|
||
|
||
|
||
def test_external_flex_audit_carries_trace_chain_hash():
|
||
"""1) flex.schedule.external 审计含 traceChainHash/traceCount/traceSummary（非空、格式正确）。"""
|
||
rationale, _ = _run_external_once()
|
||
chain_hash = rationale["traceChainHash"]
|
||
assert isinstance(chain_hash, str) and chain_hash, "traceChainHash 非空"
|
||
assert _HEX64.match(chain_hash), "traceChainHash 应为 64 位小写十六进制 SHA-256"
|
||
assert rationale["traceCount"] == 3, "链应含 schedule-version/run/algorithm 三条"
|
||
summary = rationale["traceSummary"]
|
||
assert isinstance(summary, list) and len(summary) == 3
|
||
assert [it["kind"] for it in summary] == ["schedule-version", "run", "algorithm"]
|
||
assert summary[1]["ref"].startswith("run:"), "run 元素引用应为 run:<runId>"
|
||
assert summary[2]["ref"] == "algo.stub", "algorithm 元素引用应为 skillId"
|
||
|
||
|
||
def test_external_flex_evidence_has_skill_run_and_schedule_version():
|
||
"""2) evidence 含 skill:<skillId>、run:<runId>、schedule-version:<id> 引用（对齐矩阵 114 行协议）。"""
|
||
rationale, _ = _run_external_once()
|
||
evidence = rationale["evidence"]
|
||
assert any(ref.startswith("skill:") for ref in evidence), "evidence 应含 skill 引用"
|
||
assert any(ref.startswith("run:") for ref in evidence), "evidence 应含 run 引用"
|
||
assert any(ref.startswith("schedule-version:") for ref in evidence), "evidence 应含 schedule-version 引用"
|
||
|
||
|
||
def test_external_flex_trace_recomputable_same_input():
|
||
"""3) 可复算：同输入同种子再次触发外部排产得到相同 chainHash。"""
|
||
r1, _ = _run_external_once()
|
||
r2, _ = _run_external_once()
|
||
assert r1["traceChainHash"] == r2["traceChainHash"], "同输入应可复算"
|
||
assert r1["traceSummary"] == r2["traceSummary"]
|
||
assert r1["traceCount"] == r2["traceCount"] == 3
|