diff --git a/backend/app/ledger/match.py b/backend/app/ledger/match.py index a05f5db8..109d37d8 100644 --- a/backend/app/ledger/match.py +++ b/backend/app/ledger/match.py @@ -67,6 +67,7 @@ SNAPSHOT_TAMPERED = "查询快照与哈希不一致,疑似被修改,不比 SNAPSHOT_MISSING = "查询快照读不出来,不比对" NO_SQL = "查询快照里没有 SQL,不比对" PARSE_FAILED = "这次的 SQL 无法解析,没有比对" +MERGED = "合并查询的格不和口径比对(它的数来自各个输入的查询)" MORE_CANDIDATES = "另有 {n} 份相关口径没有比对(每条查询最多比 {max} 份)" #: 不比对的原因(GET 账本时照实说) @@ -364,6 +365,10 @@ def compute(sealed: Any, calibers: list[Caliber], kinds: dict[str, str | None], by_key = {c.key: {"level": c.level, "fields": c.summary()["fields"], "expressions": list(c.expressions)} for c in considered} by_artifact = {q["artifact"]: q for q in results} + # 合并查询的快照不比(read_queries 跳过了它),引用了它的格照实写为什么没比对;不进工件的 queries,统计不数它 + for artifact in sealed.queries: + by_artifact.setdefault(artifact, {"artifact": artifact, "status": "skipped", "reason": MERGED, + "candidates": []}) sealed_cards = {artifact for _, artifact in sealed.ledger} def load_card(artifact: str) -> dict[str, Any] | None: diff --git a/backend/tests/ledger_samples.py b/backend/tests/ledger_samples.py index 4992edf0..585059b0 100644 --- a/backend/tests/ledger_samples.py +++ b/backend/tests/ledger_samples.py @@ -1,6 +1,7 @@ """口径账本测试共用的虚构样本:一个店铺库、一份受管口径「周销售」、跑一次探索运行。数据全是编的。 -- 店铺库 shop(SQLite):orders(id, region, status, amount, paid_at, phone),九单,状态有 paid / refunded / open。 +- 店铺库(SQLite):orders(id, region, status, amount, paid_at, phone),九单,状态有 paid / refunded / open。每个用例一个 + 名字不同的数据源(fresh_source):候选口径按数据源预筛,别的用例发布的口径不会混进来,也不必去动别的用例的数据。 - 口径「周销售」v1(受管):入口两个字段 start / end,取数节点 fetch 的 SQL 是 SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' AND paid_at >= '{{ input.start }}' AND paid_at < '{{ input.end }}', 口径卡 card 上一个指标 gmv = cell(nodes.fetch, 0, 'gmv')。 @@ -55,11 +56,30 @@ def build_shop(path) -> None: db.close() +#: 这个用例的数据源名(fresh_source 换):图里的查询工具是 db_query__<它> +CURRENT = {"source": "shop"} + + +def tool() -> str: + return f"db_query__{CURRENT['source']}" + + +async def fresh_source(path) -> str: + """建库、起一个这个用例专用的数据源(名字不重复),返回数据源名。""" + import uuid + + build_shop(path) + CURRENT["source"] = f"shop_{uuid.uuid4().hex[:8]}" + await set_source(path) + return CURRENT["source"] + + async def set_source(path, *, options: dict[str, Any] | None = None) -> str: + name = CURRENT["source"] async with SessionLocal() as session: - row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() + row = (await session.execute(select(DataSource).where(DataSource.name == name))).scalar_one_or_none() if row is None: - row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) + 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 {}) @@ -73,7 +93,7 @@ def caliber_graph(sql: str = TEMPLATE, *, expression: str = "cell(nodes.fetch, 0 return chain( node("start", "input", fields=[{"name": "start", "required": True}, {"name": "end", "required": True}, {"name": "store", "default": "all"}]), - node("fetch", "tool", tool="db_query__shop", args={"sql": sql}), + node("fetch", "tool", tool=tool(), args={"sql": sql}), node("card", "metrics", caliber="周销售", caliber_version="v1", metrics=[ {"id": "gmv", "name": "销售额", "unit": "元", "decimals": 2, "format": "thousands", "expression": expression}]), node("write", "report", instructions="写周销售"), @@ -99,7 +119,7 @@ async def publish(name: str = "周销售", *, graph: dict[str, Any] | None = Non def explore_graph(sql: str, *, card_expression: str | None = None) -> dict[str, Any]: - parts = [node("start", "input"), node("q", "tool", tool="db_query__shop", args={"sql": sql})] + parts = [node("start", "input"), node("q", "tool", tool=tool(), args={"sql": sql})] if card_expression: parts.append(node("mine", "metrics", caliber="我的口径", caliber_version="v1", metrics=[ {"id": "gmv", "name": "销售额", "unit": "元", "decimals": 2, "format": "thousands", diff --git a/backend/tests/test_ledger_api.py b/backend/tests/test_ledger_api.py index d3fda04a..885c2579 100644 --- a/backend/tests/test_ledger_api.py +++ b/backend/tests/test_ledger_api.py @@ -18,7 +18,7 @@ from httpx import ASGITransport, AsyncClient from sqlalchemy import select from app.db.base import SessionLocal -from app.db.models import AuditEvent, Workflow, WorkflowVersion +from app.db.models import AuditEvent, Workflow from app.engine.runner import run_manager from app.ledger import match as ledger_match from app.main import app @@ -27,16 +27,9 @@ from tests import ledger_samples as S @pytest.fixture(autouse=True) async def engine_up(tmp_path): - from app.db.base import utcnow - await run_manager.setup() path = tmp_path / "shop.db" - S.build_shop(path) - await S.set_source(path) - async with SessionLocal() as session: - for row in (await session.execute(select(WorkflowVersion).where(WorkflowVersion.retired_at.is_(None)))).scalars(): - row.retired_at = utcnow() - await session.commit() + await S.fresh_source(path) yield path await ledger_match.drain() await run_manager.shutdown() @@ -115,8 +108,8 @@ async def test_masked_binding_values_and_sql_fragments_are_withheld(client, monk await S.publish(graph=S.caliber_graph(template)) graph = S.chain( S.node("start", "input"), - S.node("p", "tool", tool="db_query__shop", args={"sql": "SELECT phone FROM orders ORDER BY id"}), - S.node("q", "tool", tool="db_query__shop", + S.node("p", "tool", tool=S.tool(), args={"sql": "SELECT phone FROM orders ORDER BY id"}), + S.node("q", "tool", tool=S.tool(), args={"sql": "SELECT SUM(amount) AS gmv FROM orders WHERE phone = '13800000001'"}), S.node("write", "report", instructions="写答案"), S.node("out", "output", fields=[{"name": "答案", "value": "{{ nodes.write.text }}"}])) diff --git a/backend/tests/test_ledger_match.py b/backend/tests/test_ledger_match.py index 3b7267ab..315efed1 100644 --- a/backend/tests/test_ledger_match.py +++ b/backend/tests/test_ledger_match.py @@ -30,16 +30,7 @@ from tests import ledger_samples as S async def engine_up(tmp_path): await run_manager.setup() path = tmp_path / "shop.db" - S.build_shop(path) - await S.set_source(path) - # 每个用例自己发布口径:先把别的用例发布的退役掉,候选口径只看这一个用例的 - from app.db.models import WorkflowVersion - from app.db.base import utcnow - - async with SessionLocal() as session: - for row in (await session.execute(select(WorkflowVersion).where(WorkflowVersion.retired_at.is_(None)))).scalars(): - row.retired_at = utcnow() - await session.commit() + await S.fresh_source(path) yield path await ledger_match.drain() await run_manager.shutdown() @@ -68,7 +59,7 @@ async def test_an_equivalent_query_is_proven_and_its_number_comes_from_the_calib assert content["format"] == store.FORMAT and content["algo"] == store.ALGO and content["run_id"] == row.id assert content["manifest_hash"] == row.manifest_hash [query] = content["queries"] - assert query["status"] == "proven" and query["source"] == "shop" and query["tables"] == ["orders"] + assert query["status"] == "proven" and query["source"] == S.CURRENT["source"] and query["tables"] == ["orders"] [cand] = query["candidates"] assert cand["caliber"] == f"{wf}@1" and cand["level"] == "governed" and cand["node_id"] == "fetch" assert cand["binding"] == {"start": "2026-09-01", "end": "2026-09-08"} @@ -283,7 +274,7 @@ async def test_only_sealed_query_snapshots_are_compared(monkeypatch): await S.publish() row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。") - await put_json({"sql": S.WITH_REFUNDS, "source": "shop", "columns": ["gmv"], "rows": [[1]]}, + await put_json({"sql": S.WITH_REFUNDS, "source": S.CURRENT["source"], "columns": ["gmv"], "rows": [[1]]}, kind="query_snapshot", run_id=row.id, node_id="q") await ledger_match.match_run(row.id, trigger="on_demand") recorded, _ = await matched(row.id) @@ -298,3 +289,17 @@ async def test_cells_inside_a_table_are_labelled_the_same_way(monkeypatch): [labels] = [r["labels"] for r in recorded.content["reports"].values()] assert len(labels) >= 2 and all(label["kind"] == "caliber" and label["caliber"] == f"{wf}@1" for label in labels.values()) + + +async def test_backfill_starts_in_the_background_when_the_scheduler_is_on(monkeypatch): + started = [] + + async def fake_backfill(**kw): + started.append(kw) + return 0 + + monkeypatch.setattr(settings, "scheduler", True) + monkeypatch.setattr(ledger_match, "backfill", fake_backfill) + assert ledger_match.start_backfill() is True + await ledger_match.drain() + assert started == [{}]