mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 00:11:59 +00:00
ThearchyHelios
506 lines
26 KiB
Python
506 lines
26 KiB
Python
"""合并查询节点（merge）：画图时的校验、执行、证据链和下钻、助手说明。
|
||
|
||
端到端用两个独立的 SQLite 夹具（tests/fixtures/merge/）：门店库的销售订单、会员库的到店记录。两条查询各自在自己的
|
||
库里聚合到「日期 + 门店」，合并查询在库外按键合并，口径卡用 cell(nodes.merge, …) 算转化率和客单价，报告引用合并
|
||
结果的单元格和口径卡的指标。断言报告里的数能追到两边的原始查询，下钻不指错格子。
|
||
|
||
不调用任何模型：报告的写作者换成脚本（monkeypatch MockChatModel._decide）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import sqlite3
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
from langchain_core.messages import AIMessage
|
||
from sqlalchemy import select
|
||
|
||
from app.core import artifact_store
|
||
from app.core.config import settings
|
||
from app.data.engine import engines
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import DataSource, Run, RunEvent
|
||
from app.engine.evidence import iter_segments
|
||
from app.engine.merge_query import MERGE_SOURCE
|
||
from app.engine.runner import run_manager
|
||
from app.engine.schema import GraphSpec, NodeType, validate_graph
|
||
from app.main import app
|
||
|
||
FIXTURES = Path(__file__).parent / "fixtures" / "merge"
|
||
|
||
SALES_SQL = ("SELECT order_date AS 日期, store_id AS 门店, COUNT(*) AS 订单数, SUM(amount) AS 销售额 FROM orders "
|
||
"GROUP BY order_date, store_id ORDER BY order_date, store_id")
|
||
# 故意和门店库排成不同的顺序：合并结果的第 0 行在会员库的结果里是第 2 行
|
||
VISITS_SQL = ("SELECT visit_date AS 日期, store_code AS 门店, COUNT(*) AS 到店人数 FROM visits "
|
||
"GROUP BY visit_date, store_code ORDER BY visit_date DESC, store_code")
|
||
MERGE_SQL = ("SELECT s.日期, s.门店, s.订单数, s.销售额, v.到店人数 FROM s JOIN v ON s.日期 = v.日期 AND s.门店 = v.门店 "
|
||
"ORDER BY s.日期, s.门店")
|
||
TEXT = ("5 月 1 日湖滨店到店 [[v:Q3.r0.到店人数]] 人，转化率 [[m:conversion]]，客单价 [[m:aov]]。\n\n"
|
||
"[[table:Q3 cols=日期,门店,订单数,到店人数 rows=0-3]]")
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 夹具
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
@pytest.fixture
|
||
async def engine_up():
|
||
await run_manager.setup()
|
||
yield
|
||
await run_manager.shutdown()
|
||
|
||
|
||
@pytest.fixture
|
||
async def client():
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||
yield c
|
||
|
||
|
||
async def _source(name: str, path: Path, options: dict | None = None) -> str:
|
||
async with SessionLocal() as session:
|
||
row = (await session.execute(select(DataSource).where(DataSource.name == name))).scalar_one_or_none()
|
||
if row is None:
|
||
row = DataSource(name=name, kind="sqlite", database=str(path), readonly=True, enabled=True)
|
||
session.add(row)
|
||
row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, options or {}
|
||
await session.commit()
|
||
return row.id
|
||
|
||
|
||
@pytest.fixture
|
||
async def sources(tmp_path, engine_up):
|
||
"""把两个夹具脚本建成两个独立的 SQLite 库，各接成一个只读数据源。"""
|
||
ids = []
|
||
for name in ("stores", "members"):
|
||
path = tmp_path / f"{name}.db"
|
||
db = sqlite3.connect(path)
|
||
db.executescript((FIXTURES / f"{name}.sql").read_text("utf-8"))
|
||
db.commit()
|
||
db.close()
|
||
ids.append(await _source(f"merge_{name}", path))
|
||
for sid in ids:
|
||
await engines.invalidate(sid)
|
||
yield
|
||
for sid in ids:
|
||
await engines.invalidate(sid)
|
||
|
||
|
||
def node(nid, ntype, **config):
|
||
return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}}
|
||
|
||
|
||
def edge(a, b):
|
||
return {"source": a, "target": b}
|
||
|
||
|
||
def store_graph(*, sales_sql=SALES_SQL, visits_sql=VISITS_SQL, merge_sql=MERGE_SQL, visits_args=None, report=True,
|
||
inputs=None):
|
||
"""入口分两路查两个库，汇合到合并查询，再接口径卡、报告、成果。"""
|
||
nodes = [
|
||
node("start", "input"),
|
||
node("q_sales", "tool", tool="db_query__merge_stores", args={"sql": sales_sql}),
|
||
node("q_visits", "tool", tool="db_query__merge_members", args={"sql": visits_sql, **(visits_args or {})}),
|
||
node("merge", "merge", inputs=inputs or {"s": "q_sales", "v": "q_visits"}, sql=merge_sql),
|
||
node("card", "metrics", caliber="门店经营", caliber_version="v1", metrics=[
|
||
{"id": "conversion", "name": "转化率", "decimals": 4, "format": "percent_of_ratio",
|
||
"expression": "cell(nodes.merge, 0, '订单数') / cell(nodes.merge, 0, '到店人数')"},
|
||
{"id": "aov", "name": "客单价", "unit": "元", "decimals": 1,
|
||
"expression": "cell(nodes.merge, 0, '销售额') / cell(nodes.merge, 0, '订单数')"}]),
|
||
]
|
||
edges = [edge("start", "q_sales"), edge("start", "q_visits"), edge("q_sales", "merge"), edge("q_visits", "merge"),
|
||
edge("merge", "card")]
|
||
if report:
|
||
nodes += [node("write", "report", instructions="写门店日报", on_violation="flag"),
|
||
node("out", "output", fields=[{"name": "日报", "value": "{{ nodes.write.text }}"}])]
|
||
edges += [edge("card", "write"), edge("write", "out")]
|
||
else:
|
||
nodes.append(node("out", "output", fields=[{"name": "转化率", "value": "{{ nodes.card.metrics }}"}]))
|
||
edges.append(edge("card", "out"))
|
||
return {"nodes": nodes, "edges": edges}
|
||
|
||
|
||
def script(monkeypatch, text=TEXT):
|
||
from app.providers import mock_model
|
||
|
||
monkeypatch.setattr(mock_model.MockChatModel, "_decide", lambda self, messages: AIMessage(content=text))
|
||
|
||
|
||
async def wait(run_id: str) -> Run:
|
||
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:
|
||
return row
|
||
raise AssertionError(f"run {run_id} 没有结束：{row.status}")
|
||
|
||
|
||
async def run_graph(monkeypatch, graph=None, text=TEXT) -> Run:
|
||
script(monkeypatch, text)
|
||
return await wait((await run_manager.start(graph=graph or store_graph(), input_payload={})).id)
|
||
|
||
|
||
async def events(run_id: str, etype: str, node_id: str | None = None) -> list[dict]:
|
||
async with SessionLocal() as session:
|
||
query = select(RunEvent).where(RunEvent.run_id == run_id, RunEvent.type == etype)
|
||
if node_id:
|
||
query = query.where(RunEvent.node_id == node_id)
|
||
return [e.data for e in (await session.execute(query.order_by(RunEvent.seq))).scalars()]
|
||
|
||
|
||
def load_doc(row: Run) -> dict:
|
||
doc_id = row.output["_evidence"]["doc_artifact"]
|
||
return json.loads((settings.data_dir / "artifacts" / doc_id[:2] / f"{doc_id}.json").read_text("utf-8"))
|
||
|
||
|
||
def seg_of(doc: dict, text: str, n: int = 0) -> dict:
|
||
return [s for s in iter_segments(doc) if s["text"] == text and s.get("cite")][n]
|
||
|
||
|
||
def alias_of(doc: dict, node_id: str) -> str:
|
||
[alias] = [a for a, e in doc["catalog"].items() if e.get("kind") == "query" and e.get("node_id") == node_id]
|
||
return alias
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 画图时的校验
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def merge_spec(merge_cfg: dict, *, extra_nodes=(), extra_edges=()) -> GraphSpec:
|
||
base = store_graph(report=False)
|
||
nodes = [n if n["id"] != "merge" else node("merge", "merge", **merge_cfg) for n in base["nodes"]]
|
||
return GraphSpec.model_validate({"nodes": [*nodes, *extra_nodes], "edges": [*base["edges"], *extra_edges]})
|
||
|
||
|
||
def merge_errors(spec: GraphSpec) -> list:
|
||
return [i for i in validate_graph(spec).issues if i.node_id == "merge" and i.level == "error"]
|
||
|
||
|
||
def test_node_type_exists():
|
||
assert NodeType.MERGE == "merge"
|
||
|
||
|
||
def test_a_well_formed_merge_validates():
|
||
spec = merge_spec({"inputs": {"s": "q_sales", "v": "q_visits"}, "sql": MERGE_SQL})
|
||
assert merge_errors(spec) == []
|
||
|
||
|
||
def test_merge_of_a_merge_is_allowed():
|
||
spec = merge_spec({"inputs": {"s": "q_sales", "v": "q_visits"}, "sql": MERGE_SQL},
|
||
extra_nodes=[node("again", "merge", inputs={"m": "merge"}, sql="SELECT * FROM m")],
|
||
extra_edges=[edge("merge", "again")])
|
||
assert [i for i in validate_graph(spec).issues if i.node_id == "again" and i.level == "error"] == []
|
||
|
||
|
||
def test_inputs_are_required():
|
||
[issue] = merge_errors(merge_spec({"inputs": {}, "sql": MERGE_SQL}))
|
||
assert issue.field == "inputs" and issue.code == "merge.inputs_missing"
|
||
|
||
|
||
def test_input_pointing_to_a_missing_node():
|
||
[issue] = merge_errors(merge_spec({"inputs": {"s": "q_sales", "v": "nope"}, "sql": MERGE_SQL}))
|
||
assert issue.field == "inputs.v" and "不存在" in issue.message and issue.code == "merge.input_invalid"
|
||
|
||
|
||
@pytest.mark.parametrize("extra, why", [
|
||
(node("think", "llm", prompt="写点什么"), "不产出查询结果"),
|
||
(node("calc", "tool", tool="calculator", args={"expression": "1+1"}), "不是数据库查询工具"),
|
||
(node("asker", "agent", prompt="查一下", tools=["db_query__merge_stores"]), "Agent"),
|
||
])
|
||
def test_input_must_produce_a_query_result(extra, why):
|
||
spec = merge_spec({"inputs": {"s": "q_sales", "v": extra["id"]}, "sql": MERGE_SQL},
|
||
extra_nodes=[extra], extra_edges=[edge("start", extra["id"]), edge(extra["id"], "merge")])
|
||
[issue] = merge_errors(spec)
|
||
assert issue.field == "inputs.v" and why in issue.message and issue.code == "merge.input_invalid"
|
||
|
||
|
||
def test_agent_inputs_explain_how_to_fix():
|
||
extra = node("asker", "agent", prompt="查一下", tools=["db_query__merge_stores"])
|
||
spec = merge_spec({"inputs": {"s": "q_sales", "v": "asker"}, "sql": MERGE_SQL},
|
||
extra_nodes=[extra], extra_edges=[edge("start", "asker"), edge("asker", "merge")])
|
||
[issue] = merge_errors(spec)
|
||
assert "查询多次" in issue.message and "调用工具" in issue.message
|
||
|
||
|
||
def test_input_must_be_upstream():
|
||
lonely = node("q_other", "tool", tool="db_query__merge_stores", args={"sql": "SELECT 1 AS x"})
|
||
spec = merge_spec({"inputs": {"s": "q_sales", "v": "q_other"}, "sql": MERGE_SQL},
|
||
extra_nodes=[lonely], extra_edges=[edge("start", "q_other")])
|
||
[issue] = merge_errors(spec)
|
||
assert issue.field == "inputs.v" and "上游" in issue.message
|
||
|
||
|
||
@pytest.mark.parametrize("alias", ["1s", "a-b", "select", "sqlite_x"])
|
||
def test_alias_must_be_a_legal_table_name(alias):
|
||
[issue] = merge_errors(merge_spec({"inputs": {"s": "q_sales", alias: "q_visits"}, "sql": MERGE_SQL}))
|
||
assert issue.field == f"inputs.{alias}" and issue.code == "merge.alias_invalid" and alias in issue.message
|
||
|
||
|
||
@pytest.mark.parametrize("sql, why", [
|
||
("", "还没有填写"),
|
||
("DELETE FROM s", "SELECT 或 WITH"),
|
||
("PRAGMA table_info(s)", "SELECT 或 WITH"),
|
||
("SELECT * FROM s; SELECT * FROM v", "一条"),
|
||
])
|
||
def test_sql_must_be_one_select_or_with(sql, why):
|
||
[issue] = merge_errors(merge_spec({"inputs": {"s": "q_sales", "v": "q_visits"}, "sql": sql}))
|
||
assert issue.field == "sql" and why in issue.message and issue.code == "merge.sql_invalid"
|
||
|
||
|
||
def test_merge_inputs_count_as_references_to_the_upstream_nodes():
|
||
from app.engine.variables import analyze
|
||
|
||
report = analyze(merge_spec({"inputs": {"s": "q_sales", "v": "q_visits"}, "sql": MERGE_SQL}))
|
||
refs = {v.path: v.refs for v in report.variables}
|
||
assert any(r.node_id == "merge" for r in refs["nodes.q_sales"])
|
||
|
||
|
||
def test_publish_checks_treat_the_merge_as_traceable_evidence():
|
||
"""受管级别的发布检查：合并结果喂给口径卡、报告，和数据源查询一样算可追溯的证据，不在合并节点上报任何问题。"""
|
||
from app.engine.governance import publish_issues
|
||
|
||
graph = store_graph()
|
||
for n in graph["nodes"]:
|
||
if n["id"] == "write":
|
||
n["data"]["config"].update(numbers="strict", on_violation="fail", claims="require_citation",
|
||
analysis_claims="strict")
|
||
if n["id"] == "out":
|
||
n["data"]["config"]["contract"] = {"report_from": "write", "metrics_from": ["card"],
|
||
"required": ["conversion"], "strict": True}
|
||
issues = publish_issues(GraphSpec.model_validate(graph), level="governed")
|
||
assert [i for i in issues if i.node_id == "merge"] == []
|
||
assert [i for i in issues if i.level == "error"] == [], [i.message for i in issues if i.level == "error"]
|
||
|
||
|
||
def test_labels_and_governance_know_the_type():
|
||
from app.engine.governance import _traceable
|
||
from app.engine.labels import field_label, type_label
|
||
from app.engine.upgrade import _queries
|
||
|
||
spec = merge_spec({"inputs": {"s": "q_sales", "v": "q_visits"}, "sql": MERGE_SQL})
|
||
merge = spec.node_map()["merge"]
|
||
assert type_label("merge") == "合并查询"
|
||
assert field_label("sql", "merge") == "合并 SQL" and field_label("inputs", "merge") == "输入"
|
||
assert _traceable(merge) and _queries(merge)
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 执行与证据链
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_end_to_end_two_databases(monkeypatch, sources, client):
|
||
row = await run_graph(monkeypatch)
|
||
assert row.status == "succeeded", row.error
|
||
|
||
[finished] = await events(row.id, "node.finished", "merge")
|
||
output = artifact_store.load(finished["artifact"])
|
||
assert output["columns"] == ["日期", "门店", "订单数", "销售额", "到店人数"]
|
||
assert output["rows"] == [["2026-05-01", "S01", 3, 300, 6], ["2026-05-01", "S02", 2, 150, 4],
|
||
["2026-05-02", "S01", 2, 250, 5], ["2026-05-02", "S02", 1, 75, 4]]
|
||
assert output["source"] == MERGE_SOURCE and output["warnings"] == []
|
||
|
||
# 台账：一条和数据源查询同形的 query 条目，报告据此编号
|
||
[entry] = finished["evidence"]
|
||
assert entry["kind"] == "query" and entry["source"] == MERGE_SOURCE and entry["artifact"] == output["artifact"]
|
||
assert entry["rows"] == 4 and entry["truncated"] is False
|
||
snapshot = artifact_store.load(entry["artifact"])
|
||
assert [i["alias"] for i in snapshot["inputs"]] == ["s", "v"]
|
||
assert snapshot["sources"] == ["merge_stores", "merge_members"]
|
||
assert snapshot["sql"] == MERGE_SQL
|
||
|
||
# 运行事件：合并的输入、SQL、行数，运行详情据此展示
|
||
[done] = await events(row.id, "merge.end", "merge")
|
||
assert done["sql"] == MERGE_SQL and done["rows"] == 4 and done["warnings"] == []
|
||
assert [(i["alias"], i["node_id"], i["rows"], i["source"]) for i in done["inputs"]] == [
|
||
("s", "q_sales", 4, "merge_stores"), ("v", "q_visits", 4, "merge_members")]
|
||
assert done["query_artifact"] == entry["artifact"]
|
||
|
||
doc = load_doc(row)
|
||
q3 = doc["catalog"]["Q3"]
|
||
assert q3["node_id"] == "merge" and q3["source"] == MERGE_SOURCE and q3["artifact"] == entry["artifact"]
|
||
assert {alias_of(doc, "q_sales"), alias_of(doc, "q_visits")} == {"Q1", "Q2"}
|
||
metric = doc["catalog"]["m:conversion"]
|
||
assert metric["value"] == 0.5 and metric["rendered"] == "50.00%"
|
||
# 口径卡的两个输入都按 cell() 落在合并结果的那一格上，并按快照核对过
|
||
card = artifact_store.load(metric["artifact"])
|
||
conversion = next(m for m in card["metrics"] if m["id"] == "conversion")
|
||
assert [(i["via"], i["status"], i["artifact"], i["locator"]) for i in conversion["inputs"]] == [
|
||
("tool_cell", "verified", entry["artifact"], {"row": 0, "column": "订单数"}),
|
||
("tool_cell", "verified", entry["artifact"], {"row": 0, "column": "到店人数"})]
|
||
text = row.output["日报"]
|
||
assert "到店 6 人" in text and "50.00%" in text and "100.0元" in text
|
||
|
||
|
||
async def test_cell_segment_traces_to_both_original_queries(monkeypatch, sources, client):
|
||
row = await run_graph(monkeypatch)
|
||
assert row.status == "succeeded", row.error
|
||
doc = load_doc(row)
|
||
qv = alias_of(doc, "q_visits")
|
||
qs = alias_of(doc, "q_sales")
|
||
seg = seg_of(doc, "6")
|
||
body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{seg['id']}")).json()
|
||
merge_step, visits_step = body["chain"]
|
||
|
||
assert merge_step["step"] == "query" and merge_step["alias"] == "Q3" and merge_step["source"] == MERGE_SOURCE
|
||
assert merge_step["hash_ok"] is True and merge_step["sealed"] is True
|
||
assert "mask_note" not in merge_step # 合并查询不是数据源，不能说它「已不存在」
|
||
merged = merge_step["merge"]
|
||
assert merged["sql"] == MERGE_SQL and merged["warnings"] == []
|
||
assert [(i["alias"], i["node_id"], i["query"], i["rows"], i["source"], i["sealed"]) for i in merged["inputs"]] == [
|
||
("s", "q_sales", qs, 4, "merge_stores", True), ("v", "q_visits", qv, 4, "merge_members", True)]
|
||
assert merged["traced"] == [{"cell": [0, "到店人数"], "input": "v", "query": qv, "row": 2, "column": "到店人数"}]
|
||
|
||
# 接着给出会员库那次查询，高亮的正是那一格，值相同
|
||
assert visits_step["step"] == "query" and visits_step["alias"] == qv and visits_step["merged_into"] == "Q3"
|
||
assert visits_step["highlight"]["cells"] == [[2, "到店人数"]]
|
||
at = visits_step["row_index"].index(2)
|
||
assert visits_step["rows"][at][visits_step["columns"].index("到店人数")] == 6
|
||
|
||
|
||
async def test_metric_segment_traces_through_the_merge(monkeypatch, sources, client):
|
||
row = await run_graph(monkeypatch)
|
||
doc = load_doc(row)
|
||
qs, qv = alias_of(doc, "q_sales"), alias_of(doc, "q_visits")
|
||
body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{seg_of(doc, '50.00%')['id']}")).json()
|
||
metric, *inputs = body["chain"][:3]
|
||
assert metric["step"] == "metric" and metric["value"] == 0.5
|
||
assert [(i["via"], i["status"], i["query"], i["cell"]) for i in inputs] == [
|
||
("tool_cell", "verified", "Q3", "Q3.r0.订单数"), ("tool_cell", "verified", "Q3", "Q3.r0.到店人数")]
|
||
steps = [s for s in body["chain"] if s["step"] == "query"]
|
||
assert [s["alias"] for s in steps] == ["Q3", qs, qv]
|
||
merge_step, sales_step, visits_step = steps
|
||
assert sorted(t["input"] for t in merge_step["merge"]["traced"]) == ["s", "v"]
|
||
assert sales_step["highlight"]["cells"] == [[0, "订单数"]] and sales_step["merged_into"] == "Q3"
|
||
assert visits_step["highlight"]["cells"] == [[2, "到店人数"]]
|
||
|
||
|
||
async def test_provenance_of_a_merged_cell_follows_the_trace(monkeypatch, sources, client):
|
||
"""手工源上的格：追到会员库那次查询，结论和直接引用那一格相同（not_upload），并记下经过的合并。"""
|
||
row = await run_graph(monkeypatch)
|
||
doc = load_doc(row)
|
||
seg = seg_of(doc, "6")
|
||
out = (await client.get(f"/api/runs/{row.id}/evidence/segments/{seg['id']}/provenance")).json()
|
||
assert (out["status"], out["reason"]["code"]) == ("none", "not_upload")
|
||
assert out["cell"] == {"alias": "Q3", "row": 0, "column": "到店人数", "artifact": doc["catalog"]["Q3"]["artifact"]}
|
||
assert out["merge"] == [{"alias": "Q3", "node_id": "merge", "input": "v", "query": alias_of(doc, "q_visits"),
|
||
"row": 2, "column": "到店人数"}]
|
||
|
||
|
||
async def test_a_computed_merge_column_has_no_cell_lineage(monkeypatch, sources, client):
|
||
sql = ("SELECT s.日期, s.门店, s.订单数 + v.到店人数 AS 合计 FROM s JOIN v ON s.日期 = v.日期 AND s.门店 = v.门店 "
|
||
"ORDER BY s.日期, s.门店")
|
||
graph = store_graph(merge_sql=sql)
|
||
graph["nodes"] = [n for n in graph["nodes"] if n["id"] != "card"]
|
||
graph["edges"] = [e for e in graph["edges"] if "card" not in (e["source"], e["target"])] + [edge("merge", "write")]
|
||
row = await run_graph(monkeypatch, graph, text="合计 [[v:Q3.r0.合计]]。")
|
||
assert row.status == "succeeded", row.error
|
||
doc = load_doc(row)
|
||
seg = seg_of(doc, "9")
|
||
body = (await client.get(f"/api/runs/{row.id}/evidence/segments/{seg['id']}")).json()
|
||
[merge_step] = body["chain"]
|
||
[traced] = merge_step["merge"]["traced"]
|
||
assert traced["input"] is None and traced["query"] is None and "逐格来历" in traced["note"]
|
||
out = (await client.get(f"/api/runs/{row.id}/evidence/segments/{seg['id']}/provenance")).json()
|
||
assert (out["status"], out["reason"]["code"], out["version"]) == ("none", "merge_no_lineage", None)
|
||
assert out["merge"] == [{"alias": "Q3", "node_id": "merge", "input": None, "query": None, "row": None,
|
||
"column": None}]
|
||
|
||
|
||
async def test_evidence_graph_links_the_merge_to_its_inputs(monkeypatch, sources, client):
|
||
row = await run_graph(monkeypatch)
|
||
doc = load_doc(row)
|
||
body = (await client.get(f"/api/runs/{row.id}/evidence")).json()
|
||
merged = sorted((e["to"], e["input"]) for e in body["edges"] if e["rel"] == "merged_from" and e["from"] == "Q3")
|
||
assert merged == sorted([(alias_of(doc, "q_sales"), "s"), (alias_of(doc, "q_visits"), "v")])
|
||
|
||
|
||
def test_judge_excerpt_names_the_inputs_and_masks_their_sources():
|
||
"""裁判看合并结果时，知道合并了哪几个输入；输入数据源现在设的遮罩照样生效。"""
|
||
from app.engine import judge
|
||
from app.engine.merge_query import MergeInput, execute
|
||
|
||
sales = {"columns": ["门店", "订单数"], "rows": [["S01", 3]], "truncated": False, "source": "merge_stores"}
|
||
visits = {"columns": ["门店", "手机号"], "rows": [["S01", "13800000000"]], "truncated": False,
|
||
"source": "merge_members"}
|
||
snap = execute([MergeInput("s", "q_sales", "a" * 64, sales), MergeInput("v", "q_visits", "b" * 64, visits)],
|
||
"SELECT s.门店, s.订单数, v.手机号 FROM s JOIN v ON s.门店 = v.门店").snapshot()
|
||
entry = {"kind": "query", "artifact": "m" * 64, "source": MERGE_SOURCE}
|
||
text = judge._query_excerpt("Q3", entry, {0}, ["订单数"], False, lambda _a: snap,
|
||
{"merge_members": ["手机号"]})
|
||
assert text.startswith("【Q3】合并查询结果（共 1 行）")
|
||
assert "合并的输入：s（节点 q_sales，数据源 merge_stores，1 行）、v（节点 q_visits，数据源 merge_members，1 行）" in text
|
||
assert "13800000000" not in text and "遮罩" in text
|
||
|
||
|
||
async def test_a_truncated_input_fails_the_node(monkeypatch, sources):
|
||
row = await run_graph(monkeypatch, store_graph(visits_args={"limit": 2}, report=False))
|
||
assert row.status == "failed"
|
||
[failed] = await events(row.id, "node.failed", "merge")
|
||
assert "「v」" in failed["error"] and "截断" in failed["error"] and "聚合" in failed["error"]
|
||
|
||
|
||
async def test_warnings_reach_the_output_and_the_events(monkeypatch, sources):
|
||
"""门店编号一边是文本 '01'、一边是数 1；只按门店合并、没按日期：两道警告都要看得见。"""
|
||
sales_sql = ("SELECT order_date AS 日期, substr(store_id, 2) AS 门店, COUNT(*) AS 订单数, SUM(amount) AS 销售额 "
|
||
"FROM orders GROUP BY order_date, store_id ORDER BY 1, 2")
|
||
visits_sql = ("SELECT visit_date AS 日期, CAST(substr(store_code, 2) AS INTEGER) AS 门店, COUNT(*) AS 到店人数 "
|
||
"FROM visits GROUP BY visit_date, store_code ORDER BY 1, 2")
|
||
merge_sql = "SELECT s.门店, s.订单数, s.销售额, v.到店人数 FROM s JOIN v ON s.门店 = v.门店"
|
||
row = await run_graph(monkeypatch, store_graph(sales_sql=sales_sql, visits_sql=visits_sql, merge_sql=merge_sql,
|
||
report=False))
|
||
assert row.status == "succeeded", row.error
|
||
[finished] = await events(row.id, "node.finished", "merge")
|
||
output = artifact_store.load(finished["artifact"])
|
||
assert sorted(w["code"] for w in output["warnings"]) == ["key_type_mismatch", "rows_grew"]
|
||
[done] = await events(row.id, "merge.end", "merge")
|
||
assert sorted(w["code"] for w in done["warnings"]) == ["key_type_mismatch", "rows_grew"]
|
||
logs = [e for e in await events(row.id, "log", "merge") if e.get("level") == "warn"]
|
||
assert sorted(e["code"] for e in logs) == ["merge_key_type", "merge_rows_grew"]
|
||
assert all(e["message"] for e in logs)
|
||
# 结构化字段随日志带上：运行时间线读字段，不从中文里解析
|
||
key = next(e for e in logs if e["code"] == "merge_key_type")
|
||
assert key["keys"] == ["s.门店", "v.门店"] and key["detail"].startswith("s.门店 是文本") and "v.门店 是数值" in key["detail"]
|
||
grew = next(e for e in logs if e["code"] == "merge_rows_grew")
|
||
assert grew["rows"] == done["rows"] and grew["input"] in ("s", "v") and isinstance(grew["input_rows"], int)
|
||
assert grew["rows"] > grew["input_rows"] and "有重复的键" in grew["duplicates"]
|
||
|
||
|
||
async def test_rerunning_the_node_on_the_same_inputs_gives_the_same_snapshot(monkeypatch, sources):
|
||
"""重放、继续运行时节点从头再执行：同样的输入得到同一份快照、同一条台账，下游的出处不变。"""
|
||
from app.engine.context import NodeContext, RunContext
|
||
from app.engine.nodes.merge import run_merge
|
||
|
||
row = await run_graph(monkeypatch, store_graph(report=False))
|
||
assert row.status == "succeeded", row.error
|
||
outputs = {}
|
||
for nid in ("q_sales", "q_visits"):
|
||
[finished] = await events(row.id, "node.finished", nid)
|
||
outputs[nid] = artifact_store.load(finished["artifact"])
|
||
spec = GraphSpec.model_validate(store_graph(report=False))
|
||
ctx = NodeContext(node=spec.node_map()["merge"],
|
||
run=RunContext(run_id=row.id, thread_id=row.id, spec=spec, agent_limits={}))
|
||
state = {"nodes": outputs, "trail": [], "evidence": []}
|
||
first, second = await run_merge(state, ctx), await run_merge(state, ctx)
|
||
assert first == second
|
||
assert first["evidence"][0]["artifact"] == first["nodes"]["merge"]["artifact"]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 助手
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
def test_node_reference_explains_when_to_use_merge():
|
||
from app.api.copilot import NODE_REFERENCE
|
||
|
||
assert "- merge：" in NODE_REFERENCE
|
||
block = NODE_REFERENCE.split("- merge：", 1)[1].split("\n- ", 1)[0]
|
||
for phrase in ("inputs", "sql", "不同的库", "同一个库", "一条 SQL", "口径卡", "截断", "cell(nodes."):
|
||
assert phrase in block, phrase
|