diff --git a/backend/app/api/evidence.py b/backend/app/api/evidence.py index e647c40c..2b6a4a2e 100644 --- a/backend/app/api/evidence.py +++ b/backend/app/api/evidence.py @@ -1466,7 +1466,9 @@ def _located(locator: dict[str, Any]) -> list[tuple[Any, Any]]: async def _caliber_source(source: Any) -> dict[str, Any] | None: - """钉住别处口径卡(caliber_from)时,定义取自哪个工作流的哪一版;带上工作流现在的名字方便认。""" + """钉住别处口径卡(caliber_from)时,定义取自哪个工作流的哪一版;带上工作流现在的名字方便认。取数节点拷进这张图时 + 改了名的(caliber_from.nodes,next-round 02 做法 4),来源里照带 nodes 映射:指标的输入写的是改名后的节点,面板据此 + 说清它们对应那一版里的哪个取数节点。""" if not isinstance(source, dict) or not source.get("workflow_id"): return None async with SessionLocal() as session: @@ -1478,7 +1480,11 @@ def _caliber_upgrade(event: dict[str, Any] | None) -> dict[str, Any] | None: """这张口径卡在正式运行发起时的升版处置:钉在哪一版、上游最新是哪一版、怎么处置的。""" if not isinstance(event, dict): return None - return {k: event.get(k) for k in ("workflow_id", "caliber_node", "pinned", "latest", "policy", "policy_label")} + out = {k: event.get(k) for k in ("workflow_id", "caliber_node", "pinned", "latest", "policy", "policy_label")} + if isinstance(event.get("caliber_nodes"), dict): + # 取数节点拷进这张图时改了名(caliber_from.nodes):照给映射;没改名的卡没有这个键,老运行一字不差 + out["caliber_nodes"] = dict(event["caliber_nodes"]) + return out #: 双盲复算节点产出里一个指标的结果(engine/nodes/dual_check.py 的 SAME / DIFFERENT / UNCHECKED),原样交给界面 diff --git a/backend/app/engine/compare.py b/backend/app/engine/compare.py index d4a0a4c5..49839bf1 100644 --- a/backend/app/engine/compare.py +++ b/backend/app/engine/compare.py @@ -65,7 +65,7 @@ from app.engine.evidence import DEFAULT_FORMAT, RenderError, is_number, render_n from app.engine.expressions import ExpressionError, cell_value, leaf_refs, parse_expression from app.engine.labels import TYPE_LABEL, field_label from app.engine.merge_query import MERGE_SOURCE -from app.engine.nodes.metrics import _REF_HEAD, _display_of +from app.engine.nodes.metrics import _REF_HEAD, _display_of, source_expression from app.engine.toolcalls import QUERY_PREFIX UNCHANGED = "unchanged" @@ -568,9 +568,13 @@ def _shown(value: Any) -> str: return text if len(text) <= _SHOWN_MAX else text[:_SHOWN_MAX] + "…" -def _expression(metric: dict[str, Any]) -> str: - """表达式的规范写法:空白、括号写法不同的算同一个。解析不了的照原文。""" - raw = str(metric.get("expression") or "").strip() +def _expression(metric: dict[str, Any], source: Any = None) -> str: + """表达式的规范写法:空白、括号写法不同的算同一个。解析不了的照原文。 + + source 是这张卡的来源(指标集上的 source):钉住别处口径卡、取数节点拷进图里时改了名的(caliber_from.nodes, + next-round 02 做法 4),先按映射换回那一版里的写法再比——同一份口径换了节点名不算口径变了;取数本身变没变照旧按 + 输入的出处说。""" + raw = source_expression(str(metric.get("expression") or "").strip(), source) try: return ast.unparse(parse_expression(raw)[0]) except (ExpressionError, ValueError): @@ -1715,7 +1719,7 @@ def _caliber_reasons(bm: dict[str, Any], tm: dict[str, Any], bc: dict[str, Any], defs: tuple[dict[str, Any] | None, dict[str, Any] | None] = (None, None)) -> list[str]: """口径变了的地方,一处一句;没变返回空列表。defs 是两次运行图里这个指标的定义(老卡没记小数位、格式时用)。""" out = [] - if _expression(bm) != _expression(tm): + if _expression(bm, bc.get("source")) != _expression(tm, tc.get("source")): out.append(f"表达式由「{_shown(bm.get('expression'))}」改为「{_shown(tm.get('expression'))}」") before, after = str(bc.get("caliber_version") or ""), str(tc.get("caliber_version") or "") if before != after: diff --git a/backend/app/engine/governance.py b/backend/app/engine/governance.py index 75a33045..4f78c8f2 100644 --- a/backend/app/engine/governance.py +++ b/backend/app/engine/governance.py @@ -699,6 +699,10 @@ async def unresolved_caliber_upgrades( ref = node.config["caliber_from"] pinned, wf_id = ref.get("workflow_version"), ref.get("workflow_id") extra = {"caliber_node": ref.get("node_id")} + if isinstance(ref.get("nodes"), dict) and ref["nodes"]: + # 拷进这张图时改了名的取数节点(next-round 02 做法 4):升版事件照记映射,证据面板说得清这张卡读的是哪几个 + # 节点;升不升版、怎么处置和没改名的卡同一套规则 + extra["caliber_nodes"] = dict(ref["nodes"]) what = f"「{node.title}」(口径卡)引用的口径卡固定在 v{pinned}" else: continue diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py index 06649df3..c289fdc9 100644 --- a/backend/app/engine/nodes/metrics.py +++ b/backend/app/engine/nodes/metrics.py @@ -1,5 +1,6 @@ from __future__ import annotations +import ast import json import math import re @@ -282,8 +283,87 @@ async def _pinned_caliber(ctx: NodeContext) -> tuple[dict[str, Any], dict[str, A raise NodeError(ctx.node.id, f"「{name}」v{version} 中的「{title}」本身也引用自其他工作流的口径卡(源头是 {where})。" "请直接固定到源头的口径卡,否则升版时无法正确比对版本") source = {"workflow_id": wf_id, "workflow_version": version, "node_id": node_id} + metrics = cfg.get("metrics") or [] + rename = ref.get("nodes") if isinstance(ref, dict) else None + if rename not in (None, {}): + # 节点改名映射(next-round 02 做法 4):那一版的式子读 nodes.fetch,这张图里拷过来的取数节点叫 c1_fetch。式子按 + # 语法树把 nodes.<原 id> 的根换成新 id 再算;来源里照记映射,证据、对比、升版处置据此认得是同一份口径 + if problem := rename_problem(rename): + raise NodeError(ctx.node.id, f"「{field_label('caliber_from')}」的节点改名映射有误:{problem}") + metrics = [rename_definition(m, rename) for m in metrics] + source["nodes"] = dict(rename) return {"caliber": cfg.get("caliber") or title, "caliber_version": str(cfg.get("caliber_version") or "v1"), - "metrics": cfg.get("metrics") or []}, source + "metrics": metrics}, source + + +#: 改名映射里的节点 id:要能写进式子的取值链(nodes.),和画布上的节点 id 同一种写法 +_NODE_ID = re.compile(r"^[A-Za-z_][\w-]*$") + + +def rename_problem(rename: Any) -> str | None: + """caliber_from.nodes 的形状对不对:{原 id: 新 id},两边都是非空的节点 id,新 id 不重复。对的是 None。 + 画布校验(engine/schema._caliber_from_issues)和运行时同一个判断;新 id 在不在这张图里由校验另查。""" + if not isinstance(rename, dict) or not rename: + return "需要写成「原节点 id → 这张图里的节点 id」的对照" + seen: set[str] = set() + for old, new in rename.items(): + if not isinstance(old, str) or not _NODE_ID.match(old) or not isinstance(new, str) or not _NODE_ID.match(new): + return f"「{old}」→「{new}」不是节点 id 的写法" + if new in seen: + return f"有两个节点都改名成了「{new}」" + seen.add(new) + return None + + +class _RenameNodes(ast.NodeTransformer): + """nodes.<原 id>(含 cell(nodes.<原 id>, …) 的第一个参数、nodes['原 id'])的根换成新 id。别的根(vars、input)不动。""" + + def __init__(self, rename: dict[str, str]) -> None: + self.rename = rename + self.hits = 0 + + def visit_Attribute(self, node: ast.Attribute) -> ast.AST: + if isinstance(node.value, ast.Name) and node.value.id == "nodes" and node.attr in self.rename: + self.hits += 1 + return ast.copy_location(ast.Attribute(value=node.value, attr=self.rename[node.attr], ctx=node.ctx), node) + return self.generic_visit(node) + + def visit_Subscript(self, node: ast.Subscript) -> ast.AST: + key = node.slice + if isinstance(node.value, ast.Name) and node.value.id == "nodes" and isinstance(key, ast.Constant) \ + and isinstance(key.value, str) and key.value in self.rename: + self.hits += 1 + return ast.copy_location(ast.Subscript(value=node.value, slice=ast.Constant(self.rename[key.value]), + ctx=node.ctx), node) + return self.generic_visit(node) + + +def rename_expression(expression: str, rename: dict[str, str]) -> str: + """把式子里 nodes.<原 id> 的根按映射换成新 id(语法树上换,字符串字面量里的字不动)。一处都没换到、解析不了的原样交回: + 解析不了的式子后面求值时照常报「表达式有误」。""" + try: + tree = ast.parse(expression, mode="eval") + except SyntaxError: + return expression + renamer = _RenameNodes(rename) + tree = ast.fix_missing_locations(renamer.visit(tree)) + return ast.unparse(tree) if renamer.hits else expression + + +def rename_definition(definition: Any, rename: dict[str, str]) -> Any: + """一个指标定义按映射改写式子(别的项原样)。不是对象的原样交回,后面照常报「指标定义不完整」。""" + if not isinstance(definition, dict) or not isinstance(definition.get("expression"), str): + return definition + return {**definition, "expression": rename_expression(definition["expression"], rename)} + + +def source_expression(expression: str, source: Any) -> str: + """卡上记的式子换回那一版里的写法(来源里有改名映射时按映射反着换):对比两次运行的口径(engine/compare)按它比, + 同一份口径拷进图里时取数节点改了名,不算口径变了。没有映射的原样交回。""" + rename = source.get("nodes") if isinstance(source, dict) else None + if not isinstance(rename, dict) or not rename or rename_problem(rename) is not None: + return expression + return rename_expression(expression, {new: old for old, new in rename.items()}) #: 新写法的小数位范围。负数是取整到十位、百位(round(x, -2)) diff --git a/backend/app/engine/schema.py b/backend/app/engine/schema.py index 7284f28b..8f6fca69 100644 --- a/backend/app/engine/schema.py +++ b/backend/app/engine/schema.py @@ -1443,7 +1443,7 @@ def evidence_issues(spec: GraphSpec) -> list[ValidationIssue]: "evidence_role", level="error") elif node.type == NodeType.METRICS: if cfg.get("caliber_from") not in (None, "", {}): - out.extend(_caliber_from_issues(node, UPGRADE_POLICIES, UPGRADE_POLICY_LABELS)) + out.extend(_caliber_from_issues(node, UPGRADE_POLICIES, UPGRADE_POLICY_LABELS, spec)) warned: set[str] = set() for i, producer in _caliber_inputs(node, spec): role = str(producer.config.get("evidence_role") or "compute") @@ -1457,8 +1457,12 @@ def evidence_issues(spec: GraphSpec) -> list[ValidationIssue]: return out -def _caliber_from_issues(node: GraphNode, policies: set[str], labels: dict[str, str]) -> list[ValidationIssue]: - """钉住别的工作流里一张口径卡:{workflow_id, workflow_version, node_id} 三项都要写对。""" +def _caliber_from_issues(node: GraphNode, policies: set[str], labels: dict[str, str], + spec: GraphSpec | None = None) -> list[ValidationIssue]: + """钉住别的工作流里一张口径卡:{workflow_id, workflow_version, node_id} 三项都要写对。 + + 可选的 nodes(节点改名映射,next-round 02 做法 4):那一版口径卡读的取数节点拷进这张图时改了名,{原 id: 这张图里的 id}。 + 形状和运行时同一个判断(metrics.rename_problem);改名后的节点要在这张图里、在口径卡上游——不然运行时口径卡取不到输入。""" cfg = node.config ref = cfg.get("caliber_from") out: list[ValidationIssue] = [] @@ -1475,6 +1479,21 @@ def _caliber_from_issues(node: GraphNode, policies: set[str], labels: dict[str, or int(version) < 1: add(f"「{field_label('caliber_from')}」需要选择具体的版本号(1、2…),当前为「{version}」:不固定版本," "口径会随上游变化", "caliber_from.workflow_version") + if ref.get("nodes") not in (None, {}): + from app.engine.nodes.metrics import rename_problem + + if problem := rename_problem(ref.get("nodes")): + add(f"「{field_label('caliber_from')}」的节点改名映射有误:{problem}", "caliber_from.nodes") + elif spec is not None: + nodes = spec.node_map() + upstream = _ancestors(spec, node.id) + for old, new in ref["nodes"].items(): + if new not in nodes: + add(f"「{field_label('caliber_from')}」把取数节点「{old}」改名为「{new}」,但工作流中没有节点「{new}」", + "caliber_from.nodes") + elif new not in upstream: + add(f"「{field_label('caliber_from')}」改名后的取数节点「{nodes[new].title}」不在口径卡的上游:口径卡计算时" + "它还没有执行。请把它连到口径卡之前", "caliber_from.nodes") if cfg.get("metrics"): add(f"已设置「{field_label('caliber_from')}」,将按所引用的口径卡计算,此处填写的「{field_label('metrics')}」" "不会生效", "metrics", level="warning") diff --git a/backend/tests/test_caliber_from_nodes.py b/backend/tests/test_caliber_from_nodes.py new file mode 100644 index 00000000..be4a6800 --- /dev/null +++ b/backend/tests/test_caliber_from_nodes.py @@ -0,0 +1,247 @@ +"""口径卡钉住别处口径卡时的节点改名映射(caliber_from.nodes,next-round 02 做法 4 ③)。 + +那一版的口径卡读 cell(nodes.fetch, 0, 'gmv');拷进这张图时取数节点改名成 c1_fetch(组合式问数、对账、召回同一张图里放 +两份口径会撞 id),caliber_from 带 nodes={fetch: c1_fetch}。要一起认这个映射的几处: + +- 运行时(metrics._pinned_caliber):式子按语法树把 nodes.fetch 的根换成 c1_fetch 再算,指标集的 source 照记映射; +- 画布校验(schema._caliber_from_issues):映射的形状、改名后的节点在不在图里、在不在口径卡上游; +- 升版检查(governance.unresolved_caliber_upgrades):照常挡、照常记事件,事件带上映射; +- 证据(指标步骤的 source、输入链到改名后的那条查询)、推演(按改名后的式子原样重算); +- 运行对比(compare):同一份口径换了节点名不算口径变了。 + +没写 nodes 的老图行为一字不差:式子原文、指标集、升版事件都和以前一样(单独测)。数据是虚构店铺库。 +""" +from __future__ import annotations + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + +from app.core import artifact_store +from app.db.base import SessionLocal +from app.db.models import RunEvent +from app.engine.nodes.metrics import rename_expression, rename_problem, source_expression +from app.engine.runner import run_manager +from app.engine.schema import GraphSpec, validate_graph +from app.main import app +from tests import ledger_samples as S + +RENDERED = ("SELECT SUM(amount) AS gmv FROM orders WHERE status = 'paid' " + "AND paid_at >= '2026-09-01' AND paid_at < '2026-09-08'") + + +@pytest.fixture(autouse=True) +async def engine_up(tmp_path): + from app.ledger import match as ledger_match + + await run_manager.setup() + await S.fresh_source(tmp_path / "shop.db") + yield + await ledger_match.drain() + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +def composed(wf: str, *, rename: dict | None = None, fetch_id: str = "c1_fetch", card_id: str = "c1", + version: int = 1, **card) -> dict: + ref = {"workflow_id": wf, "workflow_version": version, "node_id": "card"} + if rename is not None: + ref["nodes"] = rename + return S.chain(S.node("start", "input"), S.node(fetch_id, "tool", tool=S.tool(), args={"sql": RENDERED}), + S.node(card_id, "metrics", caliber_from=ref, **card), + S.node("out", "output", fields=[{"name": "r", "value": f"{{{{ nodes.{card_id}.text }}}}"}])) + + +async def finished(run_id: str, node_id: str) -> RunEvent: + async with SessionLocal() as session: + return (await session.execute(select(RunEvent).where( + RunEvent.run_id == run_id, RunEvent.type == "node.finished", RunEvent.node_id == node_id))).scalars().one() + + +async def run(graph: dict): + return await S.wait((await run_manager.start(graph=graph, input_payload={})).id) + + +# -------------------------------------------------------------------------- +# 改写式子 +# -------------------------------------------------------------------------- + + +def test_renaming_rewrites_only_the_roots_of_node_chains(): + rename = {"fetch": "c1_fetch"} + assert rename_expression("cell(nodes.fetch, 0, 'gmv')", rename) == "cell(nodes.c1_fetch, 0, 'gmv')" + assert rename_expression("nodes.fetch.rows[0]['gmv'] / 2", rename) == "nodes.c1_fetch.rows[0]['gmv'] / 2" + assert rename_expression("nodes['fetch'].row_count", rename) == "nodes['c1_fetch'].row_count" + # 字符串里的字、别的根、别的节点不动 + assert rename_expression("cell(nodes.other, 0, 'nodes.fetch') + vars.fetch", rename) == \ + "cell(nodes.other, 0, 'nodes.fetch') + vars.fetch" + # 一处都没换到的原样交回(不按 unparse 改写空白) + assert rename_expression("cell(nodes.other,0,'gmv')", rename) == "cell(nodes.other,0,'gmv')" + assert rename_expression("cell(nodes.fetch,", rename) == "cell(nodes.fetch," + # 反着换回那一版里的写法:对比口径按它比 + assert source_expression("cell(nodes.c1_fetch, 0, 'gmv')", {"nodes": rename}) == "cell(nodes.fetch, 0, 'gmv')" + assert source_expression("cell(nodes.c1_fetch, 0, 'gmv')", {"workflow_id": "w"}) == "cell(nodes.c1_fetch, 0, 'gmv')" + + +def test_mapping_shape_problems_are_named(): + assert rename_problem({"fetch": "c1_fetch"}) is None + assert rename_problem([]) and rename_problem({}) + assert "节点 id" in rename_problem({"fetch": "c1 fetch"}) + assert "两个节点" in rename_problem({"a": "x", "b": "x"}) + + +# -------------------------------------------------------------------------- +# 运行时、证据、推演 +# -------------------------------------------------------------------------- + + +async def test_a_renamed_fetch_feeds_the_pinned_card(client): + wf = await S.publish() + row = await run(composed(wf, rename={"fetch": "c1_fetch"})) + assert row.status == "succeeded", row.error + event = await finished(row.id, "c1") + card = artifact_store.load(event.data["artifact"]) + assert (card["caliber"], card["caliber_version"]) == ("周销售", "v1") + [metric] = card["metrics"] + assert metric["value"] == 170.5 and metric["rendered"] == "170.50元" + # 卡上记的是这张图里实际算的式子;来源照记映射 + assert metric["expression"] == "cell(nodes.c1_fetch, 0, 'gmv')" + assert card["source"] == {"workflow_id": wf, "workflow_version": 1, "node_id": "card", + "nodes": {"fetch": "c1_fetch"}} + [entry] = event.data["evidence"] + assert entry["source"]["nodes"] == {"fetch": "c1_fetch"} + [item] = metric["inputs"] + assert item["via"] == "tool_cell" and item["node_id"] == "c1_fetch" and item["artifact"] + + # 证据:指标步骤的来源带映射,输入链到改名后的那条查询的那一格 + resp = await client.get(f"/api/runs/{row.id}/evidence/metrics/c1/gmv") + assert resp.status_code == 200, resp.text + body = resp.json() + [step] = [s for s in body["chain"] if s["step"] == "metric"] + assert step["source"]["nodes"] == {"fetch": "c1_fetch"} and step["source"]["workflow_name"] == "周销售" + [inp] = [s for s in body["chain"] if s["step"] == "input"] + assert inp["node_id"] == "c1_fetch" and inp["artifact"] == item["artifact"] + + # 推演:按改名后的式子原样重算,改一个输入照常算 + resp = await client.post(f"/api/runs/{row.id}/evidence/whatif", + json={"card": "c1", "overrides": {"inputs": {"cell(nodes.c1_fetch, 0, 'gmv')": 200}}}) + assert resp.status_code == 200, resp.text + [m] = resp.json()["metrics"] + assert m["old"] == 170.5 and m["new"] == 200 and m["rendered_new"] == "200.00元" + + +async def test_without_the_mapping_the_renamed_fetch_is_not_found(): + """没写映射:式子照那一版读 nodes.fetch,这张图里没有它——缺输入,照实失败,不会算错。""" + wf = await S.publish() + row = await run(composed(wf)) + assert row.status == "failed" and "口径卡计算失败" in (row.error or "") + + +async def test_old_graphs_without_the_mapping_are_unchanged(): + """没写 nodes:式子原文一个字不改(不经 unparse),来源只有那三项。""" + graph = S.caliber_graph(expression="cell(nodes.fetch,0,'gmv')") + wf = await S.publish(graph=graph) + pinned = {"workflow_id": wf, "workflow_version": 1, "node_id": "card"} + row = await run(S.chain(S.node("start", "input"), S.node("fetch", "tool", tool=S.tool(), args={"sql": RENDERED}), + S.node("mine", "metrics", caliber_from=pinned), + S.node("out", "output", fields=[{"name": "r", "value": "{{ nodes.mine.text }}"}]))) + assert row.status == "succeeded", row.error + [entry] = (await finished(row.id, "mine")).data["evidence"] + card = artifact_store.load(entry["artifact"]) + assert card["source"] == pinned and entry["source"] == pinned + assert [m["expression"] for m in card["metrics"]] == ["cell(nodes.fetch,0,'gmv')"] + assert set(card) == {"kind", "caliber", "caliber_version", "metrics", "source"} + + +async def test_a_malformed_mapping_is_refused_before_running(): + wf = await S.publish() + with pytest.raises(ValueError, match="节点改名映射有误"): + await run_manager.start(graph=composed(wf, rename={"fetch": "c1 fetch"}), input_payload={}) + + +# -------------------------------------------------------------------------- +# 画布校验、升版检查 +# -------------------------------------------------------------------------- + + +def _codes(graph: dict) -> list[tuple[str | None, str]]: + return [(i.field, i.message) for i in validate_graph(GraphSpec.model_validate(graph)).issues if i.level == "error"] + + +def test_validation_checks_the_mapping(): + good = composed("wf1", rename={"fetch": "c1_fetch"}) + assert not [f for f, _ in _codes(good) if (f or "").startswith("caliber_from")] + shape = _codes(composed("wf1", rename={"fetch": ""})) + assert any(f == "caliber_from.nodes" and "映射有误" in m for f, m in shape) + missing = _codes(composed("wf1", rename={"fetch": "nowhere"})) + assert any(f == "caliber_from.nodes" and "没有节点「nowhere」" in m for f, m in missing) + # 改名后的节点在口径卡下游:算卡的时候它还没执行 + late = composed("wf1", rename={"fetch": "c1_fetch"}) + late["edges"] = [{"source": "start", "target": "c1"}, {"source": "c1", "target": "c1_fetch"}, + {"source": "c1_fetch", "target": "out"}] + assert any(f == "caliber_from.nodes" and "不在口径卡的上游" in m for f, m in _codes(late)) + + +async def test_upgrade_checks_treat_renamed_cards_like_any_pinned_card(): + from app.engine.governance import unresolved_caliber_upgrades + + wf = await S.publish() + await S.publish(version=2, workflow_id=wf) + async with SessionLocal() as session: + errors, _ = await unresolved_caliber_upgrades( + session, GraphSpec.model_validate(composed(wf, rename={"fetch": "c1_fetch"}))) + assert errors and "上游已有 v2" in errors[0] + _, [event] = await unresolved_caliber_upgrades(session, GraphSpec.model_validate( + composed(wf, rename={"fetch": "c1_fetch"}, upgrade_policy="dual"))) + assert event["caliber_nodes"] == {"fetch": "c1_fetch"} and event["pinned"] == 1 and event["latest"] == 2 + # 没改名的卡:事件和以前一样,没有这个键 + plain = composed(wf, fetch_id="fetch", card_id="mine", upgrade_policy="dual") + _, [old] = await unresolved_caliber_upgrades(session, GraphSpec.model_validate(plain)) + assert "caliber_nodes" not in old + + +# -------------------------------------------------------------------------- +# 运行对比 +# -------------------------------------------------------------------------- + + +def test_compare_sees_the_same_caliber_under_different_node_names(): + from app.engine.compare import _caliber_reasons + + def side(prefix: str) -> tuple[dict, dict]: + metric = {"id": "gmv", "expression": f"cell(nodes.{prefix}fetch, 0, 'gmv')", "decimals": 2, + "format": "thousands", "unit": "元"} + card = {"caliber": "周销售", "caliber_version": "v1", "metrics": [metric], + "source": {"workflow_id": "w", "workflow_version": 1, "node_id": "card", + **({"nodes": {"fetch": f"{prefix}fetch"}} if prefix else {})}} + return metric, card + + (bm, bc), (tm, tc) = side(""), side("c1_") + assert _caliber_reasons(bm, tm, bc, tc) == [] + # 映射反着换回来之后式子还是不同的,照旧说表达式改了 + tm2 = {**tm, "expression": "cell(nodes.c1_fetch, 0, 'gmv') * 2"} + assert any("表达式由" in r for r in _caliber_reasons(bm, tm2, bc, tc)) + + +async def test_runs_of_the_same_composed_graph_compare_unchanged(): + from app.db.models import Workflow + + wf = await S.publish() + graph = composed(wf, rename={"fetch": "c1_fetch"}) + async with SessionLocal() as session: + saved = Workflow(name="组合的答案", graph=graph) + session.add(saved) + await session.commit() + saved_id = saved.id + a, b = [await S.wait((await run_manager.start(graph=graph, workflow_id=saved_id, input_payload={})).id) + for _ in range(2)] + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + resp = await c.get(f"/api/runs/{b.id}/compare", params={"base": a.id}) + assert resp.status_code == 200, resp.text + [metric] = resp.json()["metrics"] + assert metric["classification"] == "unchanged", metric