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