From 97710463feb3c0697dfe25de945d214708baf7c1 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 3 Nov 2026 19:14:23 +0800 Subject: [PATCH] =?UTF-8?q?=E8=A1=A5=E5=AD=90=E5=9B=BE=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E8=BF=9B=E5=85=A5=E7=88=B6=E5=9B=BE=E8=B4=A6=E6=9C=AC=E6=AF=94?= =?UTF-8?q?=E5=AF=B9=E7=9A=84=E5=9B=9E=E5=BD=92=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/tests/test_subgraph_evidence.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/backend/tests/test_subgraph_evidence.py b/backend/tests/test_subgraph_evidence.py index f03dcc21..e971b378 100644 --- a/backend/tests/test_subgraph_evidence.py +++ b/backend/tests/test_subgraph_evidence.py @@ -20,6 +20,8 @@ from app.db.base import SessionLocal from app.db.models import RunEvent from app.engine.lineage import traced, untraced from app.engine.runner import run_manager +from app.ledger import match as ledger_match +from app.ledger import store as ledger_store from tests import ledger_samples as S PAID = "SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid'" @@ -528,6 +530,22 @@ async def test_the_bundle_and_whatif_read_a_subgraph_card_in_its_own_layer(clien assert cnt.get("new") == cnt.get("old") == 1, cnt +async def test_the_ledger_matches_the_subgraph_query_against_its_published_caliber(): + """子图的查询快照进了父图封存，账本就能把它和子图那一版的已发布口径比对出来。""" + sub_id = await publish(fetch_card_graph()) + started = await run_manager.start( + graph=parent(sub_id, contract_from=None, + out=[{"name": "数", "value": "{{ nodes.sub.output.gmv }}"}]), + input_payload={}, origin_kind="canvas") + row = await S.wait(started.id) + assert row.status == "succeeded", row.error + assert await ledger_match.match_run(row.id, trigger="on_demand") + recorded = await ledger_store.latest(row.id) + [query] = [q for q in (recorded.content or {}).get("queries") or [] if q.get("node_id") == "sub/fetch"] + assert query["status"] == "proven", query + assert query["candidates"][0]["caliber"] == f"{sub_id}@1" + + async def card_entries(run_id: str) -> list[dict]: from app.api.evidence import _sealed