mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:21:58 +00:00
ThearchyHelios
486 lines
27 KiB
Python
486 lines
27 KiB
Python
"""分析入口接真算子跑一遍（不用替身）：agent 的五个分析工具（变化拆解含下钻，波动基线、趋势、差异检验、可比性检查各一次）、
|
||
分析卡（波动基线 + 变化拆解；输入是合并查询的一次）、口径卡的可比性挂钩、审批恢复时的重放。断言工件、台账、交回模型的那一段、
|
||
节点产出，并用 analysis.recompute 按封存的工件复算、逐字段相同。
|
||
|
||
虚构零售库 shop：orders(day, region, channel, amount)，2026-03-02 起 12 个整周，每周每个地区、渠道两笔（周三、周日），
|
||
最后一周华东掉三成。模型换成脚本（MockChatModel._decide），不调用任何模型。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import sqlite3
|
||
from datetime import date, timedelta
|
||
|
||
import pytest
|
||
from langchain_core.messages import AIMessage, ToolMessage
|
||
from sqlalchemy import select
|
||
|
||
import app.analysis as analysis_ops
|
||
from app.analysis import contract
|
||
from app.core import artifact_store
|
||
from app.core.artifact_store import canonical_json, content_hash
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import Run, RunEvent
|
||
from app.engine.runner import run_manager
|
||
|
||
TOOL = "db_query__shop"
|
||
DECOMPOSE = contract.TOOL_NAMES["decompose"]
|
||
BASELINE, TREND, DIFFER, COMPARABLE = (contract.TOOL_NAMES[op] for op in ("baseline", "trend", "differ", "comparable"))
|
||
WEEK = "date(day, '-6 days', 'weekday 1')"
|
||
BY_REGION = (f"SELECT region, {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE day >= '2026-05-11' "
|
||
"GROUP BY region, week ORDER BY region, week")
|
||
TOTALS = f"SELECT {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE day >= '2026-05-11' GROUP BY week ORDER BY week"
|
||
EAST = (f"SELECT channel, {WEEK} AS week, SUM(amount) AS gmv FROM orders WHERE region = '华东' AND day >= '2026-05-11' "
|
||
"GROUP BY channel, week ORDER BY channel, week")
|
||
WEEKLY = f"SELECT {WEEK} AS week, SUM(amount) AS gmv FROM orders GROUP BY week ORDER BY week"
|
||
TWO_WEEKS = {"period": "week", "base": "2026-05-11", "target": "2026-05-18", "measure": "gmv"}
|
||
A1_ARGS = {"mode": "additive", "query": "Q1", "by": ["region"], **TWO_WEEKS, "label": "销售额",
|
||
"headline": {"base": "Q2.r0.gmv", "target": "Q2.r1.gmv"}, "hypothesis": "下降是不是集中在某个地区"}
|
||
A2_ARGS = {"mode": "additive", "query": "Q3", "by": ["channel"], **TWO_WEEKS, "parent": {"of": "A1", "key": ["华东"]},
|
||
"hypothesis": "华东是哪个渠道掉的"}
|
||
|
||
|
||
def _orders():
|
||
out, i = [], 0
|
||
base = {"华东": 400, "华南": 300, "华北": 250, "西南": 50}
|
||
for w in range(12):
|
||
monday = date(2026, 3, 2) + timedelta(weeks=w)
|
||
for region, amount in base.items():
|
||
for channel, share in (("线上", 3), ("线下", 2)):
|
||
value = amount * share // 5 + (w % 3) * 2
|
||
if w == 11 and region == "华东":
|
||
value = value * 7 // 10
|
||
for offset in (2, 6):
|
||
i += 1
|
||
out.append((i, (monday + timedelta(days=offset)).isoformat(), region, channel, value))
|
||
return out
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
async def engine_up():
|
||
await run_manager.setup()
|
||
yield
|
||
await run_manager.shutdown()
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
async def shop(tmp_path, scratch_source):
|
||
"""虚构零售库 shop（tests/conftest.py 的 scratch_source：同名的旧源连同数据目录、现查缓存先删掉，用完再删）。"""
|
||
path = tmp_path / "shop.db"
|
||
db = sqlite3.connect(path)
|
||
db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, channel TEXT, amount INTEGER);")
|
||
db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", _orders())
|
||
db.commit()
|
||
db.close()
|
||
return await scratch_source("shop", path)
|
||
|
||
|
||
def node(nid, ntype, **config):
|
||
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
|
||
|
||
|
||
def chain(*nodes):
|
||
return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]}
|
||
|
||
|
||
class Script:
|
||
"""Agent 依次发 calls 里的工具调用，再给结论；seen 记下每次带工具的调用看到的消息。"""
|
||
|
||
def __init__(self, monkeypatch, calls):
|
||
from app.providers import mock_model
|
||
|
||
self.seen: list[list] = []
|
||
|
||
def decide(model, messages):
|
||
if not model.tools:
|
||
return AIMessage(content="收尾")
|
||
self.seen.append(list(messages))
|
||
done = sum(isinstance(m, ToolMessage) for m in messages)
|
||
if done < len(calls):
|
||
name, args = calls[done]
|
||
return AIMessage(content="", tool_calls=[{"name": name, "args": args, "id": f"call_{done}"}])
|
||
return AIMessage(content="销售额下降主要来自华东。")
|
||
|
||
monkeypatch.setattr(mock_model.MockChatModel, "_decide", decide)
|
||
|
||
def replies(self):
|
||
return [m.content for m in self.seen[-1] if isinstance(m, ToolMessage)]
|
||
|
||
|
||
class Counting:
|
||
"""真算子外面包一层只数次数（重放时不该再算）。"""
|
||
|
||
def __init__(self, monkeypatch):
|
||
self.calls = 0
|
||
real = analysis_ops.run
|
||
|
||
async def run(*args, **kwargs):
|
||
self.calls += 1
|
||
return await real(*args, **kwargs)
|
||
|
||
monkeypatch.setattr(analysis_ops, "run", run)
|
||
|
||
|
||
async def wait(run_id, statuses=("succeeded", "failed")):
|
||
for _ in range(600):
|
||
await asyncio.sleep(0.05)
|
||
async with SessionLocal() as session:
|
||
row = await session.get(Run, run_id)
|
||
if row.status in statuses and (row.status == "interrupted" or row.manifest_seq is not None):
|
||
return row
|
||
raise AssertionError(f"run {run_id} 没有结束：{row.status}")
|
||
|
||
|
||
async def events(run_id, etype=None, node_id=None):
|
||
async with SessionLocal() as session:
|
||
q = select(RunEvent).where(RunEvent.run_id == run_id)
|
||
if etype:
|
||
q = q.where(RunEvent.type == etype)
|
||
if node_id:
|
||
q = q.where(RunEvent.node_id == node_id)
|
||
return [e.data for e in (await session.execute(q.order_by(RunEvent.seq))).scalars()]
|
||
|
||
|
||
def agent_graph(approval="never", tools=(TOOL, DECOMPOSE)):
|
||
return chain(node("start", "input"),
|
||
node("bot", "agent", prompt="销售额为什么下降", tools=list(tools), approval=approval),
|
||
node("out", "output", fields=[{"name": "answer", "value": "{{ nodes.bot.text }}"}]))
|
||
|
||
|
||
def recomputes(stored):
|
||
"""按封存的工件复算（查询快照、上一层结论都按 sha256 取回），和封存的逐字段相同。"""
|
||
again = analysis_ops.recompute(stored, artifact_store.load, load_result=artifact_store.load)
|
||
assert contract.same(stored, again), {k: (stored.get(k), again.get(k)) for k in stored if stored.get(k) != again.get(k)}
|
||
|
||
|
||
AGENT_CALLS = [(TOOL, {"sql": BY_REGION}), (TOOL, {"sql": TOTALS}), (TOOL, {"sql": EAST}),
|
||
(DECOMPOSE, A1_ARGS), (DECOMPOSE, A2_ARGS)]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# agent 分析工具：拆解、下钻
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_an_agent_decomposes_and_drills_down_with_the_real_operator(monkeypatch):
|
||
script = Script(monkeypatch, AGENT_CALLS)
|
||
run = await run_manager.start(graph=agent_graph(), input_payload={})
|
||
row = await wait(run.id)
|
||
assert row.status == "succeeded", row.error
|
||
|
||
ends = await events(row.id, "tool.end", "bot")
|
||
q1, q2, q3 = (ends[i]["query_artifact"] for i in range(3))
|
||
a1_artifact, a2_artifact = ends[3]["analysis_artifact"], ends[4]["analysis_artifact"]
|
||
a1, a2 = artifact_store.load(a1_artifact), artifact_store.load(a2_artifact)
|
||
|
||
# 结论：华东是主要来源，下钻到渠道是线上；总数闭合（headline 是 Q2 的两格）
|
||
assert a1["status"] == "ok" and a1["verdict"] == {"code": "driver", "subjects": [["华东"]], "direction": "down"}
|
||
assert "主要来自华东" in a1["phrase"] and a1["hypothesis"] == "下降是不是集中在某个地区"
|
||
assert {i["ref"] for i in a1["inputs"]} >= {"Q1", "Q2.r0.gmv", "Q2.r1.gmv"}
|
||
assert a2["verdict"]["code"] == "driver" and a2["verdict"]["subjects"] == [["线上"]]
|
||
# 下钻：算子按上一层结论的内容哈希认 parent，就是存下的那份工件
|
||
[parent] = [i for i in a2["inputs"] if i.get("role") == "parent"]
|
||
assert parent == {"ref": "A1", "artifact": a1_artifact, "role": "parent"}
|
||
assert content_hash(canonical_json(a1)) == a1_artifact
|
||
|
||
# 台账：两条 kind analysis，带 call_id，输入是那几份查询快照（下钻的另有上一层结论）
|
||
[finished] = await events(row.id, "node.finished", "bot")
|
||
analyses = [e for e in finished["evidence"] if e["kind"] == "analysis"]
|
||
assert [(e["call_id"], e["artifact"]) for e in analyses] == [("call_3", a1_artifact), ("call_4", a2_artifact)]
|
||
assert set(analyses[0]["inputs"]) == {q1, q2} and set(analyses[1]["inputs"]) == {q3, a1_artifact}
|
||
assert analyses[0] == contract.ledger_entry(a1, node_id="bot", exec_no=1, artifact=a1_artifact, call_id="call_3")
|
||
|
||
# 交回模型的：编号、判定语、关键数
|
||
replies = script.replies()
|
||
assert [r.split("\n", 1)[0] for r in replies] == ["【证据 Q1】", "【证据 Q2】", "【证据 Q3】", "【分析 A1】", "【分析 A2】"]
|
||
body = json.loads(replies[3].split("\n", 1)[1])
|
||
assert body["phrase"] == a1["phrase"] and body["verdict"] == "driver" and body["values"] == a1["values"]
|
||
assert body["table"]["rows"][0][0] == "华东"
|
||
|
||
# 按封存的工件复算，逐字段相同
|
||
recomputes(a1)
|
||
recomputes(a2)
|
||
|
||
|
||
async def agent_analysis(monkeypatch, tool, calls):
|
||
"""Agent 依次查库、最后调一次分析工具（真算子）：交回 (那份结论, 交回模型的那一段的 JSON)。顺带断言台账记了这一条（带
|
||
call_id）、交回模型的编号是「【分析 A1】」、判定语和结论一字不差，并按封存的工件复算、逐字段相同。"""
|
||
script = Script(monkeypatch, calls)
|
||
run = await run_manager.start(graph=agent_graph(tools=(TOOL, tool)), input_payload={})
|
||
row = await wait(run.id)
|
||
assert row.status == "succeeded", row.error
|
||
ends = await events(row.id, "tool.end", "bot")
|
||
artifact = ends[-1]["analysis_artifact"]
|
||
stored = artifact_store.load(artifact)
|
||
[finished] = await events(row.id, "node.finished", "bot")
|
||
[entry] = [e for e in finished["evidence"] if e["kind"] == "analysis"]
|
||
assert entry == contract.ledger_entry(stored, node_id="bot", exec_no=1, artifact=artifact,
|
||
call_id=f"call_{len(calls) - 1}")
|
||
assert set(entry["inputs"]) == {e["query_artifact"] for e in ends[:-1]}
|
||
head, body = script.replies()[-1].split("\n", 1)
|
||
assert head == "【分析 A1】" and json.loads(body)["phrase"] == stored["phrase"]
|
||
recomputes(stored)
|
||
return stored, json.loads(body)
|
||
|
||
|
||
def _live(stored):
|
||
"""结论里现查的表范围：按这次运行的数据源探 orders.day。"""
|
||
return [(p["table"], p["column"], p["basis"], p["max"]) for p in stored["probes"]]
|
||
|
||
|
||
async def test_an_agent_puts_the_week_into_its_history_with_the_real_operator(monkeypatch):
|
||
stored, body = await agent_analysis(monkeypatch, BASELINE, [
|
||
(TOOL, {"sql": WEEKLY}),
|
||
(BASELINE, {"query": "Q1", "period": "week", "measure": "gmv", "label": "销售额", "hypothesis": "这周是不是异常"})])
|
||
assert stored["op"] == "baseline" and stored["status"] == "ok" and stored["hypothesis"] == "这周是不是异常"
|
||
assert stored["verdict"] == {"code": "far_outside_band", "direction": "down", "subjects": []}
|
||
assert stored["phrase"].startswith("销售额本期变化 -10.6%，明显超出过去 10 周的常规波动范围")
|
||
assert "record_low" in {c["code"] for c in stored["caveats"]} and body["verdict"] == "far_outside_band"
|
||
assert _live(stored) == [("orders", "day", "live", "2026-05-24")] # 本期完不完整按现查的表范围判
|
||
|
||
|
||
async def test_an_agent_asks_for_a_trend_with_the_real_operator(monkeypatch):
|
||
stored, body = await agent_analysis(monkeypatch, TREND, [
|
||
(TOOL, {"sql": WEEKLY}), (TREND, {"query": "Q1", "period": "week", "measure": "gmv", "label": "销售额"})])
|
||
assert stored["op"] == "trend" and stored["status"] == "ok"
|
||
assert stored["verdict"] == {"code": "no_trend", "direction": None, "subjects": []}
|
||
assert stored["phrase"] == "过去 12 周销售额没有明显的上升或下降趋势" and stored["values"]["n"] == "12"
|
||
assert _live(stored) == [("orders", "day", "live", "2026-05-24")] and body["values"] == stored["values"]
|
||
|
||
|
||
async def test_an_agent_tests_a_difference_with_the_real_operator(monkeypatch):
|
||
rows = "SELECT region, amount FROM orders WHERE region IN ('华东', '华南') ORDER BY region, day"
|
||
stored, body = await agent_analysis(monkeypatch, DIFFER, [
|
||
(TOOL, {"sql": rows}),
|
||
(DIFFER, {"mode": "rows", "query": "Q1", "by": ["region"], "value": "amount", "label": "单笔金额"})])
|
||
assert stored["op"] == "differ" and stored["mode"] == "rows" and stored["status"] == "ok"
|
||
assert stored["verdict"] == {"code": "differs", "direction": "up", "subjects": [["华东"], ["华南"]]}
|
||
assert stored["phrase"] == "华东的单笔金额（196.83）高于华南（152），差异站得住"
|
||
assert stored["values"]["n_a"] == stored["values"]["n_b"] == "48" and stored["values"]["test"] == "welch"
|
||
assert body["subjects"] == [["华东"], ["华南"]]
|
||
|
||
|
||
async def test_an_agent_checks_two_weeks_are_comparable_with_the_real_operator(monkeypatch):
|
||
stored, body = await agent_analysis(monkeypatch, COMPARABLE, [
|
||
(TOOL, {"sql": _week_sql("2026-05-18", "2026-05-24")}), (TOOL, {"sql": _week_sql("2026-05-11", "2026-05-17")}),
|
||
(COMPARABLE, {"base_query": "Q2", "target_query": "Q1", "measure": "gmv"})])
|
||
assert stored["op"] == "comparable" and stored["status"] == "ok" and stored["phrase"] == "两期可比"
|
||
assert stored["params"]["base_query"] == "Q2" and stored["params"]["target_query"] == "Q1"
|
||
values = stored["values"]
|
||
assert (values["base_start"], values["base_end"], values["target_start"], values["target_end"]) == (
|
||
"2026-05-11", "2026-05-17", "2026-05-18", "2026-05-24")
|
||
assert {values[k] for k in ("same_query", "window", "complete", "weekdays", "data")} == {"pass"}
|
||
assert _live(stored) == [("orders", "day", "live", "2026-05-24")] and body["verdict"] == "comparable"
|
||
|
||
|
||
async def test_replaying_after_approvals_does_not_recompute(monkeypatch):
|
||
counted = Counting(monkeypatch)
|
||
script = Script(monkeypatch, AGENT_CALLS)
|
||
run = await run_manager.start(graph=agent_graph(approval="always"), input_payload={})
|
||
for _ in AGENT_CALLS:
|
||
row = await wait(run.id, ("interrupted", "failed", "succeeded"))
|
||
assert row.status == "interrupted", row.error
|
||
await run_manager.resume(run.id, {"approved": True})
|
||
row = await wait(run.id, ("succeeded", "failed", "interrupted"))
|
||
assert row.status == "succeeded", row.error
|
||
|
||
assert counted.calls == 2, "重放时分析又算了一次"
|
||
ends = await events(row.id, "tool.end", "bot")
|
||
assert len(ends) == len(AGENT_CALLS)
|
||
a1_artifact, a2_artifact = ends[3]["analysis_artifact"], ends[4]["analysis_artifact"]
|
||
assert [r.split("\n", 1)[0] for r in script.replies()][-2:] == ["【分析 A1】", "【分析 A2】"]
|
||
[parent] = [i for i in artifact_store.load(a2_artifact)["inputs"] if i.get("role") == "parent"]
|
||
assert parent["artifact"] == a1_artifact
|
||
[finished] = await events(row.id, "node.finished", "bot")
|
||
assert [e["artifact"] for e in finished["evidence"] if e["kind"] == "analysis"] == [a1_artifact, a2_artifact]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 分析卡：波动基线 + 变化拆解
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_an_analysis_card_runs_the_real_operators():
|
||
items = [
|
||
{"id": "wow", "op": "baseline", "from": "weekly", "params": {"period": "week", "measure": "gmv", "label": "销售额"}},
|
||
{"id": "where", "op": "decompose", "mode": "additive", "from": "by_region",
|
||
"params": {"by": ["region"], **TWO_WEEKS, "label": "销售额",
|
||
"headline": {"base": "weekly.r10.gmv", "target": "weekly.r11.gmv"}}},
|
||
]
|
||
graph = chain(node("start", "input"), node("weekly", "tool", tool=TOOL, args={"sql": WEEKLY}),
|
||
node("by_region", "tool", tool=TOOL, args={"sql": BY_REGION}),
|
||
node("card", "analysis", items=items),
|
||
node("out", "output", fields=[{"name": "判定", "value": "{{ nodes.card.results.wow.phrase }}"}]))
|
||
run = await run_manager.start(graph=graph, input_payload={})
|
||
row = await wait(run.id)
|
||
assert row.status == "succeeded", row.error
|
||
|
||
[finished] = await events(row.id, "node.finished", "card")
|
||
results = artifact_store.load(finished["artifact"])["results"]
|
||
wow, where = results["wow"], results["where"]
|
||
assert wow["status"] == "ok" and wow["verdict"] in ("outside_band", "far_outside_band") and wow["direction"] == "down"
|
||
assert where["status"] == "ok" and where["verdict"] == "driver" and "主要来自华东" in where["phrase"]
|
||
assert row.output["判定"] == wow["phrase"]
|
||
|
||
baseline = artifact_store.load(wow["artifact"])
|
||
# 本期完不完整按现查的表范围判（analysis.make_prober 按这次运行的数据源探 orders.day）：探到了，不再说核对不了
|
||
[probe] = baseline["probes"]
|
||
assert (probe["source"], probe["table"], probe["column"], probe["basis"]) == ("shop", "orders", "day", "live")
|
||
assert probe["max"] == "2026-05-24"
|
||
assert not [c for c in baseline["caveats"] if c["code"] == "comparability"]
|
||
decomposed = artifact_store.load(where["artifact"])
|
||
assert decomposed["verdict"]["subjects"] == [["华东"]]
|
||
entries = finished["evidence"]
|
||
assert [(e["kind"], e["item"], e["artifact"]) for e in entries] == [
|
||
("analysis", "wow", wow["artifact"]), ("analysis", "where", where["artifact"])]
|
||
recomputes(baseline)
|
||
recomputes(decomposed)
|
||
|
||
|
||
async def test_an_analysis_card_reads_a_merged_query():
|
||
"""分析卡的输入是合并查询（merge 节点）：按合并查询节点的做法取它的快照，输入写明 sqlite 方言（合并查询在 SQLite 里
|
||
执行），真算子照常拆、复算逐字段相同。"""
|
||
where = {"id": "where", "op": "decompose", "mode": "additive", "from": "merged",
|
||
"params": {"by": ["region"], **TWO_WEEKS, "label": "销售额"}}
|
||
graph = chain(node("start", "input"), node("by_region", "tool", tool=TOOL, args={"sql": BY_REGION}),
|
||
node("merged", "merge", inputs={"a": "by_region"}, sql="SELECT region, week, gmv FROM a"),
|
||
node("card", "analysis", items=[where]),
|
||
node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.results.where.phrase }}"}]))
|
||
run = await run_manager.start(graph=graph, input_payload={})
|
||
row = await wait(run.id)
|
||
assert row.status == "succeeded", row.error
|
||
[finished] = await events(row.id, "node.finished", "card")
|
||
results = artifact_store.load(finished["artifact"])["results"]
|
||
stored = artifact_store.load(results["where"]["artifact"])
|
||
assert stored["verdict"]["code"] == "driver" and stored["verdict"]["subjects"] == [["华东"]]
|
||
assert row.output["r"] == stored["phrase"] and "主要来自华东" in stored["phrase"]
|
||
[given] = stored["inputs"]
|
||
[merged] = await events(row.id, "node.finished", "merged")
|
||
assert (given["ref"], given["role"], given["dialect"]) == ("merged", "data", "sqlite")
|
||
assert given["artifact"] == merged["evidence"][0]["artifact"]
|
||
assert [(e["kind"], e["item"]) for e in finished["evidence"]] == [("analysis", "where")]
|
||
recomputes(stored)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 口径卡的可比性挂钩：真 comparable
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def _week_sql(start: str, end: str) -> str:
|
||
return f"SELECT SUM(amount) AS gmv FROM orders WHERE day BETWEEN '{start}' AND '{end}'"
|
||
|
||
|
||
@pytest.mark.parametrize("this_week, verdict, has_reasons", [
|
||
(("2026-05-18", "2026-05-24"), "comparable", False),
|
||
(("2026-05-18", "2026-05-23"), "conditional", True), # 本期 6 天对上期 7 天：窗口长度、周末天数都不一样
|
||
])
|
||
async def test_a_caliber_card_checks_two_periods_with_the_real_operator(this_week, verdict, has_reasons):
|
||
metrics = [{"id": "wow", "name": "销售额环比", "decimals": 4,
|
||
"expression": "(cell(nodes.this_week, 0, 'gmv') - cell(nodes.last_week, 0, 'gmv'))"
|
||
" / cell(nodes.last_week, 0, 'gmv')"}]
|
||
graph = chain(node("start", "input"), node("this_week", "tool", tool=TOOL, args={"sql": _week_sql(*this_week)}),
|
||
node("last_week", "tool", tool=TOOL, args={"sql": _week_sql("2026-05-11", "2026-05-17")}),
|
||
node("card", "metrics", caliber="周报口径", metrics=metrics),
|
||
node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}]))
|
||
run = await run_manager.start(graph=graph, input_payload={})
|
||
row = await wait(run.id)
|
||
assert row.status == "succeeded", row.error
|
||
[finished] = await events(row.id, "node.finished", "card")
|
||
[metric] = artifact_store.load(finished["artifact"])["metrics"]
|
||
found = metric["comparability"]
|
||
stored = artifact_store.load(found["artifact"])
|
||
assert stored["op"] == "comparable" and stored["params"]["base_query"] == "last_week"
|
||
assert stored["params"]["target_query"] == "this_week"
|
||
assert found == {"artifact": found["artifact"], "verdict": verdict, "phrase": stored["phrase"],
|
||
"reasons": analysis_ops.comparable_reasons(stored)}
|
||
assert bool(found["reasons"]) is has_reasons
|
||
# 整期按现查的表范围判
|
||
assert any(p.get("table") == "orders" and p.get("basis") == "live" for p in stored["probes"])
|
||
recomputes(stored)
|
||
assert not [e for e in await events(row.id, "log", "card") if e.get("code") == "metric_comparability_skipped"]
|
||
|
||
|
||
def _region_sql(region: str, start: str, end: str) -> str:
|
||
return f"SELECT SUM(amount) AS gmv FROM orders WHERE region = '{region}' AND day BETWEEN '{start}' AND '{end}'"
|
||
|
||
|
||
@pytest.mark.parametrize("south_week, verdict", [
|
||
(("2026-05-18", "2026-05-24"), None), # 同一周、只差地区：是两段不是两期，不查
|
||
(("2026-05-11", "2026-05-17"), "incomparable"), # 日期和地区都不同：两边除了日期，筛选的还不一样
|
||
])
|
||
async def test_a_ratio_between_two_regions_is_not_a_period_comparison(south_week, verdict):
|
||
"""华东对华南的倍数：两份同形 SQL 只差地区字面量，不做可比性检查，带契约的出口照常正式出具（以前判成两期不可比、降档）。
|
||
日期和地区都不同的照查，真算子判不可比，出口记缺口降档。"""
|
||
metrics = [{"id": "times", "name": "华东对华南的倍数", "decimals": 4,
|
||
"expression": "cell(nodes.east, 0, 'gmv') / cell(nodes.south, 0, 'gmv')"}]
|
||
graph = chain(node("start", "input"),
|
||
node("east", "tool", tool=TOOL, args={"sql": _region_sql("华东", "2026-05-18", "2026-05-24")}),
|
||
node("south", "tool", tool=TOOL, args={"sql": _region_sql("华南", *south_week)}),
|
||
node("card", "metrics", caliber="周报口径", metrics=metrics),
|
||
node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}],
|
||
contract={"metrics_from": ["card"], "required": ["times"]}))
|
||
run = await run_manager.start(graph=graph, input_payload={})
|
||
row = await wait(run.id)
|
||
assert row.status == "succeeded", row.error
|
||
[finished] = await events(row.id, "node.finished", "card")
|
||
[metric] = artifact_store.load(finished["artifact"])["metrics"]
|
||
issuance = row.output["_issuance"]
|
||
assert not [e for e in await events(row.id, "log", "card") if e.get("code") == "metric_comparability_skipped"]
|
||
if verdict is None:
|
||
assert "comparability" not in metric
|
||
assert issuance["tier"] == "formal" and not issuance.get("gaps") and not issuance.get("comparability")
|
||
return
|
||
found = metric["comparability"]
|
||
assert found["verdict"] == verdict and found["reasons"] == ["两边除了日期，筛选的还不一样"]
|
||
assert issuance["tier"] == "degraded"
|
||
assert issuance["gaps"] == ["指标「华东对华南的倍数」两期不可比：两边除了日期，筛选的还不一样"]
|
||
recomputes(artifact_store.load(found["artifact"]))
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 测试夹具：复用名叫 shop 的源，前后都删干净
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_a_scratch_source_starts_clean_and_forgets_its_probes(tmp_path):
|
||
"""分析相关的用例都建名叫 shop 的源（tests/conftest.py 的 scratch_source）：别的用例留下的数据目录、coverage 的现查缓存里
|
||
按同一个源 id 记下的表范围（5 分钟有效）会改掉这次的判定。sqlite_source 建的源是新的 id，同名的旧源连同数据目录、现查
|
||
缓存先删掉；forget_source 删源（数据目录随外键级联删除）、清掉现查缓存里它的表范围。"""
|
||
from app.analysis.probe import make_prober
|
||
from app.data import catalog as data_catalog
|
||
from app.data import coverage
|
||
from app.data.catalog import make_item
|
||
from app.db.models import CatalogNote, DataSource
|
||
from tests.fixtures.sources import forget_source, sqlite_source
|
||
|
||
path = tmp_path / "other.db"
|
||
db = sqlite3.connect(path)
|
||
db.executescript("CREATE TABLE orders (day TEXT, amount INTEGER); INSERT INTO orders VALUES ('2026-01-05', 1);")
|
||
db.close()
|
||
|
||
async def leave_traces(source_id):
|
||
async with SessionLocal() as session:
|
||
await data_catalog.write_entry(session, source_id, "orders", {"columns": {"amount": {
|
||
"measure": make_item("flow", "human")}}}, if_version=0, actor=None)
|
||
row = await session.get(DataSource, source_id)
|
||
probe = await make_prober({"scratch": row})("scratch", "orders", "day")
|
||
assert probe["max"] == "2026-01-05" and probe["basis"] == "live"
|
||
assert any(k[0] == source_id for k in coverage._LIVE)
|
||
|
||
async def traces(source_id):
|
||
async with SessionLocal() as session:
|
||
notes = (await session.execute(select(CatalogNote).where(CatalogNote.source_id == source_id))).scalars().all()
|
||
row = await session.get(DataSource, source_id)
|
||
return {"row": row is not None, "catalog": len(notes), "probes": sum(k[0] == source_id for k in coverage._LIVE)}
|
||
|
||
old = await sqlite_source("scratch", path)
|
||
await leave_traces(old)
|
||
new = await sqlite_source("scratch", path)
|
||
assert new != old and await traces(old) == {"row": False, "catalog": 0, "probes": 0}
|
||
assert await traces(new) == {"row": True, "catalog": 0, "probes": 0}
|
||
await leave_traces(new)
|
||
await forget_source(new)
|
||
assert await traces(new) == {"row": False, "catalog": 0, "probes": 0}
|