mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 05:21:57 +00:00
ThearchyHelios
705 lines
37 KiB
Python
705 lines
37 KiB
Python
"""按配方导入的接口（期 2，WP-5c）：真管线的端到端，加上用假 Pipeline 测的接口形状与错误码。
|
||
|
||
真管线（default_pipeline）跑「结构仿照客流表」的合成夹具（tests/fixtures/xlsx/flow.py，假名、随机数）：
|
||
- 暂存 → 规则起草 → 回答问题 → 改表名（PUT）→ 试运行 → 提交；
|
||
- 同版式的下月文件「上传新一期」→ 回执与差异卡 → 启用；
|
||
- 同一文件再传 → same_as_import → 提交不新建记录；
|
||
- 简单导入的源切换为按配方导入。
|
||
其余（错误码、AI 起草的门槛、并发、路由不被截走）用 test_recipe_imports 里的假 Pipeline。不调用任何真实模型。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import datetime as dt
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
from httpx import ASGITransport, AsyncClient
|
||
from sqlalchemy import select
|
||
|
||
from app.core import artifact_store
|
||
from app.core.config import settings
|
||
from app.data import recipe, table_versions
|
||
from app.data.engine import engines, run_query
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import DataSource, ImportStaging, SourceSnapshot, TableImport, TableRecipe
|
||
from app.main import app
|
||
from tests.fixtures.names import letters
|
||
from tests.fixtures.xlsx.flow import FLOW_EXPECT, flow_workbook
|
||
from tests.test_recipe_imports import ( # noqa: F401 - fake 是夹具，导入即可用
|
||
FLOW_FULL, api_stage, fake, raw_bytes, recipe_source, staged,
|
||
)
|
||
|
||
XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
|
||
FLOW_RECIPE = json.loads((Path(__file__).parent / "fixtures" / "recipes" / "flow_recipe.json").read_text("utf-8"))
|
||
AUG, SEP = dt.date(2026, 8, 1), dt.date(2026, 9, 1)
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def store(monkeypatch, tmp_path):
|
||
"""每个测试一个独立的数据目录：上传目录、原件存档、工件都在里面。"""
|
||
data = tmp_path / "data"
|
||
(data / "uploads").mkdir(parents=True)
|
||
monkeypatch.setattr(settings, "data_dir", data)
|
||
yield data
|
||
|
||
|
||
@pytest.fixture
|
||
async def client():
|
||
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
||
yield c
|
||
|
||
|
||
def unique(prefix: str = "wp5c") -> str:
|
||
return f"{prefix}_{letters(10)}"
|
||
|
||
|
||
async def stage(client, name: str, raw: bytes, filename: str, **form: str):
|
||
return await client.post("/api/datasources/imports/stage", files={"file": (filename, raw, XLSX)},
|
||
data={"name": name, **form})
|
||
|
||
|
||
async def scalar(source_id: str, sql: str):
|
||
async with SessionLocal() as session:
|
||
row = await session.get(DataSource, source_id)
|
||
view = await table_versions.resolve_source(session, row)
|
||
try:
|
||
return (await run_query(view, sql)).rows[0][0]
|
||
finally:
|
||
await engines.invalidate(view.id)
|
||
|
||
|
||
async def get(model, key):
|
||
async with SessionLocal() as session:
|
||
return await session.get(model, key)
|
||
|
||
|
||
def rename_wide(rec: dict[str, Any], old: str, new: str) -> dict[str, Any]:
|
||
"""把规则起草的宽表名改成 new：表名、写入它的分段（id 一起改）、关系里引用的表名。"""
|
||
rec = json.loads(json.dumps(rec, ensure_ascii=False))
|
||
for sheet in rec["sheets"]:
|
||
for block in sheet["blocks"]:
|
||
for seg in block.get("segments", []):
|
||
if seg.get("table") == old:
|
||
seg["table"] = new
|
||
if seg.get("id") == old:
|
||
seg["id"] = new
|
||
for t in rec["tables"]:
|
||
if t["name"] == old:
|
||
t["name"] = new
|
||
for r in rec["relations"]:
|
||
if r.get("table") == old:
|
||
r["table"] = new
|
||
for side in ("a", "b"):
|
||
if isinstance(r.get(side), dict) and r[side].get("table") == old:
|
||
r[side]["table"] = new
|
||
return rec
|
||
|
||
|
||
def all_confirms(staging: dict[str, Any]) -> list[str]:
|
||
return [i["id"] for i in staging["trial"]["confirm_items"]]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 真管线：首次导入
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def first_import(client, name: str, seed: int = 1) -> dict[str, Any]:
|
||
"""D00 走完首次导入，返回提交的响应。"""
|
||
raw, fn = flow_workbook(AUG, 31, seed=seed)
|
||
resp = await stage(client, name, raw, fn)
|
||
assert resp.status_code == 201, resp.text
|
||
st = resp.json()
|
||
sid = st["id"]
|
||
# 期 3：q_mode（这份报表每期怎么更新）答「每期替换」，与参考配方的哈希一致（P3-SPEC 12.0、1.5）
|
||
answers = {q["id"]: {"value": "register" if q["id"].startswith("q_relation")
|
||
else "replace" if q["id"] == "q_mode" else "null"}
|
||
for q in st["questions"]}
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/answers", json={"answers": answers})
|
||
assert resp.status_code == 200, resp.text
|
||
wide = next(t["name"] for t in st["recipe"]["tables"] if t["name"].endswith("_按日"))
|
||
resp = await client.put(f"/api/datasources/imports/{sid}/recipe",
|
||
json={"recipe": rename_wide(resp.json()["recipe"], wide, "日客流")})
|
||
assert resp.status_code == 200, resp.text
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/trial", json={})
|
||
assert resp.status_code == 200, resp.text
|
||
st = resp.json()
|
||
assert st["trial"]["status"] == "passed", st["trial"]["problems"]
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/commit",
|
||
json={"trial_id": st["trial"]["trial_id"], "confirmations": all_confirms(st)})
|
||
assert resp.status_code == 201, resp.text
|
||
return {"staging": st, "commit": resp.json()}
|
||
|
||
|
||
async def test_real_pipeline_first_import_stage_draft_answer_trial_commit(client):
|
||
name = unique()
|
||
raw, fn = flow_workbook(AUG, 31, seed=3)
|
||
resp = await stage(client, name, raw, fn)
|
||
assert resp.status_code == 201, resp.text
|
||
st = resp.json()
|
||
assert st["kind"] == "first" and st["status"] == "drafting"
|
||
assert st["source"]["exists"] is False and st["source"]["import_mode"] is None
|
||
assert st["draft"]["complete"] is True and "failures_for_model" not in st["draft"]
|
||
assert st["ai"]["offered"] is False
|
||
assert len(st["cards"]) == 10
|
||
assert st["recipe_problems"] == [] and st["recipe_origin"] == "rules"
|
||
assert st["grids"][0]["sheet"] == "客流汇总" and st["grids"][0]["bounds"] == "B2:AG30"
|
||
assert st["sheets"][0]["nonempty"] == FLOW_EXPECT["nonempty"]
|
||
assert st["marks"], "干跑的区域标记"
|
||
qids = [q["id"] for q in st["questions"]]
|
||
assert {"q_relation:F1", "q_relation:F2"} <= set(qids)
|
||
sid = st["id"]
|
||
# 新源在提交前不存在
|
||
async with SessionLocal() as session:
|
||
assert (await session.execute(select(DataSource).where(DataSource.name == name))).first() is None
|
||
|
||
# 回答问题：先选「不登记」再改选「登记」，等于直接选「登记」
|
||
first = await client.post(f"/api/datasources/imports/{sid}/answers", json={"answers": {
|
||
"q_relation:F1": {"value": "dismiss", "reason": "合成理由：判断为巧合"}}})
|
||
assert first.status_code == 200, first.text
|
||
assert next(r for r in first.json()["recipe"]["relations"] if r["id"] == "R1")["kind"] == "dismissed"
|
||
answers = {q: {"value": "register" if q.startswith("q_relation") else "replace" if q == "q_mode" else "null"}
|
||
for q in qids}
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/answers", json={"answers": answers})
|
||
assert resp.status_code == 200, resp.text
|
||
st = resp.json()
|
||
assert st["answers"]["q_relation:F1"] == {"value": "register", "reason": None}
|
||
assert [q["id"] for q in st["questions"]] == qids, "问题集合按起点配方生成，回答不改变它"
|
||
answered = st["answers"]
|
||
|
||
# 改表名（PUT）：起点换成它；与参考配方逐字相同。期 3（第 8 节遗留项 2）：PUT 进来的配方已经体现了之前的
|
||
# 回答，回答照旧保留（与新起点一致），没有被丢弃的回答
|
||
resp = await client.put(f"/api/datasources/imports/{sid}/recipe",
|
||
json={"recipe": rename_wide(st["recipe"], "客流汇总_按日", "日客流")})
|
||
assert resp.status_code == 200, resp.text
|
||
st = resp.json()
|
||
assert st["answers"] == answered and st["answers_dropped"] == []
|
||
assert st["recipe_origin"] == "manual" and st["recipe_problems"] == []
|
||
ref, _ = recipe.validate_recipe(FLOW_RECIPE, origin="manual")
|
||
mine, _ = recipe.validate_recipe(st["recipe"], origin="manual")
|
||
assert recipe.recipe_sha256(mine) == recipe.recipe_sha256(ref)
|
||
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/trial", json={})
|
||
assert resp.status_code == 200, resp.text
|
||
st = resp.json()
|
||
trial = st["trial"]
|
||
assert st["status"] == "trialed" and trial["status"] == "passed", trial["problems"]
|
||
assert set(all_confirms(st)) == set(FLOW_EXPECT["confirms"])
|
||
assert {c["id"]: c["status"] for c in trial["checks"]} == FLOW_EXPECT["checks"]
|
||
assert trial["diff"] is None and trial["same_as_import"] is None
|
||
assert "extraction" not in trial and "null_counts" not in trial
|
||
assert trial["receipt"]["canonicalized_total"] == 0 and trial["receipt"]["db_sha256"]
|
||
assert {t["name"]: t["rows"] for t in trial["receipt"]["tables"]} == FLOW_EXPECT["rows"]
|
||
assert "已逐日核对" in trial["notes"]["日客流"]["comment"]
|
||
assert trial["notes"]["日客流"]["columns"]["全日客流"] == "单位：人次"
|
||
|
||
# 少勾一项 → 422，detail 列出缺的那一项
|
||
some = all_confirms(st)
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/commit",
|
||
json={"trial_id": trial["trial_id"], "confirmations": some[1:]})
|
||
assert resp.status_code == 422 and resp.json()["code"] == "confirm_required"
|
||
assert some[0] in resp.json()["detail"]
|
||
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/commit",
|
||
json={"trial_id": trial["trial_id"], "confirmations": some, "signed_by": "测试员"})
|
||
assert resp.status_code == 201, resp.text
|
||
out = resp.json()
|
||
assert out["unchanged"] is False and out["build_reused"] is False
|
||
src = out["source"]
|
||
assert src["name"] == name and src["origin"] == "upload" and src["import_mode"] == "recipe"
|
||
assert src["current_recipe"]["id"] == out["recipe_id"] and src["current_recipe"]["seq"] == 1
|
||
assert src["current_recipe"]["origin"] == "manual"
|
||
assert src["open_staging"] is None
|
||
assert await scalar(src["id"], 'SELECT COUNT(*) FROM "时段客流"') == FLOW_EXPECT["rows"]["时段客流"]
|
||
|
||
# 表结构里的说明；schema_cache 承诺了导入清单
|
||
resp = await client.get(f"/api/datasources/{src['id']}/schema", params={"table": "日客流"})
|
||
assert resp.status_code == 200, resp.text
|
||
schema = resp.json()
|
||
assert "已逐日核对" in json.dumps(schema, ensure_ascii=False)
|
||
snap = await get(SourceSnapshot, out["snapshot_id"])
|
||
imp = await get(TableImport, out["import_id"])
|
||
assert snap.schema_cache["import_mode"] == "recipe"
|
||
assert snap.schema_cache["import_manifests"] == [imp.manifest_artifact]
|
||
manifest = artifact_store.load(imp.manifest_artifact)
|
||
assert manifest["format"] == "agentlab-import-manifest/1"
|
||
assert manifest["db_sha256"] == snap.db_sha256 == manifest["trial_db_sha256"]
|
||
for key in ("axes", "tables", "ledger", "period", "labels", "outside_text", "lineage", "canonicalized"):
|
||
assert key in manifest["receipt"], key
|
||
assert "problems" not in manifest["receipt"]
|
||
assert manifest["recipe"]["id"] == out["recipe_id"] and manifest["signed_by"] == {"name": "测试员", "verified": False}
|
||
assert {c["id"] for c in manifest["checks"]} >= {"C1", "C2"}
|
||
assert {c["id"] for c in imp.checks} >= {"C1", "C2"}
|
||
assert imp.period_start == "2026-08-01" and imp.period_end == "2026-08-31" and imp.signed_by == "测试员"
|
||
assert imp.recipe_id == out["recipe_id"] and imp.staging_id == sid
|
||
assert {c["id"] for c in imp.confirmations} == set(some)
|
||
rec = await get(TableRecipe, out["recipe_id"])
|
||
assert rec.status == "active" and rec.seq == 1 and rec.staging_id == sid
|
||
# 暂存区结束即瘦身
|
||
resp = await client.get(f"/api/datasources/imports/{sid}")
|
||
st = resp.json()
|
||
assert st["status"] == "committed" and st["grids"] == [] and st["trial"] is None
|
||
assert st["committed_import_id"] == out["import_id"]
|
||
row = await get(ImportStaging, sid)
|
||
assert row.scan is None and row.grid_preview is None and row.trial_path is None and row.recipe_sha256
|
||
# 已结束的暂存区不能再写
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/trial", json={})
|
||
assert resp.status_code == 409 and resp.json()["code"] == "staging_closed"
|
||
# 按配方导入的源：简单上传 409、当前配方、导入记录的新字段
|
||
resp = await client.post("/api/datasources/upload", files={"file": (fn, raw, XLSX)}, data={"name": name})
|
||
assert resp.status_code == 409 and resp.json()["code"] == "recipe_source"
|
||
resp = await client.get(f"/api/datasources/{src['id']}/recipe")
|
||
assert resp.status_code == 200, resp.text
|
||
cur = resp.json()
|
||
assert cur["recipe_id"] == out["recipe_id"] and cur["recipe"]["mode"] == "replace"
|
||
resp = await client.get(f"/api/datasources/{src['id']}/imports")
|
||
item = resp.json()[0]
|
||
assert item["recipe_id"] == out["recipe_id"] and item["period_start"] == "2026-08-01"
|
||
assert item["overrides"] == 0 and item["waivers"] == 0 and item["signed_by"] == "测试员"
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 真管线：上传新一期
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_real_pipeline_reupload_next_month_receipt_diff_and_activate(client):
|
||
name = unique()
|
||
done = await first_import(client, name)
|
||
source_id = done["commit"]["source"]["id"]
|
||
first_snapshot = done["commit"]["snapshot_id"]
|
||
raw, fn = flow_workbook(SEP, 30, seed=7)
|
||
resp = await client.post(f"/api/datasources/{source_id}/reupload", files={"file": (fn, raw, XLSX)})
|
||
assert resp.status_code == 201, resp.text
|
||
st = resp.json()
|
||
assert st["kind"] == "reupload" and st["status"] == "trialed" and st["draft"] is None
|
||
assert st["ai"]["offered"] is False and st["source"]["import_mode"] == "recipe"
|
||
trial = st["trial"]
|
||
assert trial["status"] == "passed", trial["problems"]
|
||
assert {d["kind"] for d in trial["diff"]} == {"period", "rows"}
|
||
assert not any(d["requires_confirm"] for d in trial["diff"])
|
||
# 配方没变：配方类确认项都不出，只剩本期类（这里没有）
|
||
assert all_confirms(st) == []
|
||
assert trial["base_snapshot_id"] == first_snapshot
|
||
assert {t["name"]: t["rows"] for t in trial["receipt"]["tables"]} == {"日客流": 30, "时段客流": 510,
|
||
"时段客流_表内合计": 90}
|
||
# 数据源列表里看得到这个未结束的导入
|
||
resp = await client.get("/api/datasources")
|
||
row = next(s for s in resp.json() if s["id"] == source_id)
|
||
assert row["open_staging"]["id"] == st["id"] and row["open_staging"]["kind"] == "reupload"
|
||
|
||
resp = await client.post(f"/api/datasources/imports/{st['id']}/commit",
|
||
json={"trial_id": trial["trial_id"], "confirmations": []})
|
||
assert resp.status_code == 201, resp.text
|
||
out = resp.json()
|
||
assert out["unchanged"] is False and out["recipe_id"] == done["commit"]["recipe_id"]
|
||
assert out["source"]["current_snapshot"]["id"] == out["snapshot_id"] != first_snapshot
|
||
assert await scalar(source_id, 'SELECT COUNT(*) FROM "日客流"') == 30
|
||
async with SessionLocal() as session:
|
||
recipes = list((await session.execute(
|
||
select(TableRecipe).where(TableRecipe.source_id == source_id))).scalars())
|
||
imps = list((await session.execute(
|
||
select(TableImport).where(TableImport.source_id == source_id).order_by(TableImport.seq))).scalars())
|
||
assert len(recipes) == 1, "配方没变：不新建配方"
|
||
# 被替换的那一期：发布后的回收会把没有运行引用的版本标成 retired（原件按配方源最近 12 次保留）
|
||
assert imps[0].status in ("superseded", "retired") and imps[1].status == "active"
|
||
assert imps[0].raw_state == "kept"
|
||
assert imps[1].recipe_id == recipes[0].id and imps[1].period_start == "2026-09-01"
|
||
|
||
|
||
async def test_real_pipeline_same_file_again_is_unchanged(client):
|
||
name = unique()
|
||
raw, fn = flow_workbook(AUG, 31, seed=11)
|
||
done = await first_import(client, name, seed=11)
|
||
source_id = done["commit"]["source"]["id"]
|
||
resp = await client.post(f"/api/datasources/{source_id}/reupload", files={"file": (fn, raw, XLSX)})
|
||
assert resp.status_code == 201, resp.text
|
||
st = resp.json()
|
||
trial = st["trial"]
|
||
assert trial["same_as_import"]["id"] == done["commit"]["import_id"]
|
||
assert trial["diff"] == []
|
||
resp = await client.post(f"/api/datasources/imports/{st['id']}/commit",
|
||
json={"trial_id": trial["trial_id"], "confirmations": all_confirms(st)})
|
||
assert resp.status_code == 201, resp.text
|
||
out = resp.json()
|
||
assert out["unchanged"] is True and out["import_id"] == done["commit"]["import_id"]
|
||
assert out["snapshot_id"] == done["commit"]["snapshot_id"]
|
||
async with SessionLocal() as session:
|
||
n = len(list((await session.execute(
|
||
select(TableImport).where(TableImport.source_id == source_id))).scalars()))
|
||
assert n == 1
|
||
row = await get(ImportStaging, st["id"])
|
||
assert row.status == "committed" and row.trial_path is None
|
||
|
||
|
||
async def test_real_pipeline_changed_context_input_invalidates_earlier_trial_id(client):
|
||
"""trial_key 含人工录入（WP-5c：改配方、改录入后失效）：同一配方、同一统计期换了署名再试运行，trial_id
|
||
必须换；拿前一次（另一份录入）的 trial_id 提交返回 409 trial_required，不能发布成最新那次的结果。"""
|
||
raw, fn = flow_workbook(AUG, 31, seed=1201, variant="noperiod")
|
||
resp = await stage(client, unique(), raw, fn)
|
||
assert resp.status_code == 201, resp.text
|
||
sid = resp.json()["id"]
|
||
resp = await client.put(f"/api/datasources/imports/{sid}/recipe", json={"recipe": FLOW_RECIPE})
|
||
assert resp.status_code == 200, resp.text
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/trial", json={})
|
||
assert resp.status_code == 200 and resp.json()["trial"]["status"] == "needs_input", resp.text
|
||
period = {"统计期": {"start": "2026-08-01", "end": "2026-08-31"}}
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/trial",
|
||
json={"context_inputs": period, "signed_by": "录入员甲"})
|
||
first = resp.json()
|
||
assert first["trial"]["status"] == "passed", first["trial"]["problems"]
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/trial",
|
||
json={"context_inputs": period, "signed_by": "录入员乙"})
|
||
second = resp.json()["trial"]
|
||
assert second["status"] == "passed" and second["trial_id"] != first["trial"]["trial_id"]
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/commit", json={
|
||
"trial_id": first["trial"]["trial_id"], "confirmations": all_confirms(first)})
|
||
assert resp.status_code == 409 and resp.json()["code"] == "trial_required", resp.text
|
||
assert (await get(ImportStaging, sid)).status != "committed"
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 真管线：AI 起草（假模型）
|
||
# --------------------------------------------------------------------------
|
||
|
||
KV_RECIPE = {
|
||
"recipe_format": "agentlab-recipe/2",
|
||
"sheets": [{"id": "s1", "match": {"name": "基本情况"}, "blocks": [{
|
||
"id": "列表1", "layout": "list", "table": "基本情况", "after_title": "基本情况登记表",
|
||
"columns": [{"header": "单位名称", "name": "项目", "type": "TEXT"},
|
||
{"header": "示例单位甲", "name": "取值", "type": "TEXT"}]}]}],
|
||
"tables": [{"name": "基本情况"}],
|
||
}
|
||
|
||
|
||
async def test_real_pipeline_ai_draft_sends_exactly_the_previewed_text(client, monkeypatch):
|
||
"""规则起草认不出的表单（lab.kv_form）→ 查看要发送的全文 → 同意 → 假模型起草 → 试运行 → 提交。
|
||
|
||
模型第一次收到的系统提示 + 提示词，与同意框里展示的全文逐字相同；不调用任何真实模型。
|
||
"""
|
||
from langchain_core.messages import AIMessage
|
||
|
||
from app.data import recipe_ai
|
||
from app.data.recipe_types import AiAvailability
|
||
from tests.fixtures.xlsx.lab import kv_form
|
||
|
||
calls: list[list[Any]] = []
|
||
|
||
class Model:
|
||
async def ainvoke(self, messages):
|
||
calls.append(list(messages))
|
||
return AIMessage(content=json.dumps(KV_RECIPE, ensure_ascii=False),
|
||
usage_metadata={"input_tokens": 3000, "output_tokens": 400, "total_tokens": 3400})
|
||
|
||
async def resolve(session):
|
||
return Model(), "fake-model", AiAvailability(True, model="fake-model", provider="假接入")
|
||
|
||
monkeypatch.setattr(recipe_ai, "resolve_draft_model", resolve)
|
||
raw, fn = kv_form()
|
||
resp = await stage(client, unique(), raw, fn)
|
||
assert resp.status_code == 201, resp.text
|
||
st = resp.json()
|
||
assert st["draft"]["complete"] is False and st["ai"]["offered"] is True and st["ai"]["available"] is True
|
||
sid = st["id"]
|
||
resp = await client.get(f"/api/datasources/imports/{sid}/draft-ai/preview")
|
||
assert resp.status_code == 200, resp.text
|
||
pv = resp.json()
|
||
assert pv["model"] == "fake-model" and "# 配方 JSON Schema" in pv["text"] and "# 单位词表" in pv["text"]
|
||
assert "示例单位甲" not in pv["text"].split("# 工作表压缩表示")[0], "系统提示里没有表格内容"
|
||
assert "35600" not in pv["text"] and "860" not in pv["text"], "数字格的值不外发"
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/draft-ai",
|
||
json={"consent": True, "preview_sha256": pv["sha256"], "signed_by": "测试员"})
|
||
assert resp.status_code == 200, resp.text
|
||
st = resp.json()
|
||
first = calls[0]
|
||
assert first[0].content + "\n\n" + first[1].content == pv["text"]
|
||
assert st["recipe_origin"] == "ai" and st["recipe_problems"] == [] and st["cards"][0]["id"] == "ai_origin"
|
||
assert st["ai_consents"][0]["preview_sha256"] == pv["sha256"] and st["ai_draft"]["total_tokens"] == 3400
|
||
|
||
st = (await client.post(f"/api/datasources/imports/{sid}/trial", json={})).json()
|
||
assert st["trial"]["status"] == "passed", st["trial"]["problems"]
|
||
ids = all_confirms(st)
|
||
assert {"ai_origin", "rename:基本情况.项目", "columns:基本情况"} <= set(ids)
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/commit",
|
||
json={"trial_id": st["trial"]["trial_id"],
|
||
"confirmations": [i for i in ids if i != "ai_origin"]})
|
||
assert resp.status_code == 422 and "ai_origin" in resp.json()["detail"]
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/commit",
|
||
json={"trial_id": st["trial"]["trial_id"], "confirmations": ids})
|
||
assert resp.status_code == 201, resp.text
|
||
out = resp.json()
|
||
rec = await get(TableRecipe, out["recipe_id"])
|
||
assert rec.origin == "ai" and rec.ai_usage["calls"][0]["input_tokens"] == 3000
|
||
imp = await get(TableImport, out["import_id"])
|
||
manifest = artifact_store.load(imp.manifest_artifact)
|
||
assert manifest["ai"]["consents"][0]["preview_sha256"] == pv["sha256"]
|
||
assert await scalar(out["source"]["id"], 'SELECT COUNT(*) FROM "基本情况"') > 0
|
||
|
||
|
||
async def test_real_pipeline_model_switch_after_preview_is_stale_and_model_resolved_once(client, monkeypatch):
|
||
"""AU-7 / SE-4：看完预览后助手的模型从本机换成外部服务 → 409 ai_preview_stale，不发送；重新查看再同意时，
|
||
同意记录、实际调用、用量记的是同一个模型，POST 里只解析一次模型。"""
|
||
from langchain_core.messages import AIMessage
|
||
|
||
from app.data import recipe_ai
|
||
from app.data.recipe_types import AiAvailability
|
||
from tests.fixtures.xlsx.lab import kv_form
|
||
|
||
sent: list[str] = []
|
||
current = {"name": "local-model-A", "provider": "本机接入"}
|
||
resolves: list[str] = []
|
||
|
||
class Model:
|
||
def __init__(self, name: str) -> None:
|
||
self.name = name
|
||
|
||
async def ainvoke(self, messages):
|
||
sent.append(self.name)
|
||
resolves.append("send")
|
||
return AIMessage(content=json.dumps(KV_RECIPE, ensure_ascii=False),
|
||
usage_metadata={"input_tokens": 10, "output_tokens": 5, "total_tokens": 15})
|
||
|
||
async def resolve(session):
|
||
resolves.append(current["name"])
|
||
return (Model(current["name"]), current["name"],
|
||
AiAvailability(True, model=current["name"], provider=current["provider"]))
|
||
|
||
monkeypatch.setattr(recipe_ai, "resolve_draft_model", resolve)
|
||
raw, fn = kv_form()
|
||
sid = (await stage(client, unique(), raw, fn)).json()["id"]
|
||
pv = (await client.get(f"/api/datasources/imports/{sid}/draft-ai/preview")).json()
|
||
assert pv["model"] == "local-model-A"
|
||
current.update(name="cloud-model-B", provider="外部接入")
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/draft-ai",
|
||
json={"consent": True, "preview_sha256": pv["sha256"]})
|
||
assert resp.status_code == 409 and resp.json()["code"] == "ai_preview_stale", resp.text
|
||
assert sent == []
|
||
pv = (await client.get(f"/api/datasources/imports/{sid}/draft-ai/preview")).json()
|
||
assert pv["model"] == "cloud-model-B"
|
||
resolves.clear()
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/draft-ai",
|
||
json={"consent": True, "preview_sha256": pv["sha256"]})
|
||
assert resp.status_code == 200, resp.text
|
||
st = resp.json()
|
||
# 发送之前只解析了一次（同意的门槛里那次）；之后的是响应里查可用性，与发送无关
|
||
assert resolves[:resolves.index("send")] == ["cloud-model-B"], resolves
|
||
assert sent == ["cloud-model-B"]
|
||
assert st["ai_consents"][-1]["model"] == "cloud-model-B" and st["ai_consents"][-1]["provider"] == "外部接入"
|
||
assert [u["model"] for u in st["ai_draft"]["usage"]] == ["cloud-model-B"]
|
||
|
||
|
||
# --------------------------------------------------------------------------
|
||
# 假管线：接口形状与错误码
|
||
# --------------------------------------------------------------------------
|
||
|
||
|
||
async def test_stage_errors(client, fake, monkeypatch):
|
||
resp = await api_stage(client, "Bad-Name")
|
||
assert resp.status_code == 400 and resp.json()["code"] == "name_invalid"
|
||
resp = await client.post("/api/datasources/imports/stage", data={"name": unique()})
|
||
assert resp.status_code == 422 and resp.json()["code"] == "body_invalid"
|
||
resp = await api_stage(client, unique(), raw=b"bad file")
|
||
assert resp.status_code == 400 and resp.json()["code"] == "file_unreadable"
|
||
assert "无法读取" in resp.json()["detail"]
|
||
# 手工源同名 → name_taken；已按配方导入 → recipe_source_exists
|
||
manual = unique()
|
||
async with SessionLocal() as session:
|
||
session.add(DataSource(name=manual, kind="sqlite", database="/tmp/合成.db"))
|
||
await session.commit()
|
||
resp = await api_stage(client, manual)
|
||
assert resp.status_code == 409 and resp.json()["code"] == "name_taken"
|
||
done = await recipe_source(client)
|
||
resp = await api_stage(client, done["source"]["name"])
|
||
assert resp.status_code == 409 and resp.json()["code"] == "recipe_source_exists"
|
||
# 超限 → 413（两条上传路径共用 read_upload）
|
||
monkeypatch.setattr(settings, "max_upload_mb", 0)
|
||
resp = await api_stage(client, unique())
|
||
assert resp.status_code == 413 and resp.json()["code"] == "file_too_large"
|
||
resp = await client.post("/api/datasources/upload", files={"file": ("x.xlsx", b"PK", XLSX)},
|
||
data={"name": unique()})
|
||
assert resp.status_code == 413 and resp.json()["code"] == "file_too_large"
|
||
|
||
|
||
async def test_not_found_and_body_invalid(client, fake):
|
||
for method, path in (("get", "/api/datasources/imports/nope"), ("delete", "/api/datasources/imports/nope")):
|
||
resp = await getattr(client, method)(path)
|
||
assert resp.status_code == 404 and resp.json()["code"] == "staging_not_found", path
|
||
resp = await client.post("/api/datasources/imports/nope/trial", json={})
|
||
assert resp.status_code == 404 and resp.json()["code"] == "staging_not_found"
|
||
st = await staged(client)
|
||
sid = st["id"]
|
||
for method, path, kw in (
|
||
("post", "answers", {"content": b"[1, 2]"}), ("put", "recipe", {"content": b"not json"}),
|
||
("put", "recipe", {"json": {"recipe": [1]}}), ("post", "answers", {"json": {"answers": ["x"]}}),
|
||
("post", "commit", {"json": {"trial_id": "x", "confirmations": "mode"}}),
|
||
("post", "commit", {"json": {"trial_id": "x", "acceptances": {"R1": "理由"}}}),
|
||
):
|
||
headers = {"content-type": "application/json"} if "content" in kw else None
|
||
resp = await getattr(client, method)(f"/api/datasources/imports/{sid}/{path}", headers=headers, **kw)
|
||
assert resp.status_code == 422 and resp.json()["code"] == "body_invalid", (path, kw)
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/commit", json={})
|
||
assert resp.status_code == 409 and resp.json()["code"] == "trial_required"
|
||
|
||
|
||
async def test_put_recipe_with_schema_problems_is_stored_but_cannot_trial(client, fake):
|
||
st = await staged(client)
|
||
sid = st["id"]
|
||
broken = {"recipe_format": "agentlab-recipe/2", "tables": []}
|
||
resp = await client.put(f"/api/datasources/imports/{sid}/recipe", json={"recipe": broken})
|
||
assert resp.status_code == 200, resp.text
|
||
out = resp.json()
|
||
assert out["recipe"] == broken and out["recipe_problems"][0]["code"] == "schema"
|
||
assert out["recipe_problems"][0]["path"]
|
||
assert fake.calls["detect_facts"][-1] == broken
|
||
resp = await client.post(f"/api/datasources/imports/{sid}/trial", json={})
|
||
assert resp.status_code == 409 and resp.json()["code"] == "recipe_invalid"
|
||
# 改回合法的配方：问题集合按新的起点重算
|
||
fake.calls["questions_for"].clear()
|
||
resp = await client.put(f"/api/datasources/imports/{sid}/recipe", json={"recipe": FLOW_FULL})
|
||
assert resp.json()["recipe_problems"] == [] and fake.calls["questions_for"][-1]["recipe"] == FLOW_FULL
|
||
|
||
|
||
async def test_signed_by_from_header_and_body(client, fake):
|
||
st = await staged(client)
|
||
from urllib.parse import quote
|
||
|
||
resp = await client.post(f"/api/datasources/imports/{st['id']}/trial", json={},
|
||
headers={"X-Actor": quote("页眉署名")})
|
||
assert resp.status_code == 200
|
||
assert (await get(ImportStaging, st["id"])).signed_by == "页眉署名"
|
||
resp = await client.post(f"/api/datasources/imports/{st['id']}/trial", json={"signed_by": "请求体署名"},
|
||
headers={"X-Actor": quote("页眉署名")})
|
||
assert (await get(ImportStaging, st["id"])).signed_by == "请求体署名"
|
||
|
||
|
||
async def test_datasource_out_new_fields(client, fake):
|
||
manual = unique()
|
||
async with SessionLocal() as session:
|
||
session.add(DataSource(name=manual, kind="sqlite", database="/tmp/合成.db"))
|
||
await session.commit()
|
||
done = await recipe_source(client)
|
||
rows = {s["name"]: s for s in (await client.get("/api/datasources")).json()}
|
||
assert rows[manual]["import_mode"] is None and rows[manual]["current_recipe"] is None
|
||
assert rows[manual]["open_staging"] is None
|
||
mine = rows[done["source"]["name"]]
|
||
assert mine["import_mode"] == "recipe"
|
||
assert set(mine["current_recipe"]) == {"id", "seq", "origin", "activated_at", "signed_by"}
|
||
re_ = (await client.post(f"/api/datasources/{mine['id']}/reupload",
|
||
files={"file": ("x.xlsx", raw_bytes(), XLSX)})).json()
|
||
rows = {s["name"]: s for s in (await client.get("/api/datasources")).json()}
|
||
assert rows[done["source"]["name"]]["open_staging"]["id"] == re_["id"]
|
||
assert set(rows[done["source"]["name"]]["open_staging"]) == {"id", "kind", "status", "created_at"}
|
||
|
||
|
||
async def test_routes_are_not_taken_by_datasource_wildcards(client, fake):
|
||
"""/imports/… 和 /{source_id}/reupload|redraft|recipe 不被 datasources 的通配路由截走：每个打一次，看 code。"""
|
||
resp = await client.get("/api/datasources/imports/imports")
|
||
assert resp.json().get("code") == "staging_not_found"
|
||
resp = await client.post("/api/datasources/nope/reupload", files={"file": ("x.xlsx", b"PK", XLSX)})
|
||
assert resp.status_code == 404 and resp.json().get("code") == "source_not_found"
|
||
resp = await client.get("/api/datasources/nope/recipe")
|
||
assert resp.status_code == 404 and resp.json().get("code") == "source_not_found"
|
||
resp = await client.delete("/api/datasources/imports/nope")
|
||
assert resp.json().get("code") == "staging_not_found"
|
||
resp = await client.get("/api/datasources/imports/nope/draft-ai/preview")
|
||
assert resp.json().get("code") == "staging_not_found"
|
||
|
||
|
||
async def test_switch_from_simple_reports_mask_lost_and_compress_masks_case_insensitively(client, monkeypatch):
|
||
"""AU-6 / SE-9：简单导入遮罩了「工资_元」，切换后规则起草的列名是「工资」：确认项里必勾 mask_lost:工资_元。
|
||
遮罩写成大小写不同的「PHONE」时，AI 压缩表示同样按遮罩列处理（与证据面板一致，不区分大小写）。"""
|
||
import io
|
||
|
||
from openpyxl import Workbook
|
||
|
||
from app.data import recipe_imports
|
||
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "员工"
|
||
ws.append(["工号", "姓名", "Phone", "工资（元）"])
|
||
for i in range(12):
|
||
ws.append([f"E{i:03d}", f"员工{i}", f"1380000{i:04d}", 5000 + i * 137])
|
||
buf = io.BytesIO()
|
||
wb.save(buf)
|
||
raw = buf.getvalue()
|
||
name = unique()
|
||
resp = await client.post("/api/datasources/upload", files={"file": ("员工.xlsx", raw, XLSX)}, data={"name": name})
|
||
assert resp.status_code == 201, resp.text
|
||
src = resp.json()["source"]["id"]
|
||
resp = await client.patch(f"/api/datasources/{src}", json={"options": {"mask_columns": ["工资_元", "PHONE"]}})
|
||
assert resp.status_code == 200, resp.text
|
||
st = (await stage(client, name, raw, "员工.xlsx")).json()
|
||
assert st["kind"] == "switch"
|
||
st = (await client.post(f"/api/datasources/imports/{st['id']}/trial", json={})).json()
|
||
items = {i["id"]: i for i in st["trial"]["confirm_items"]}
|
||
assert "mask_lost:工资_元" in items, list(items)
|
||
assert "mask_lost:PHONE" not in items # 新结构里还有 Phone（不区分大小写）
|
||
# SE-9：压缩表示的遮罩表头不区分大小写
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import ImportStaging
|
||
|
||
async with SessionLocal() as session:
|
||
staging = await session.get(ImportStaging, st["id"])
|
||
heads = await recipe_imports._mask_headers(session, staging)
|
||
assert "Phone" in heads and "工资（元）" in heads
|
||
|
||
|
||
async def test_bad_number_cell_is_file_unreadable_with_fixed_text(client):
|
||
"""SE-7：数字格的 <v> 里写了非数字：stage 回 400 file_unreadable（带 code、固定中文文案），不回英文原文和格子内容。"""
|
||
from openpyxl import Workbook
|
||
|
||
from tests.fixtures.xlsx import patch_parts, save
|
||
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "S"
|
||
ws["A1"], ws["B1"] = "项目", "数量"
|
||
for r in range(2, 8):
|
||
ws.cell(r, 1, f"甲{r}")
|
||
ws.cell(r, 2, r)
|
||
raw = patch_parts(save(wb), {"xl/worksheets/sheet1.xml": lambda x: x.replace("<v>3</v>", "<v>机密值abc</v>", 1)})
|
||
resp = await stage(client, unique(), raw, "bad.xlsx")
|
||
assert resp.status_code == 400 and resp.json()["code"] == "file_unreadable", resp.text
|
||
assert "机密值abc" not in resp.text and "invalid literal" not in resp.text
|
||
|
||
|
||
async def test_json_body_size_and_depth_limits(client, fake):
|
||
"""SE-5：超过 1 MB 的请求体回 413 body_too_large；嵌套超过 32 层回 422 body_invalid，不入库，之后 GET 照常。"""
|
||
from app.api import table_imports as api_mod
|
||
|
||
sid = (await stage(client, unique(), raw_bytes(), "a.xlsx")).json()["id"]
|
||
url = f"/api/datasources/imports/{sid}/recipe"
|
||
big = {"recipe": {"recipe_format": "agentlab-recipe/2", "junk": "x" * (api_mod.BODY_MAX_BYTES + 10)}}
|
||
resp = await client.put(url, json=big)
|
||
assert resp.status_code == 413 and resp.json()["code"] == "body_too_large"
|
||
deep: Any = "x"
|
||
for _ in range(300):
|
||
deep = [deep]
|
||
resp = await client.put(url, json={"recipe": {"recipe_format": "agentlab-recipe/2", "junk": deep}})
|
||
assert resp.status_code == 422 and resp.json()["code"] == "body_invalid"
|
||
resp = await client.get(f"/api/datasources/imports/{sid}")
|
||
assert resp.status_code == 200, resp.text
|
||
# 字符串里的括号不算嵌套
|
||
ok = {"recipe": {"recipe_format": "agentlab-recipe/2", "note": "[" * 100}}
|
||
resp = await client.put(url, json=ok)
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
|
||
async def test_recipe_problems_are_capped(client, monkeypatch):
|
||
"""SE-5：几千个问题只存、只回前 200 条，另附一条总数。"""
|
||
from app.data import recipe_imports
|
||
from tests.fixtures.xlsx.lab import kv_form
|
||
|
||
raw, fn = kv_form()
|
||
sid = (await stage(client, unique(), raw, fn)).json()["id"]
|
||
rec = json.loads(json.dumps(KV_RECIPE, ensure_ascii=False))
|
||
rec["tables"][0]["units"] = {f"不存在的列{i}": "元" for i in range(1000)}
|
||
resp = await client.put(f"/api/datasources/imports/{sid}/recipe", json={"recipe": rec})
|
||
assert resp.status_code == 200, resp.text
|
||
probs = resp.json()["recipe_problems"]
|
||
assert len(probs) == recipe_imports.RECIPE_PROBLEMS_MAX + 1
|
||
assert probs[-1]["code"] == "too_many_problems" and "共 1000 个" in probs[-1]["message"]
|