mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 01:51:57 +00:00
ThearchyHelios
d176ff77ff
出处判定（每个指标的分类靠它）： - 合并查询：读合并结果的指标按别名一路追到喂它的每一份上游快照（嵌套的合并也追），上游 SQL 结构变了是 sql_changed， 不拆、不归因；合并节点多接、少接了输入也算结构变化 - 图里写死的字面量改了（'paid' 改成 'refunded'、系数、LIMIT）：执行的 SQL 只差字面量，但图里的 SQL 改过，算模板变化； 图里一字未改（定时运行代入的日期）才是参数。图里的 SQL 只改了空白、大小写的不算改过 - Agent 按出处交来的字段：还要引用同一列、同一行（段），两边都引用整份结果的不算换了；换了列、换了行段的说不清 （incomparable），不再把没引用的区域算成「数据没了」去归因 - 模型、人直接交出的值（Agent 没按出处交、模型调用、人工审批……）：值变了说不清是数据还是模型本身，判 incomparable； 它自己查的库结构变了、配置变了照常标出 - 代码等确定性节点：上游的查询结果两次一模一样、交出的值却变了的，判 incomparable，不再说「变化来自上游数据」； 上游只有运行输入变了的算参数 - 图级变化落到节点上：连线有增删的终点节点、默认设置改了且没有自行设置这几项的模型类节点，都算改了配置； 「模板和取数的 SQL 都没有变化」只在整张图都没变时说 - 运行输入是列表、对象的，按还原出来的完整值比，长度不变、内容变了的是参数变化 - 同一节点按 SQL 配对时先配字面量也相同的：Agent 换了先后查东区、南区，两条各自配上自己，不再配成两条「只有参数不同」 - 一边记不出输入是哪个节点写的，说清是哪一边没记录（不再写出「None」） 口径与数值： - 口径变了的指标，喂它的取数也变了、说不清的照样写进说明（只在一边才有的输入是改表达式带来的，不重复说）； 分类次序写进 compare() 的说明 - 千分位和不分组互换只影响显示，写进说明、不算口径变化；比率与否不同才算 - 升级前的口径卡没记小数位、显示格式：按那次运行图里的指标定义核对，定义也没有就不比，不再误报「小数位由 未设置 改为 1」 - 百分比：基准极小、这次很大时十进制量化放不下，退回浮点算，不再整个对比报 500 - 没设小数位的指标拆分：三个数按同一精度量化（最多 15 位有效数字），转成浮点交出去后两效应之和仍逐位等于总变化 - 维度归因标出用的是哪个输入、两边各自哪份快照（path、base_artifact、target_artifact） 接口：已经结束、只是没有封存清单的运行，409 说「没有封存清单，不能用于对比」，不再说「运行结束后才能对比」。 测试：补上复核指出的必测项（只一边对得上不给占比、只截断基准一侧、换了数据源、按旧口径小数位取整、写它的节点多查一次、 口径和 SQL 一起变不归因、小数位和比率变化、日期类型和常量列不当维度、合计行浮点误差、只改名称、结束过又接着运行的不当 基准），以及上面每一处修正的用例。32 处变异逐条打红验证，全部被测试抓到。
393 lines
20 KiB
Python
393 lines
20 KiB
Python
"""运行对比的接口：GET /api/runs/{run_id}/compare/base 和 GET /api/runs/{run_id}/compare?base=。
|
|
|
|
两次周报运行用 tests/fixtures/compare_runs.py 的虚构运行：工件落进工件库，事件落进运行记录，按 runner 的算法封存
|
|
（不连库、不调模型）。另有一组真跑：虚构的 SQLite 库在两次运行之间改了数，再一次代进去的日期参数不同。守的是：
|
|
- 两次都结束、封存且核对一致才比：没封存、还在运行、封存后事件被改过都回 409，说清是哪一次；不是同一个工作流、
|
|
同一次运行回 422；运行不存在 404
|
|
- 对比只读：两次运行的 last_seq、事件、工件引用、工件文件一样不多
|
|
- 遮罩列（查询当时记下的、数据源现在设的）的值不出现在结果里
|
|
- 默认基准：同一定时任务优先，否则同一工作流、同一运行类别；只挑更早结束、封存过的终态运行
|
|
- 真跑：同一条 SQL 数据变了是 data_only；只是代进去的日期不同是 params_only，图一样
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import sqlite3
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy import func, select
|
|
|
|
from app.core.artifact_store import manifest_hash, put_json
|
|
from app.core.config import settings
|
|
from app.data.engine import engines
|
|
from app.db.base import SessionLocal
|
|
from app.db.models import Artifact, DataSource, Run, RunEvent, Workflow
|
|
from app.engine.runner import run_manager, verify_manifest
|
|
from app.main import app
|
|
from tests.fixtures.compare_runs import (
|
|
EAST,
|
|
NEXT_ROWS,
|
|
Fake,
|
|
amounts_in,
|
|
cell_in,
|
|
chain,
|
|
metric,
|
|
node,
|
|
week,
|
|
)
|
|
from tests.fixtures.sources import drop_source
|
|
|
|
SOURCE = "cmp_shop"
|
|
T0 = datetime(2026, 9, 7, 1, 0, tzinfo=timezone.utc)
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
async def new_workflow(name="区域周报") -> str:
|
|
async with SessionLocal() as session:
|
|
workflow = Workflow(name=name)
|
|
session.add(workflow)
|
|
await session.commit()
|
|
return workflow.id
|
|
|
|
|
|
async def persist(fake: Fake, workflow_id: str | None, *, finished=T0, status="succeeded", run_class="formal",
|
|
schedule_id=None, seal=True) -> str:
|
|
"""把一次虚构运行落库：工件进工件库（内容寻址，id 和虚构时一样），事件进运行记录，按 runner 的算法封存。"""
|
|
for content in fake.store.values():
|
|
await put_json(content, kind="compare_fixture")
|
|
async with SessionLocal() as session:
|
|
run = Run(workflow_id=workflow_id, workflow_name="区域周报", status=status, graph=fake.graph,
|
|
input=dict(fake.inputs), run_class=run_class, version=fake.version, schedule_id=schedule_id,
|
|
started_at=finished - timedelta(minutes=2), finished_at=finished if status != "running" else None,
|
|
agent_limits={})
|
|
session.add(run)
|
|
await session.flush()
|
|
rows = []
|
|
for seq, etype, node_id, data in fake.events:
|
|
session.add(RunEvent(run_id=run.id, seq=seq, type=etype, node_id=node_id, ts=float(seq), data=data))
|
|
rows.append((seq, etype, node_id, data))
|
|
run.last_seq = len(rows)
|
|
if seal:
|
|
run.manifest_hash, run.manifest_seq = manifest_hash(rows), rows[-1][0]
|
|
await session.commit()
|
|
return run.id
|
|
|
|
|
|
async def pair(base=None, target=None, **kw) -> tuple[str, str, str]:
|
|
"""(工作流, 基准运行, 这次运行)：同一个工作流的两次周报，这次的数变了。"""
|
|
workflow = await new_workflow()
|
|
base_id = await persist(base or week("r1", source=SOURCE), workflow, finished=T0)
|
|
target_id = await persist(target or week("r2", NEXT_ROWS, source=SOURCE), workflow,
|
|
finished=T0 + timedelta(days=7), **kw)
|
|
return workflow, base_id, target_id
|
|
|
|
|
|
async def footprint(*run_ids: str) -> tuple:
|
|
"""几次运行留在库里、盘上的痕迹：各自的 last_seq、封存、事件数，加上全部工件引用行数和工件文件数。"""
|
|
async with SessionLocal() as session:
|
|
runs = []
|
|
for run_id in run_ids:
|
|
run = await session.get(Run, run_id)
|
|
count = await session.scalar(select(func.count()).select_from(RunEvent).where(RunEvent.run_id == run_id))
|
|
runs.append((run.last_seq, run.manifest_hash, run.manifest_seq, run.status, count))
|
|
artifacts = await session.scalar(select(func.count()).select_from(Artifact))
|
|
files = sum(1 for _ in (settings.data_dir / "artifacts").rglob("*.json"))
|
|
return runs, artifacts, files
|
|
|
|
|
|
async def get(client, path: str, status: int = 200) -> dict:
|
|
resp = await client.get(path)
|
|
assert resp.status_code == status, resp.text
|
|
return resp.json()
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 对比
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
async def test_compares_two_sealed_runs(client):
|
|
_, base_id, target_id = await pair()
|
|
body = await get(client, f"/api/runs/{target_id}/compare?base={base_id}")
|
|
assert body["target"]["id"] == target_id and body["base"]["id"] == base_id
|
|
assert set(body["target"]) == {"id", "workflow_name", "version", "run_class", "started_at", "finished_at",
|
|
"schedule_id"}
|
|
assert body["target"]["finished_at"].endswith("+00:00") and body["base"]["run_class"] == "formal"
|
|
assert body["summary"]["data_only"] == 2 and sum(body["summary"].values()) == 2
|
|
east = next(m for m in body["metrics"] if m["metric_id"] == "east")
|
|
assert (east["classification"], east["base"], east["target"], east["delta_rendered"]) == (
|
|
"data_only", 1200.5, 1500.5, "+300.0元")
|
|
total = next(m for m in body["metrics"] if m["metric_id"] == "total")
|
|
assert [r["share"] for r in total["attribution"]["rows"]] == [1.5, -0.5, 0.0]
|
|
assert [(q["node_id"], q["structure"]) for q in body["queries"]] == [("fetch", "same"), ("bot", "same")]
|
|
assert body["template"] == {"base_version": 3, "target_version": 3, "same_graph": True, "nodes": []}
|
|
assert body["notes"] == []
|
|
|
|
|
|
async def test_comparing_writes_nothing(client):
|
|
_, base_id, target_id = await pair()
|
|
before = await footprint(base_id, target_id)
|
|
await get(client, f"/api/runs/{target_id}/compare?base={base_id}")
|
|
await get(client, f"/api/runs/{target_id}/compare/base")
|
|
assert await footprint(base_id, target_id) == before
|
|
for run_id in (base_id, target_id):
|
|
assert (await verify_manifest(run_id))["ok"] is True
|
|
|
|
|
|
async def test_unsealed_runs_are_refused(client):
|
|
workflow = await new_workflow()
|
|
base_id = await persist(week("r1", source=SOURCE), workflow)
|
|
target_id = await persist(week("r2", NEXT_ROWS, source=SOURCE), workflow, finished=T0 + timedelta(days=7),
|
|
seal=False)
|
|
body = await get(client, f"/api/runs/{target_id}/compare?base={base_id}", 409)
|
|
assert body["code"] == "compare_unsealed" and body["detail"].startswith("这次运行")
|
|
# 已经结束、只是没有封存清单的：说没有封存清单，不说「运行结束后才能对比」
|
|
assert "没有封存清单" in body["detail"] and "尚未结束" not in body["detail"]
|
|
body = await get(client, f"/api/runs/{base_id}/compare?base={target_id}", 409)
|
|
assert body["code"] == "compare_unsealed" and body["detail"].startswith("作为基准的运行")
|
|
# 封存过、又被接着运行（还在运行）的：封存范围之后还会变，同样不比
|
|
running = await persist(week("r3", NEXT_ROWS, source=SOURCE), workflow, status="running")
|
|
body = await get(client, f"/api/runs/{running}/compare?base={base_id}", 409)
|
|
assert body["code"] == "compare_unsealed" and "尚未结束" in body["detail"]
|
|
|
|
|
|
async def test_a_broken_seal_is_refused(client):
|
|
_, base_id, target_id = await pair()
|
|
async with SessionLocal() as session:
|
|
event = (await session.execute(select(RunEvent).where(
|
|
RunEvent.run_id == base_id, RunEvent.type == "tool.end"))).scalars().first()
|
|
event.data = {**event.data, "duration_ms": 999}
|
|
await session.commit()
|
|
body = await get(client, f"/api/runs/{target_id}/compare?base={base_id}", 409)
|
|
assert body["code"] == "compare_seal_broken" and body["detail"].startswith("作为基准的运行")
|
|
assert "封存核对不一致" in body["detail"]
|
|
|
|
|
|
async def test_mismatched_runs_are_refused(client):
|
|
_, base_id, target_id = await pair()
|
|
other = await persist(week("r9", source=SOURCE), await new_workflow("另一个周报"))
|
|
body = await get(client, f"/api/runs/{target_id}/compare?base={other}", 422)
|
|
assert body["code"] == "compare_mismatch" and "同一个工作流" in body["detail"]
|
|
body = await get(client, f"/api/runs/{target_id}/compare?base={target_id}", 422)
|
|
assert body["code"] == "compare_mismatch" and "同一次运行" in body["detail"]
|
|
unsaved = await persist(week("r8", source=SOURCE), None)
|
|
body = await get(client, f"/api/runs/{unsaved}/compare?base={base_id}", 422)
|
|
assert body["code"] == "compare_mismatch"
|
|
body = await get(client, f"/api/runs/{target_id}/compare", 422)
|
|
assert body["code"] == "compare_bad_request"
|
|
await get(client, f"/api/runs/no-such-run/compare?base={base_id}", 404)
|
|
body = await get(client, f"/api/runs/{target_id}/compare?base=no-such-run", 404)
|
|
assert "基准" in body["detail"]
|
|
|
|
|
|
async def test_masked_columns_never_leave_the_server(client):
|
|
"""快照里记着查询当时遮了 refund，数据源现在又遮了 manager：两样的值都不出现在结果里。"""
|
|
columns = ["region", "manager", "gmv", "refund"]
|
|
types = {"region": "text", "manager": "text", "gmv": "number", "refund": "number"}
|
|
sql = "SELECT region, manager, SUM(amount) AS gmv, SUM(refund) AS refund FROM orders GROUP BY region, manager"
|
|
|
|
def make(run_id, rows, expression, version):
|
|
run = Fake(run_id, source=SOURCE)
|
|
fetch = run.tool("fetch", sql, columns, rows, types=types, masks=["refund"])
|
|
bot = run.agent("bot", sql, columns, rows, types=types, masks=["refund"])
|
|
inputs = [cell_in(0, "gmv", rows[0][2], fetch)]
|
|
value = rows[0][2]
|
|
if "refund" in expression:
|
|
inputs.append(cell_in(0, "refund", rows[0][3], fetch))
|
|
value = rows[0][2] - rows[0][3]
|
|
run.card([metric("net", expression, value, inputs),
|
|
metric("total", "sum(vars.kpi.amounts)", sum(r[2] for r in rows), [amounts_in(bot, rows)])],
|
|
version=version)
|
|
return run
|
|
|
|
async with SessionLocal() as session:
|
|
session.add(DataSource(name=SOURCE, kind="sqlite", database=":memory:", options={"mask_columns": ["manager"]}))
|
|
await session.commit()
|
|
try:
|
|
base = make("r1", [["东区", "经理甲", 1200.5, 101.25], ["南区", "经理乙", 800.0, 52.75]], EAST, "v1")
|
|
target = make("r2", [["东区", "经理甲", 1500.5, 133.5], ["南区", "经理乙", 700.0, 47.25]],
|
|
f"{EAST} - cell(nodes.fetch, 0, 'refund')", "v2")
|
|
_, base_id, target_id = await pair(base, target)
|
|
resp = await client.get(f"/api/runs/{target_id}/compare?base={base_id}")
|
|
assert resp.status_code == 200, resp.text
|
|
for secret in ("经理甲", "经理乙", "101.25", "52.75", "133.5", "47.25"):
|
|
assert secret not in resp.text, secret
|
|
body = resp.json()
|
|
net = next(m for m in body["metrics"] if m["metric_id"] == "net")
|
|
refund = next(i for i in net["inputs"] if "refund" in i["path"])
|
|
assert refund["masked"] is True and net["decomposition"] is None and "遮罩" in net["decomposition_note"]
|
|
total = next(m for m in body["metrics"] if m["metric_id"] == "total")
|
|
assert total["attribution"]["columns"] == ["region"]
|
|
finally:
|
|
await drop_source(name=SOURCE)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 默认基准
|
|
# --------------------------------------------------------------------------
|
|
|
|
|
|
async def bare(workflow_id, finished, *, status="succeeded", run_class="formal", schedule_id=None, sealed=True,
|
|
resumed=False) -> str:
|
|
"""只为挑基准用的运行记录：挑基准只看运行记录上的几列。resumed：结束、封存过又被接着运行（状态已回到运行中，
|
|
结束时间和封存还没清掉），只能靠状态把它挡在外面。"""
|
|
done = status in ("succeeded", "failed", "cancelled") or resumed
|
|
async with SessionLocal() as session:
|
|
run = Run(workflow_id=workflow_id, workflow_name="区域周报", status=status, run_class=run_class,
|
|
schedule_id=schedule_id, started_at=finished - timedelta(minutes=2),
|
|
finished_at=finished if done else None,
|
|
manifest_hash="ab" * 32 if sealed else None, manifest_seq=1 if sealed else None)
|
|
session.add(run)
|
|
await session.commit()
|
|
return run.id
|
|
|
|
|
|
async def test_the_default_base_prefers_the_same_schedule(client):
|
|
workflow = await new_workflow()
|
|
day = timedelta(days=1)
|
|
scheduled = await bare(workflow, T0, schedule_id="sched-weekly")
|
|
manual = await bare(workflow, T0 + day) # 更近，但不是同一个定时任务
|
|
current = await bare(workflow, T0 + 2 * day, schedule_id="sched-weekly")
|
|
body = await get(client, f"/api/runs/{current}/compare/base")
|
|
assert (body["base_run_id"], body["reason"]) == (scheduled, "same_schedule")
|
|
assert body["base"]["id"] == scheduled and body["base"]["schedule_id"] == "sched-weekly"
|
|
# 不是定时任务发起的：同一工作流、同一运行类别里最近的一次（定时的也算）
|
|
later = await bare(workflow, T0 + 3 * day)
|
|
body = await get(client, f"/api/runs/{later}/compare/base")
|
|
assert (body["base_run_id"], body["reason"]) == (current, "same_workflow")
|
|
# 定时任务的第一次：退回同一工作流、同一运行类别
|
|
first = await bare(workflow, T0 + 4 * day, schedule_id="sched-daily")
|
|
body = await get(client, f"/api/runs/{first}/compare/base")
|
|
assert (body["base_run_id"], body["reason"]) == (later, "same_workflow")
|
|
assert manual != body["base_run_id"]
|
|
|
|
|
|
async def test_the_default_base_skips_what_cannot_be_compared(client):
|
|
workflow = await new_workflow()
|
|
day = timedelta(days=1)
|
|
good = await bare(workflow, T0)
|
|
await bare(workflow, T0 + day, sealed=False) # 没封存
|
|
await bare(workflow, T0 + 2 * day, status="running") # 还在运行
|
|
await bare(workflow, T0 + 2 * day, status="running", resumed=True) # 结束过、又被接着运行
|
|
await bare(workflow, T0 + 3 * day, run_class="exploratory") # 别的运行类别
|
|
await bare(await new_workflow("另一个周报"), T0 + 4 * day) # 别的工作流
|
|
current = await bare(workflow, T0 + 5 * day)
|
|
await bare(workflow, T0 + 6 * day) # 比它晚结束
|
|
body = await get(client, f"/api/runs/{current}/compare/base")
|
|
assert (body["base_run_id"], body["reason"]) == (good, "same_workflow")
|
|
# 失败、取消的终态运行也算（封存过）
|
|
failed = await bare(workflow, T0 + 5 * day - timedelta(hours=1), status="failed")
|
|
assert (await get(client, f"/api/runs/{current}/compare/base"))["base_run_id"] == failed
|
|
# 第一次运行、没保存的工作流：没有可比的
|
|
body = await get(client, f"/api/runs/{good}/compare/base", 404)
|
|
assert body["code"] == "compare_no_base" and "上一次运行" in body["detail"]
|
|
await get(client, f"/api/runs/{await bare(None, T0)}/compare/base", 404)
|
|
await get(client, "/api/runs/no-such-run/compare/base", 404)
|
|
|
|
|
|
# --------------------------------------------------------------------------
|
|
# 真跑：虚构的 SQLite 库，两次运行之间改了数、换了日期参数
|
|
# --------------------------------------------------------------------------
|
|
|
|
E2E_SOURCE = "cmp_e2e"
|
|
E2E_SQL = ("SELECT region, SUM(amount) AS gmv FROM orders WHERE day BETWEEN '{{ input.start }}' AND '{{ input.end }}' "
|
|
"GROUP BY region ORDER BY region")
|
|
WEEK36_INPUT = {"start": "2026-08-31", "end": "2026-09-06"}
|
|
WEEK37_INPUT = {"start": "2026-09-07", "end": "2026-09-13"}
|
|
|
|
|
|
@pytest.fixture
|
|
async def engine_up():
|
|
await run_manager.setup()
|
|
yield
|
|
await run_manager.shutdown()
|
|
|
|
|
|
@pytest.fixture
|
|
async def orders(tmp_path):
|
|
path = tmp_path / "orders.db"
|
|
db = sqlite3.connect(path)
|
|
db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, amount REAL);")
|
|
db.executemany("INSERT INTO orders (day, region, amount) VALUES (?, ?, ?)", [
|
|
("2026-09-01", "东区", 1200.5), ("2026-09-02", "南区", 800.0), ("2026-09-03", "西区", 600.0),
|
|
("2026-09-08", "东区", 900.0), ("2026-09-09", "南区", 650.0), ("2026-09-10", "西区", 640.0)])
|
|
db.commit()
|
|
db.close()
|
|
async with SessionLocal() as session:
|
|
row = DataSource(name=E2E_SOURCE, kind="sqlite", database=str(path), readonly=True, enabled=True)
|
|
session.add(row)
|
|
await session.commit()
|
|
source_id = row.id
|
|
await engines.invalidate(source_id)
|
|
yield path, source_id
|
|
await drop_source(source_id)
|
|
|
|
|
|
def e2e_graph():
|
|
return chain(
|
|
node("start", "input", "开始"),
|
|
node("fetch", "tool", "区域取数", tool=f"db_query__{E2E_SOURCE}", args={"sql": E2E_SQL}),
|
|
node("card", "metrics", "周报口径卡", caliber="周报口径", caliber_version="v1", metrics=[
|
|
{"id": "east", "name": "东区金额", "unit": "元", "decimals": 1, "format": "thousands",
|
|
"expression": "cell(nodes.fetch, 0, 'gmv')"}]),
|
|
node("out", "output", "成果", fields=[{"name": "指标", "value": "{{ nodes.card.text }}"}]),
|
|
)
|
|
|
|
|
|
async def finished_run(workflow_id: str, inputs: dict) -> str:
|
|
run = await run_manager.start(graph=e2e_graph(), input_payload=inputs, workflow_id=workflow_id,
|
|
workflow_name="区域周报")
|
|
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:
|
|
assert row.status == "succeeded", row.error
|
|
return run.id
|
|
raise AssertionError(f"运行 {run.id} 没有结束：{row.status}")
|
|
|
|
|
|
async def test_real_runs_split_data_and_parameter_changes(client, engine_up, orders):
|
|
path, source_id = orders
|
|
workflow = await new_workflow()
|
|
first = await finished_run(workflow, WEEK36_INPUT)
|
|
db = sqlite3.connect(path)
|
|
db.execute("UPDATE orders SET amount = 1500.5 WHERE day = '2026-09-01'")
|
|
db.commit()
|
|
db.close()
|
|
await engines.invalidate(source_id)
|
|
second = await finished_run(workflow, WEEK36_INPUT)
|
|
third = await finished_run(workflow, WEEK37_INPUT)
|
|
|
|
body = await get(client, f"/api/runs/{second}/compare?base={first}")
|
|
[east] = body["metrics"]
|
|
assert (east["classification"], east["base"], east["target"], east["delta"]) == ("data_only", 1200.5, 1500.5, 300.0)
|
|
[fetch] = body["queries"]
|
|
assert (fetch["node_id"], fetch["structure"], fetch["base_rows"], fetch["target_rows"]) == ("fetch", "same", 3, 3)
|
|
assert east["inputs"][0]["base_artifact"] == fetch["base_artifact"]
|
|
# 东区金额只是三个区域里的一格：各区域的变化照列，占比为空
|
|
assert [(r["key"], r["delta"]) for r in east["attribution"]["rows"]] == [
|
|
(["东区"], 300.0), (["南区"], 0), (["西区"], 0)]
|
|
|
|
body = await get(client, f"/api/runs/{third}/compare?base={second}")
|
|
[east] = body["metrics"]
|
|
assert east["classification"] == "params_only" and east["target"] == 900.0
|
|
[fetch] = body["queries"]
|
|
assert fetch["structure"] == "params_only"
|
|
assert fetch["literals"] == [{"base": "'2026-08-31'", "target": "'2026-09-07'"},
|
|
{"base": "'2026-09-06'", "target": "'2026-09-13'"}]
|
|
assert body["template"]["same_graph"] is True
|
|
assert any("start" in n for n in body["notes"]), body["notes"]
|
|
|
|
body = await get(client, f"/api/runs/{third}/compare/base")
|
|
assert (body["base_run_id"], body["reason"]) == (second, "same_workflow")
|
|
json.dumps(body, allow_nan=False)
|