Files
ThearchyHelios e96b0180d9 可验证报告、报告导出写「被驳回」:元信息带 end_reason,核对仍按运行取消
为什么:被驳回的运行导出后,可验证报告的运行结果写「已取消」,报告导出的运行状态也写「已取消」,和系统里「被驳回」两种
说法。封存里被驳回和手动停止是同一种结束事件(运行取消),分开它们的原因在那条事件的内容里,而运行取消的内容不随文件
附带(里面有驳回人和备注)——光看封存分不出来。

做了什么:
- 两种导出的元信息都带 end_reason(bundle/build、report_export/build、对应的接口)。
- verifier.js:运行结果按运行记录写「被驳回」,核对的仍是「最后一条结束类事件是运行取消」,旁边写「已核对运行取消,被驳回
  取自运行记录」,原因里写明这一半要去系统里对照;手动停止、老运行照旧「已取消」;文件里写被驳回、事件却是运行结束的照旧
  不通过。runtime.js 的运行状态照 end_reason 写「被驳回」。
- 样本页加一份 rejected(check-bundle 样本数 17 → 18),断言写被驳回、写明取自运行记录、不提重新封存。

测试:test_bundle_rejected 用 node 载入 verifier.js 核运行信息(被驳回、手动停止、老运行、伪造);test_bundle_build 核
元信息和样本页;test_bundle_api 核两种导出接口都带 end_reason。把实现退回去回跑,被驳回、伪造两条和元信息、样本页、
接口的断言都失败(手动停止、老运行两条守的是原样,照旧通过);改完通过。
2026-10-16 12:46:48 +08:00

433 lines
22 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""可验证报告的接口:GET /api/runs/{run_id}/bundle.html。
照证据测试的夹具真跑一次:两个虚构的 SQLite 数据源(门店流水没设遮罩,会员库遮了 phone 列),口径卡、
报告引用两份快照里的格子和一个指标。核对:
- 只给已封存且封存核对一致的运行:没封存、封存后被改过都回 409;
- 导出的数据里,事件行按浏览器的拼法重算等于运行记下的 manifest_hash;
- 遮罩库上选出了号码的快照只有 id 没有内容,它的查询事件只有哈希;运行之后才设的遮罩同样生效;同一个库上只汇总消费额
(没碰号码列)的那次查询,快照照常随文件附带;
- 附件名是「<运行名>-可验证报告.html」,中文名按 RFC 5987 编码;
- 只读:导出前后运行记录、事件一条不多一条不少。
另外核对工件库新加的 read_bytes:原样取回文件字节。
"""
from __future__ import annotations
import asyncio
import base64
import hashlib
import json
import re
import sqlite3
from urllib.parse import quote
import pytest
from httpx import ASGITransport, AsyncClient
from langchain_core.messages import AIMessage
from sqlalchemy import func, select
from app.core.artifact_store import canonical_json, content_hash, put_json, read_bytes
from app.core.config import settings
from app.data.engine import engines
from app.db.base import SessionLocal
from app.db.models import DataSource, Run, RunEvent
from app.engine.runner import run_manager
from app.main import app
SHOP_SQL = "SELECT id, store, amount FROM orders ORDER BY id"
CRM_SQL = "SELECT id, name, phone, spend FROM members ORDER BY id"
TEXT = ("## 门店周报\n\n青石店本周流水 [[v:Q1.r0.amount]],会员消费 [[v:Q2.r1.spend]],"
"口径卡销售额 [[m:gmv]]。[[see:Q1]]")
@pytest.fixture(autouse=True)
async def engine_up():
await run_manager.setup()
yield
await run_manager.shutdown()
@pytest.fixture
async def client():
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
yield c
async def set_source(name, path, options=None):
async with SessionLocal() as session:
row = (await session.execute(select(DataSource).where(DataSource.name == name))).scalar_one_or_none()
if row is None:
row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True)
session.add(row)
row.database, row.kind, row.enabled, row.readonly = str(path), "sqlite", True, True
row.options = dict(options or {})
await session.commit()
return row.id
@pytest.fixture(autouse=True)
async def sources(tmp_path):
shop, crm = tmp_path / "shop.db", tmp_path / "crm.db"
db = sqlite3.connect(shop)
db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, store TEXT, amount REAL);")
db.executemany("INSERT INTO orders VALUES (?, ?, ?)", [(1, "青石店", 1280.5), (2, "白沙店", 960.0)])
db.commit()
db.close()
db = sqlite3.connect(crm)
db.executescript("CREATE TABLE members (id INTEGER PRIMARY KEY, name TEXT, phone TEXT, spend REAL);")
db.executemany("INSERT INTO members VALUES (?, ?, ?, ?)", [(1, "林一", "000-0001", 320.0),
(2, "周二", "000-0002", 188.5)])
db.commit()
db.close()
ids = [await set_source("bundle_shop", shop), await set_source("bundle_crm", crm, {"mask_columns": ["phone"]})]
for source_id in ids:
await engines.invalidate(source_id)
yield
for source_id in ids:
await engines.invalidate(source_id)
def node(nid, ntype, **config):
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
def weekly(text_nodes=(), queries=()):
nodes = [
node("start", "input"),
node("fetch", "tool", tool="db_query__bundle_shop", args={"sql": SHOP_SQL}),
node("members", "tool", tool="db_query__bundle_crm", args={"sql": CRM_SQL}),
*queries,
node("card", "metrics", caliber="周报口径", caliber_version="v2", metrics=[
{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 1, "format": "thousands",
"expression": "cell(nodes.fetch, 0, 'amount')"}]),
node("write", "report", instructions="写周报", on_violation="flag"),
*text_nodes,
node("out", "output", fields=[{"name": "周报", "value": "{{ nodes.write.text }}"}]),
]
return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
def script(monkeypatch, text=TEXT):
from app.providers import mock_model
monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=text))
async def wait(run_id: str) -> Run:
for _ in range(400):
await asyncio.sleep(0.05)
async with SessionLocal() as session:
row = await session.get(Run, run_id)
if row.status in ("succeeded", "failed") and row.manifest_seq is not None:
return row
raise AssertionError(f"run {run_id} 没跑完:{row.status}")
async def sealed_run(monkeypatch, text=TEXT) -> Run:
script(monkeypatch, text)
row = await wait((await run_manager.start(graph=weekly(), input_payload={})).id)
assert row.status == "succeeded", row.error
return row
def data_of(html: str) -> dict:
m = re.search(r'<script type="application/json" id="bundle-data">(.*?)</script>', html, re.S)
assert m, "页面里没有数据块"
return json.loads(m.group(1))
async def query_artifact(run_id: str, node_id: str) -> str:
async with SessionLocal() as session:
[end] = (await session.execute(select(RunEvent).where(
RunEvent.run_id == run_id, RunEvent.type == "tool.end", RunEvent.node_id == node_id))).scalars()
return end.data["query_artifact"]
def file_of(artifact: str):
return settings.data_dir / "artifacts" / artifact[:2] / f"{artifact}.json"
# --------------------------------------------------------------------------
# 导出
# --------------------------------------------------------------------------
async def test_export_rebuilds_the_runners_manifest(client, monkeypatch):
row = await sealed_run(monkeypatch)
resp = await client.get(f"/api/runs/{row.id}/bundle.html")
assert resp.status_code == 200, resp.text
assert resp.headers["content-type"].startswith("text/html")
bundle = data_of(resp.text)
meta = bundle["meta"]
assert meta["run_id"] == row.id and meta["manifest_hash"] == row.manifest_hash
assert meta["manifest_seq"] == row.manifest_seq and meta["legacy"] is False
lines = [f"{e['seq']}|{e['type']}|{e['node_id'] or ''}|{e['data_sha256']}"
for e in sorted(bundle["events"], key=lambda e: e["seq"])]
assert hashlib.sha256("\n".join(lines).encode()).hexdigest() == row.manifest_hash
assert max(e["seq"] for e in bundle["events"]) == row.manifest_seq
async with SessionLocal() as session:
rows = {r.seq: r for r in (await session.execute(
select(RunEvent).where(RunEvent.run_id == row.id))).scalars()}
for item in bundle["events"]:
if "canonical" in item:
assert item["canonical"] == canonical_json(rows[item["seq"]].data)
assert content_hash(item["canonical"]) == item["data_sha256"]
types = {e["type"] for e in bundle["events"] if "canonical" in e}
assert {"run.started", "report.checked", "tool.end", "node.finished"} <= types
assert not types & {"llm.start", "llm.end", "node.started", "tool.start", "log"}
async def test_the_masked_source_ships_only_hashes(client, monkeypatch):
row = await sealed_run(monkeypatch)
shop, crm = await query_artifact(row.id, "fetch"), await query_artifact(row.id, "members")
bundle = data_of((await client.get(f"/api/runs/{row.id}/bundle.html")).text)
by_id = {a["id"]: a for a in bundle["artifacts"]}
assert by_id[crm] == {"id": crm, "kind": "query", "node_id": "members", "masked": True}
assert base64.b64decode(by_id[shop]["b64"]) == file_of(shop).read_bytes()
report = next(a for a in bundle["artifacts"] if a["kind"] == "report_doc")
assert json.loads(base64.b64decode(report["b64"]))["run_id"] == row.id
assert {a["kind"] for a in bundle["artifacts"]} == {"report_doc", "metric_set", "query"}
ends = {e["node_id"]: e for e in bundle["events"] if e["type"] == "tool.end"}
assert "canonical" in ends["fetch"] and "canonical" not in ends["members"]
# 运行里有带遮罩的快照:成果节点、运行结束事件的预览可能带出查询结果,只给哈希
out = next(e for e in bundle["events"] if e["type"] == "node.finished" and e["node_id"] == "out")
finished = next(e for e in bundle["events"] if e["type"] == "run.finished")
assert "canonical" not in out and "canonical" not in finished
assert "000-0001" not in json.dumps(bundle, ensure_ascii=False)
async def test_an_aggregate_on_the_masked_source_ships_its_snapshot(client, monkeypatch):
"""会员库设了号码遮罩,另一次查询只汇总了消费额(没碰号码列):数据源查询工具真落的快照随文件附带,原字节和服务端的
文件一致,查询事件带规范字符串,报告里引用它的数字和快照里那一格对得上;选出了号码的那份照旧只给 id。"""
script(monkeypatch, "会员消费合计 [[v:Q3.r0.spend]],会员林一 [[v:Q2.r0.spend]]。[[see:Q3]]")
total = node("total", "tool", tool="db_query__bundle_crm",
args={"sql": "SELECT COUNT(*) AS members, SUM(spend) AS spend FROM members"})
row = await wait((await run_manager.start(graph=weekly(queries=[total]), input_payload={})).id)
assert row.status == "succeeded", row.error
crm, summed = await query_artifact(row.id, "members"), await query_artifact(row.id, "total")
assert json.loads(file_of(summed).read_bytes())["mask_columns"] == ["phone"]
bundle = data_of((await client.get(f"/api/runs/{row.id}/bundle.html")).text)
by_id = {a["id"]: a for a in bundle["artifacts"]}
assert base64.b64decode(by_id[summed]["b64"]) == file_of(summed).read_bytes()
assert by_id[crm] == {"id": crm, "kind": "query", "node_id": "members", "masked": True}
ends = {e["node_id"]: e for e in bundle["events"] if e["type"] == "tool.end"}
assert "canonical" in ends["total"] and "canonical" not in ends["members"]
doc = json.loads(base64.b64decode(next(a["b64"] for a in bundle["artifacts"] if a["kind"] == "report_doc")))
assert doc["catalog"]["Q3"]["artifact"] == summed
[seg] = [s for b in doc["blocks"] for u in b["units"] for s in u["segments"]
if (s.get("cite") or {}).get("alias") == "Q3"]
snapshot = json.loads(base64.b64decode(by_id[summed]["b64"]))
assert seg["cite"]["status"] == "resolved" and seg["cite"]["value"] == snapshot["rows"][0][1] == 508.5
assert "000-0001" not in json.dumps(bundle, ensure_ascii=False)
async def test_the_page_is_named_after_the_run(client, monkeypatch):
row = await sealed_run(monkeypatch)
resp = await client.get(f"/api/runs/{row.id}/bundle.html")
header = resp.headers["content-disposition"]
assert header.startswith("attachment;")
assert f"filename*=UTF-8''{quote('未保存的工作流-可验证报告.html', safe='')}" in header
assert resp.headers.get("cache-control") == "no-store"
async with SessionLocal() as session:
run = await session.get(Run, row.id)
run.workflow_name = "门店周报"
await session.commit()
header = (await client.get(f"/api/runs/{row.id}/bundle.html")).headers["content-disposition"]
assert f"filename*=UTF-8''{quote('门店周报-可验证报告.html', safe='')}" in header
async def set_masks(name: str, columns: list[str]) -> None:
"""只改数据源的遮罩列(options.mask_columns),别的不动。"""
async with SessionLocal() as session:
row = (await session.execute(select(DataSource).where(DataSource.name == name))).scalar_one()
row.options = {**(row.options or {}), "mask_columns": columns}
await session.commit()
source_id = row.id
await engines.invalidate(source_id)
async def test_a_mask_set_after_the_run_still_applies(client, monkeypatch):
"""运行之后才给数据源设遮罩:快照里没记 mask_columns,导出时照样按数据源现在的遮罩挡(和证据面板一个道理)。"""
await set_masks("bundle_crm", [])
row = await sealed_run(monkeypatch)
shop = await query_artifact(row.id, "fetch")
def shown(bundle) -> dict[str, bool]:
events = {(e["type"], e["node_id"]): "canonical" in e for e in bundle["events"]}
return {"fetch": events[("tool.end", "fetch")], "members": events[("tool.end", "members")],
"out": events[("node.finished", "out")], "finished": events[("run.finished", None)]}
# 对照:查询时、导出时都没有遮罩,快照随文件附带,成果和运行结束事件带规范字符串
before = data_of((await client.get(f"/api/runs/{row.id}/bundle.html")).text)
assert all("b64" in a for a in before["artifacts"])
assert shown(before) == {"fetch": True, "members": True, "out": True, "finished": True}
# 门店名事后设了遮罩:报告引用的那份流水快照(选出了门店名)只给 id,查询事件、成果、运行结束事件只给哈希
await set_masks("bundle_shop", ["store"])
after = data_of((await client.get(f"/api/runs/{row.id}/bundle.html")).text)
assert next(a for a in after["artifacts"] if a["id"] == shop) == {
"id": shop, "kind": "query", "node_id": "fetch", "masked": True}
assert shown(after) == {"fetch": False, "members": True, "out": False, "finished": False}
assert "b64" in next(a for a in after["artifacts"] if a["kind"] == "report_doc")
# 金额事后设了遮罩:口径卡的销售额就是流水第一行的金额(cell(nodes.fetch, 0, 'amount')),写作时还没遮,报告文档的
# 证据目录里写着原值——和事后设的遮罩同样生效,整份文档只给哈希,正文的节点完成事件也只给哈希
await set_masks("bundle_shop", ["amount"])
late = data_of((await client.get(f"/api/runs/{row.id}/bundle.html")).text)
[doc] = [a for a in late["artifacts"] if a["kind"] == "report_doc"]
assert doc == {"id": doc["id"], "kind": "report_doc", "node_id": "write", "masked": True}, doc
write = next(e for e in late["events"] if e["type"] == "node.finished" and e["node_id"] == "write")
assert "canonical" not in write
async def test_report_text_cannot_close_the_data_block(client, monkeypatch):
row = await sealed_run(monkeypatch, text="结论见下:</script><script>alert(1)</script> 流水 [[v:Q1.r0.amount]]。")
html = (await client.get(f"/api/runs/{row.id}/bundle.html")).text
assert html.count("</script>") == 2
bundle = data_of(html)
write = next(e for e in bundle["events"] if e["type"] == "node.finished" and e["node_id"] == "write")
assert "</script><script>alert(1)</script>" in write["canonical"]
async def test_events_after_the_seal_are_listed_apart(client, monkeypatch):
row = await sealed_run(monkeypatch)
async with SessionLocal() as session:
session.add(RunEvent(run_id=row.id, seq=row.manifest_seq + 3, type="evidence.judged", node_id="write",
ts=0.0, data={"verdicts": {"u1": {"verdict": "supported"}}, "model": "judge-x",
"post_seal": True}))
await session.commit()
resp = await client.get(f"/api/runs/{row.id}/bundle.html")
assert resp.status_code == 200, resp.text
bundle = data_of(resp.text)
assert max(e["seq"] for e in bundle["events"]) == row.manifest_seq
[later] = bundle["appended"]
assert later["seq"] == row.manifest_seq + 3 and later["type"] == "evidence.judged"
# --------------------------------------------------------------------------
# 拒绝、只读
# --------------------------------------------------------------------------
async def test_an_unsealed_run_is_refused(client):
async with SessionLocal() as session:
run = Run(workflow_name="门店周报", status="running", graph={"nodes": [], "edges": []})
session.add(run)
await session.commit()
run_id = run.id
resp = await client.get(f"/api/runs/{run_id}/bundle.html")
assert resp.status_code == 409
assert resp.json() == {"detail": "运行尚未结束封存,结束后才能导出", "code": "bundle_unsealed"}
@pytest.mark.parametrize("status", ["running", "interrupted", "queued"])
async def test_a_run_going_again_on_an_old_seal_is_refused(client, monkeypatch, status):
"""失败后接着跑的运行留着上一次的封存,跑完会重新封存、换掉清单哈希;早年的运行停在审批时也封存过。
这时导出去的文件,封存哈希和系统里之后记的对不上,页面上「去系统里对照」那一步就作废了:一样回 409。"""
row = await sealed_run(monkeypatch)
async with SessionLocal() as session:
run = await session.get(Run, row.id)
run.status = status
await session.commit()
assert (await client.get(f"/api/runs/{row.id}/verify")).json()["ok"] is True
resp = await client.get(f"/api/runs/{row.id}/bundle.html")
assert resp.status_code == 409
assert resp.json() == {"detail": "运行尚未结束封存,结束后才能导出", "code": "bundle_unsealed"}
async def test_a_failed_run_is_exportable(client, monkeypatch):
"""失败、取消的运行同样封存了:报告已经写出来的,照样导得出去。"""
row = await sealed_run(monkeypatch)
async with SessionLocal() as session:
run = await session.get(Run, row.id)
run.status = "failed"
await session.commit()
resp = await client.get(f"/api/runs/{row.id}/bundle.html")
assert resp.status_code == 200, resp.text
assert data_of(resp.text)["meta"]["status"] == "failed"
async def test_a_rejected_run_carries_its_end_reason(client, monkeypatch):
"""被审批人驳回结束的运行(cancelled + end_reason=rejected,05-已知遗留 (a)):可验证报告和报告导出的元信息都带上
end_reason,页面照它写「被驳回」,不写成手动停止的「已取消」。"""
row = await sealed_run(monkeypatch)
async with SessionLocal() as session:
run = await session.get(Run, row.id)
run.status, run.end_reason = "cancelled", "rejected"
await session.commit()
bundle = await client.get(f"/api/runs/{row.id}/bundle.html")
assert bundle.status_code == 200, bundle.text
meta = data_of(bundle.text)["meta"]
assert (meta["status"], meta["end_reason"]) == ("cancelled", "rejected")
report = await client.get(f"/api/runs/{row.id}/report.html")
assert report.status_code == 200, report.text
m = re.search(r'<script type="application/json" id="report-data">(.*?)</script>', report.text, re.S)
assert m and json.loads(m.group(1))["meta"]["end_reason"] == "rejected"
async def test_a_broken_seal_is_refused(client, monkeypatch):
row = await sealed_run(monkeypatch)
async with SessionLocal() as session:
event = (await session.execute(select(RunEvent).where(
RunEvent.run_id == row.id, RunEvent.type == "report.checked"))).scalar_one()
event.data = {**event.data, "ok": not event.data.get("ok")}
await session.commit()
resp = await client.get(f"/api/runs/{row.id}/bundle.html")
assert resp.status_code == 409
# 原样出现在记录页的错误提示里:一句说清,不叠冒号、不把封存核对的整句结论再拼进来
assert resp.json() == {"detail": "封存核对不一致:封存后有事件被修改、删除或插入,不能导出可验证报告。"
"点「核对」可查看核对结果", "code": "bundle_seal_broken"}
async def test_an_unknown_run_is_404(client):
resp = await client.get("/api/runs/nope/bundle.html")
assert resp.status_code == 404
async def test_exporting_appends_nothing(client, monkeypatch):
row = await sealed_run(monkeypatch)
async def snapshot():
async with SessionLocal() as session:
run = await session.get(Run, row.id)
count = (await session.execute(
select(func.count()).select_from(RunEvent).where(RunEvent.run_id == row.id))).scalar_one()
return run.last_seq, count, run.manifest_hash, run.manifest_seq, run.status
before = await snapshot()
for _ in range(2):
assert (await client.get(f"/api/runs/{row.id}/bundle.html")).status_code == 200
assert await snapshot() == before
assert (await client.get(f"/api/runs/{row.id}/verify")).json()["ok"] is True
# --------------------------------------------------------------------------
# 工件库:read_bytes
# --------------------------------------------------------------------------
async def test_read_bytes_returns_the_file_as_stored():
artifact = await put_json({"门店": "青石店", "流水": 1280.5}, kind="test")
assert read_bytes(artifact) == file_of(artifact).read_bytes()
assert read_bytes(artifact) == canonical_json({"门店": "青石店", "流水": 1280.5}).encode()
assert read_bytes("0" * 64) is None and read_bytes("../etc/passwd") is None and read_bytes("") is None
async def test_read_bytes_checks_the_hash_unless_told_not_to():
artifact = await put_json({"门店": "白沙店", "流水": 960.0}, kind="test")
path = file_of(artifact)
original = path.read_bytes()
path.write_bytes(original.replace(b"960.0", b"999.0"))
try:
with pytest.raises(ValueError):
read_bytes(artifact)
assert read_bytes(artifact, verify=False) == original.replace(b"960.0", b"999.0")
finally:
path.write_bytes(original)