Files
AgentLab/backend/tests/test_catalog_transfer.py

433 lines
24 KiB
Python

"""数据目录的导出导入(data/catalog_transfer.py):本机整理好的目录带到部署实例上。
- GET /catalog/export:附件「<数据源名>-数据目录.json」,{format: agentlab-catalog, version: 1, exported_at,
source: {name, kind}, guidance, tables: {表名: {notes, version}}},含全部状态的项(表结构里已经没有的表也带)
- POST /catalog/import/preview:只读,报告每张对上的表新增、更新、保留、冲突、跳过了什么
- POST /catalog/import:同样的报告,加每张表写入后的版本。规则:
- 表按名字对(忽略大小写;带 schema 前缀的对不上时去掉前缀再对),对不上的列在 unmatched_tables
- 文件里已确认、已驳回的项照带(来源、状态不变);目标里已确认、已驳回的项和文件不一样时保留目标的、记冲突
- 文件里推断、有确证的项只填目标的空缺,不覆盖目标已有的
- 表结构里没有的列、关系两端对不上的跳过并写原因;目标数据源遮罩的列,码值不导入;有效记录条件用到的列(按目标的方言
解析)不在目标的表结构里、引用别的表的列、带子查询的跳过(人工写的条件不限函数)
- 逐表乐观锁,一张表写不进去不影响别的;有排队或进行中的起草任务时 409 catalog_draft_busy;文件格式不对 422
- 起草要求:勾了导入、目标还没有时才写;目标已有的不覆盖
表名、列名、业务数字都是虚构的。
"""
from __future__ import annotations
import copy
import json
import uuid
from urllib.parse import quote, unquote
import pytest
from httpx import ASGITransport, AsyncClient
from app.data import catalog, catalog_jobs, catalog_transfer
from app.data.catalog import make_item
from app.data.engine import CATALOG_DRAFT_GUIDANCE_OPTION
from app.db.base import SessionLocal
from app.db.models import CatalogDraftJob, DataSource
from app.main import app as _fastapi_app
from tests.fixtures.sources import drop_source
ACTOR = {"X-Actor": quote("沈岚")}
def _col(name: str, type_: str = "INTEGER") -> dict:
return {"name": name, "type": type_, "nullable": True, "comment": None}
def _tables(*, upper: bool = False) -> dict:
"""两边的表结构:目标那边(upper)表名、列名写法不同,还少了一列 tier。"""
def n(name: str) -> str:
return name.upper() if upper else name
members_cols = [_col(n("id")), _col(n("nick"), "TEXT")] + ([] if upper else [_col("tier")])
return {
n("orders"): {"qualified": n("orders"), "primary_key": [n("id")], "comment": None, "is_view": False,
"foreign_keys": [], "unique": [],
"columns": [_col(n("id")), _col(n("status")), _col(n("amount"), "REAL"), _col(n("member_id")),
_col(n("buyer_phone"), "TEXT")]},
n("members"): {"qualified": n("members"), "primary_key": [n("id")], "comment": None, "is_view": False,
"foreign_keys": [], "unique": [], "columns": members_cols},
n("refunds"): {"qualified": n("refunds"), "primary_key": [n("id")], "comment": None, "is_view": False,
"foreign_keys": [], "unique": [], "columns": [_col(n("id")), _col(n("order_id"))]},
}
def _rel(to_table: str, columns: list[str], to_columns: list[str], *, status: str, source: str = "name",
table: str = "orders") -> dict:
return {"id": catalog.relation_id(table, columns, to_table, to_columns), "columns": columns,
"to_table": to_table, "to_columns": to_columns, "cardinality": "many_to_one", "coverage": None,
"source": source, "status": status}
#: 本机(源)的目录
SOURCE_ORDERS = {
"label": make_item("订单", "human", "confirmed", at="2026-10-01T08:00:00+00:00"),
"grain": make_item("每笔订单一行", "llm"),
# 确认、驳回过的项带着保存提案时记下的依据:导出、导入照带
"valid_filter": make_item("status <> 9", "human", "confirmed", reason="使用者说:9 是作废,统计时排除"),
"dedup": make_item("同一订单号取最后一次", "llm", "rejected", reason="使用者说:同一订单号只有一行"),
"kind": make_item("fact", "llm", "confirmed"),
"columns": {
"status": {"codes": make_item({"1": "已支付", "9": "作废"}, "human", "confirmed", complete=True),
"label": make_item("状态", "llm")},
"amount": {"unit": make_item("元", "human", "confirmed"), "measure": make_item("flow", "llm")},
"buyer_phone": {"codes": make_item({"13900000001": "测试号"}, "human", "confirmed"),
"label": make_item("联系电话", "human", "confirmed")},
# 目标表结构里没有这一列
"gift_flag": {"label": make_item("赠品标记", "human", "confirmed")},
},
"relations": [
_rel("members", ["member_id"], ["id"], status="confirmed"),
# 目标那边没有这张表
_rel("stores", ["member_id"], ["id"], status="rejected"),
],
}
SOURCE_MEMBERS = {"label": make_item("会员", "llm"), "keys": make_item(["tier"], "human", "confirmed"),
"columns": {"nick": {"label": make_item("昵称", "llm", "verified")}}}
#: 目标那边没有的表
SOURCE_GHOST = {"label": make_item("旧表", "human", "confirmed")}
#: 目标(部署实例)上已有的目录
TARGET_ORDERS = {
"label": make_item("订单表", "human", "confirmed"), # 和文件不一样、都确认过:冲突,保留目标
"grain": make_item("一行一单", "llm"), # 文件里是推断:保留目标
"valid_filter": make_item("STATUS <> 9", "llm"), # 目标是推断、文件确认过:更新
"kind": make_item("fact", "human", "confirmed"), # 一样:保留
"columns": {"STATUS": {"label": make_item("状态码", "comment")}},
}
_CREATED: list[str] = []
@pytest.fixture(autouse=True)
async def _cleanup():
yield
while _CREATED:
await drop_source(_CREATED.pop())
@pytest.fixture
async def client():
async with AsyncClient(transport=ASGITransport(app=_fastapi_app), base_url="http://test") as c:
yield c
async def _write(sid: str, table: str, notes: dict) -> None:
async with SessionLocal() as session:
entry = await catalog.read_entry(session, sid, table)
await catalog.write_entry(session, sid, table, notes, if_version=entry.version if entry else 0, actor=None)
async def _source(name: str, tables: dict, entries: dict[str, dict], **kw) -> str:
async with SessionLocal() as session:
row = DataSource(name=name, kind="sqlite", database=":memory:", schema_cache={"tables": tables}, **kw)
session.add(row)
await session.commit()
sid = row.id
_CREATED.append(sid)
for table, notes in entries.items():
await _write(sid, table, notes)
return sid
async def _pair() -> tuple[str, str]:
local = await _source(f"本机库_{uuid.uuid4().hex[:6]}", _tables(),
{"orders": SOURCE_ORDERS, "members": SOURCE_MEMBERS, "ghost_table": SOURCE_GHOST},
options={CATALOG_DRAFT_GUIDANCE_OPTION: "金额一律按元"})
remote = await _source(f"deploy_{uuid.uuid4().hex[:6]}", _tables(upper=True), {"ORDERS": TARGET_ORDERS},
options={"mask_columns": "buyer_phone"})
return local, remote
def _base(sid: str) -> str:
return f"/api/datasources/{sid}/catalog"
async def _export(client, sid: str) -> dict:
r = await client.get(f"{_base(sid)}/export")
assert r.status_code == 200, r.text
return r.json()
# --------------------------------------------------------------------------
# 导出
# --------------------------------------------------------------------------
async def test_export_is_a_json_attachment_with_every_item(client):
local, _ = await _pair()
r = await client.get(f"{_base(local)}/export")
assert r.status_code == 200, r.text
assert r.headers["content-type"].startswith("application/json")
disposition = r.headers["content-disposition"]
assert disposition.startswith("attachment;") and "filename=\"agentlab-catalog-" in disposition
async with SessionLocal() as session:
name = (await session.get(DataSource, local)).name
assert unquote(disposition.split("filename*=UTF-8''", 1)[1]) == f"{name}-数据目录.json"
body = r.json()
assert body["format"] == "agentlab-catalog" and body["version"] == 1
assert body["exported_at"] and body["source"] == {"name": name, "kind": "sqlite"}
assert body["guidance"] == "金额一律按元"
assert set(body["tables"]) == {"orders", "members", "ghost_table"}
assert body["tables"]["orders"]["version"] == 1
# 含全部状态的项:驳回的也带
assert body["tables"]["orders"]["notes"]["dedup"]["status"] == "rejected"
assert body["tables"]["orders"]["notes"] == SOURCE_ORDERS
assert body["tables"]["ghost_table"] == {"notes": SOURCE_GHOST, "version": 1}
def test_attachment_name_drops_characters_files_cannot_have():
assert catalog_transfer.attachment_name('景区/客流:库?') == "景区_客流_库_-数据目录.json"
assert catalog_transfer.attachment_name(" ") == "数据源-数据目录.json"
# --------------------------------------------------------------------------
# 预览与导入
# --------------------------------------------------------------------------
def _by(report: dict) -> dict[str, dict]:
return {t["matched_as"]: t for t in report["tables"]}
async def test_preview_reports_and_writes_nothing(client):
local, remote = await _pair()
content = await _export(client, local)
r = await client.post(f"{_base(remote)}/import/preview", json={"content": content, "include_guidance": True})
assert r.status_code == 200, r.text
report = r.json()
assert report["unmatched_tables"] == ["ghost_table"]
tables = _by(report)
assert set(tables) == {"ORDERS", "MEMBERS"}
orders = tables["ORDERS"]
assert orders["table_name"] == "orders"
conflicts = {c["path"]: c for c in orders["conflicts"]}
assert set(conflicts) == {"label"}
label = conflicts["label"]
assert label["label"] == "表的中文名" and label["file_value"] == "订单" and label["target_value"] == "订单表"
assert "保留" in label["why"]
skipped = {s["path"]: s["why"] for s in orders["skipped"]}
assert "columns.gift_flag.label" in skipped and "表结构里没有列 gift_flag" in skipped["columns.gift_flag.label"]
assert "遮罩" in skipped["columns.buyer_phone.codes"]
stores = next(p for p in skipped if p.startswith("relations."))
assert "stores" in skipped[stores]
# 新增:dedup(驳回的照带)、status 码值、amount 两项、联系电话的中文名、关联关系;更新:有效记录条件
# 保留:粒度(文件里是推断,目标已有)、表类型(一样)、STATUS 的中文名(文件里是推断,目标已有)
assert (orders["add"], orders["update"], orders["keep"]) == (6, 1, 3)
members = tables["MEMBERS"]
# 业务主键指着的列 tier 目标那边没有:跳过;会员的中文名和昵称填空缺
assert (members["add"], members["update"], members["keep"]) == (2, 0, 0)
assert [s["path"] for s in members["skipped"]] == ["keys"]
assert report["totals"] == {"tables": 2, "add": 8, "update": 1, "keep": 3, "conflicts": 1,
"skipped": len(orders["skipped"]) + 1}
assert report["guidance"] == {"file": "金额一律按元", "target": None, "action": "replace"}
async with SessionLocal() as session:
assert (await catalog.read_entry(session, remote, "ORDERS")).version == 1
assert await catalog.read_entry(session, remote, "MEMBERS") is None
assert catalog_jobs.guidance_of(await session.get(DataSource, remote)) is None
async def test_import_writes_by_the_rules_and_reports_versions(client):
local, remote = await _pair()
content = await _export(client, local)
r = await client.post(f"{_base(remote)}/import", json={"content": content, "include_guidance": True},
headers=ACTOR)
assert r.status_code == 200, r.text
report = r.json()
tables = _by(report)
assert tables["ORDERS"]["version"] == 2 and tables["MEMBERS"]["version"] == 1
assert tables["ORDERS"]["error"] is None
assert report["guidance"]["action"] == "replace"
async with SessionLocal() as session:
orders = await catalog.read_entry(session, remote, "ORDERS")
members = await catalog.read_entry(session, remote, "MEMBERS")
row = await session.get(DataSource, remote)
notes = orders.notes
assert orders.updated_by == "沈岚"
assert notes["label"]["value"] == "订单表" # 冲突:保留目标的
assert notes["grain"]["value"] == "一行一单" # 推断只填空缺
assert notes["valid_filter"] == {**SOURCE_ORDERS["valid_filter"]} # 照带:来源、状态、依据不变
assert notes["dedup"]["status"] == "rejected" and notes["dedup"]["source"] == "llm"
assert notes["dedup"]["reason"] == "使用者说:同一订单号只有一行"
codes = notes["columns"]["STATUS"]["codes"]
assert codes["value"] == {"1": "已支付", "9": "作废"} and codes["complete"] is True
assert notes["columns"]["STATUS"]["label"]["value"] == "状态码"
assert notes["columns"]["AMOUNT"]["measure"] == SOURCE_ORDERS["columns"]["amount"]["measure"]
# 遮罩列:中文名照带,码值不导入
assert notes["columns"]["BUYER_PHONE"] == {"label": SOURCE_ORDERS["columns"]["buyer_phone"]["label"]}
assert "GIFT_FLAG" not in notes["columns"] and "gift_flag" not in notes["columns"]
[rel] = notes["relations"]
assert rel["id"] == catalog.relation_id("ORDERS", ["MEMBER_ID"], "MEMBERS", ["ID"])
assert (rel["columns"], rel["to_table"], rel["to_columns"]) == (["MEMBER_ID"], "MEMBERS", ["ID"])
assert rel["status"] == "confirmed" and rel["source"] == "name"
assert catalog.validate_notes(notes) == []
assert members.notes["label"]["status"] == "proposed" and "keys" not in members.notes
assert members.notes["columns"]["NICK"]["label"]["status"] == "verified"
assert catalog_jobs.guidance_of(row) == "金额一律按元"
# 再导一次:都对上了,什么都不写;目标已经有起草要求,不覆盖
again = (await client.post(f"{_base(remote)}/import", json={"content": content, "include_guidance": True})).json()
assert again["totals"]["add"] == again["totals"]["update"] == 0
assert _by(again)["ORDERS"]["version"] == 2
assert again["guidance"] == {"file": "金额一律按元", "target": "金额一律按元", "action": "keep"}
async def test_target_rejection_beats_file_confirmation_and_guidance_is_optional(client):
local, remote = await _pair()
await _write(remote, "MEMBERS", {"label": make_item("会员", "llm", "rejected")})
async with SessionLocal() as session:
row = await session.get(DataSource, remote)
await catalog_jobs.save_guidance(session, row, "部署实例自己的要求")
content = await _export(client, local)
content["tables"]["members"]["notes"]["label"] = make_item("会员", "human", "confirmed")
report = (await client.post(f"{_base(remote)}/import",
json={"content": content, "include_guidance": False})).json()
members = _by(report)["MEMBERS"]
[conflict] = members["conflicts"]
assert conflict["path"] == "label" and conflict["target_status"] == "rejected"
assert conflict["file_status"] == "confirmed" and "驳回" in conflict["why"]
assert report["guidance"] == {"file": "金额一律按元", "target": "部署实例自己的要求", "action": "none"}
async with SessionLocal() as session:
entry = await catalog.read_entry(session, remote, "MEMBERS")
assert entry.notes["label"]["status"] == "rejected"
assert catalog_jobs.guidance_of(await session.get(DataSource, remote)) == "部署实例自己的要求"
async def test_one_table_failing_does_not_stop_the_others(client, monkeypatch):
local, remote = await _pair()
content = await _export(client, local)
real = catalog.write_entry
async def flaky(session, source_id, table, notes, **kw):
if table == "ORDERS":
raise catalog.CatalogConflict(table, 99)
return await real(session, source_id, table, notes, **kw)
monkeypatch.setattr(catalog, "write_entry", flaky)
report = (await client.post(f"{_base(remote)}/import", json={"content": content})).json()
tables = _by(report)
assert tables["ORDERS"]["error"] and "稍后" in tables["ORDERS"]["error"]
assert tables["ORDERS"]["add"] == tables["ORDERS"]["update"] == 0
assert tables["MEMBERS"]["error"] is None and tables["MEMBERS"]["version"] == 1
async def test_import_waits_for_draft_jobs_and_rejects_bad_files(client):
local, remote = await _pair()
content = await _export(client, local)
async with SessionLocal() as session:
session.add(CatalogDraftJob(id=uuid.uuid4().hex, source_id=remote, status="running"))
await session.commit()
busy = await client.post(f"{_base(remote)}/import", json={"content": content})
assert busy.status_code == 409 and busy.json()["code"] == "catalog_draft_busy"
# 预览是只读的,照常给
assert (await client.post(f"{_base(remote)}/import/preview", json={"content": content})).status_code == 200
bad = [
"不是对象",
{**content, "format": "something-else"},
{**content, "version": 2},
{**content, "tables": []},
{**content, "tables": {"orders": {"notes": "x"}}},
{**content, "guidance": 3},
]
for body in bad:
r = await client.post(f"{_base(local)}/import/preview", json={"content": body})
assert r.status_code == 422, body
assert any("一" <= ch <= "鿿" for ch in r.json()["detail"])
# 界面直接交来文件原文也认;不是 JSON 的 422
as_text = await client.post(f"{_base(local)}/import/preview", json={"content": json.dumps(content)})
assert as_text.status_code == 200 and as_text.json()["totals"]["tables"] == 2
not_json = await client.post(f"{_base(local)}/import/preview", json={"content": "{不是 JSON"})
assert not_json.status_code == 422 and "JSON" in not_json.json()["detail"]
too_long = {**content, "guidance": "长" * (catalog_jobs.GUIDANCE_MAX + 1)}
r = await client.post(f"{_base(local)}/import/preview", json={"content": too_long, "include_guidance": True})
assert r.status_code == 422 and "起草要求" in r.json()["detail"]
async def test_invalid_items_are_skipped_not_fatal(client):
local, remote = await _pair()
content = await _export(client, local)
notes = content["tables"]["orders"]["notes"]
notes["kind"] = {"value": "event", "source": "human", "status": "confirmed"} # 取值不在可选值内
notes["formula"] = {"value": "sum(amount)", "source": "human", "status": "confirmed"} # 不认识的字段
notes["columns"]["amount"]["colour"] = {"value": "红", "source": "human", "status": "confirmed"}
report = (await client.post(f"{_base(remote)}/import/preview", json={"content": content})).json()
skipped = {s["path"]: s["why"] for s in _by(report)["ORDERS"]["skipped"]}
assert "明细表" in skipped["kind"]
assert "formula" in skipped["formula"] and "columns.amount.colour" in skipped
async def test_valid_filter_columns_must_be_in_the_target_table(client):
"""有效记录条件是表级的一项,按列跳过的规矩拦不住它:解析出用到的列,不在目标的表结构里、是别的表的列、带子查询的都跳过
并写原因。人工写的条件不限函数;带本表表名(文件里的写法、目标的写法都认)、大小写不同的照带。"""
local, remote = await _pair()
content = await _export(client, local)
async def outcome(condition: str) -> tuple[str | None, int]:
body = copy.deepcopy(content)
body["tables"]["orders"]["notes"]["valid_filter"] = make_item(condition, "human", "confirmed")
report = (await client.post(f"{_base(remote)}/import/preview", json={"content": body})).json()
orders = _by(report)["ORDERS"]
why = {s["path"]: s["why"] for s in orders["skipped"]}.get("valid_filter")
return why, orders["update"]
why, update = await outcome("gift_flag = 0 AND status <> 9")
assert why and "gift_flag" in why and "目标的表结构" in why and update == 0
why, _ = await outcome("members.tier = 1")
assert why and "members.tier" in why and "别的表" in why
why, _ = await outcome("member_id IN (SELECT id FROM members)")
assert why and "子查询" in why
why, _ = await outcome("status <> 9; DELETE FROM orders")
assert why and "条件" in why
for fine in ("orders.status <> 9", "ORDERS.STATUS <> 9 AND ABS(amount) > 0", "buyer_phone IS NOT NULL"):
why, update = await outcome(fine)
assert why is None and update == 1, fine
async def test_import_needs_a_schema(client):
local, _ = await _pair()
content = await _export(client, local)
empty = await _source(f"empty_{uuid.uuid4().hex[:6]}", {}, {})
r = await client.post(f"{_base(empty)}/import/preview", json={"content": content})
assert r.status_code == 409 and r.json()["code"] == "schema_missing"
async def test_round_trip_into_an_empty_copy_reproduces_the_catalog(client):
local, _ = await _pair()
content = await _export(client, local)
copy_sid = await _source(f"copy_{uuid.uuid4().hex[:6]}", _tables(), {})
report = (await client.post(f"{_base(copy_sid)}/import", json={"content": json.loads(json.dumps(content)),
"include_guidance": True})).json()
assert report["unmatched_tables"] == ["ghost_table"]
assert all(not t["conflicts"] for t in report["tables"])
async with SessionLocal() as session:
mine = await catalog.read_entry(session, local, "orders")
theirs = await catalog.read_entry(session, copy_sid, "orders")
# 关系两端都对得上的照带;指向目标那边没有的表的跳过
expected = copy.deepcopy(mine.notes)
expected["relations"] = [r for r in mine.notes["relations"] if r["to_table"] != "stores"]
del expected["columns"]["gift_flag"] # 表结构里没有这一列
assert theirs.notes == expected
async def test_a_bad_reason_on_an_inferred_item_is_dropped_not_the_item(client):
"""手改过的文件里,推断的项带着不合规的确认时的依据(推断项本来就不该有依据):去掉依据、照常填空缺,不把整项跳过。
已确认的项依据不合规照旧跳过(依据是它的一部分,照带就得合规)。"""
local, remote = await _pair()
content = await _export(client, local)
notes = content["tables"]["orders"]["notes"]
notes["grain"] = {"value": "每笔订单一行", "source": "llm", "status": "proposed", "reason": "依" * 600}
notes["description"] = {"value": "订单头", "source": "human", "status": "confirmed", "reason": 42}
report = (await client.post(f"{_base(remote)}/import/preview", json={"content": content})).json()
skipped = {s["path"]: s["why"] for s in _by(report)["ORDERS"]["skipped"]}
assert "grain" not in skipped
assert "description" in skipped and "依据" in skipped["description"]
r = await client.post(f"{_base(remote)}/import", json={"content": content})
assert r.status_code == 200, r.text
assert "grain" not in {s["path"] for s in _by(r.json())["ORDERS"]["skipped"]}
got = (await client.get(f"{_base(remote)}/ORDERS")).json()["notes"]
assert "reason" not in got["grain"] # 目标原有的推断粒度保留(推断只填空缺),也没被挂上依据