mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 20:32:04 +00:00
ThearchyHelios
ad1092900c
复核发现两处（C-relation-coverage-inflated、C-profile-note-rewording-bumps-catalog）： 一、覆盖率虚高。key_literals 把父列是数字、子列取到的不是整数写法的文字（'WALKIN'、'GUEST0'）从覆盖率的 分母里拿掉了：散客代号占九成的一列只剩 1 个整数写法的键对上，就测成覆盖率 100%、升为「已核实」，模型的 db_schema、自查、冻结快照都会照这条错的关系写 join。基线上同一份数据是 proposed、覆盖率 3.1%。 - 整数主键里永远没有这些文字，它们照样是子表的键值，算作没对上、留在分母里： 覆盖率 = 对上的 / (拼进 IN 的 + 按类型就对不上的)。全都不是整数写法时不发父表的核对查询，覆盖率 0%、 「可能不是这条关系」，不再整条判「无法核对」。 - key_literals 改回 KeyLiterals(literals, unmatchable, unsafe)：不能安全拼进 SQL 的值单独计数，不再用 「取到的 − 拼进的 − 不同类型的」倒推（'0012' 和 '12' 写成同一个字面量时会被错说成「类型不支持核对」）。 - 被指向列是日期时间的关系整条跳过，并且在发子表取值查询之前就跳过、不占这张表的查询预算；说明改成 「各库的时间写法和精度不同，不核对这种关系的覆盖率」——子表一侧也是时间时，以前写的「两端类型不同」不照实。 - 备注里「另有 N 个键值和被指向列的类型不同，未计入」改成「其中 N 个不是整数写法，被指向列是数字，算作没对上」。 - 报告里覆盖率不足的那句「抽样覆盖率 N，保持推断」也改成「覆盖率 N」，和备注一种说法。 二、升级后第一次重新剖析，数据没变也给每张有剖析结论的表升一个版本。剖析备注改了四处说法（抽样覆盖率→ 覆盖率、子表抽样→子表取了、（只看前→（按存储顺序只看前、按前 N 行统计→按存储顺序取前 N 行统计）， catalog.same_profile_note 没认成同一句，已发布工作流记下的目录版本对不上，之后每次正式运行都报 catalog.drift。 - same_profile_note 比较前先把这四处固定句式的旧写法换成现在的写法（_LEGACY_PROFILE_WORDING，句式带着前后文 匹配，不会误换人写的原话）；算相同时保留目录里的旧备注，旧说法等结论真变了、重写备注时才换掉。 - 起草时核对有效行过滤写在模型依据后面的那句「剖析核对：按前 N 行统计」同样只认那一句的旧写法。 - MySQL / MariaDB 上新加的排序规则说明、ENUM 列定义里没出现的取值、无效日期个数是新信息，这些备注照常 升一次版本，要写进发版说明。 测试：test_catalog_profile 加文字键指向整数主键的端到端用例（去掉修复回跑：测成 verified、覆盖率 100%）； test_profile_dialect_units 加覆盖率分母、时间键不发查询、unsafe 计数的单测；test_catalog_semantics 加跨版本 用例（把剖析备注改回旧说法写进目录、同样的数据重新剖析，版本号和目录都不变；去掉规整回跑两条失败）。
619 lines
38 KiB
Python
619 lines
38 KiB
Python
"""数据目录核心语义的回归（2026-10 完整性审查 F1）。
|
||
|
||
- A1：经对话提案确认一条关系，不抹掉数据剖析测得的基数和覆盖率；新增关系没写基数时按目标表的键推算，
|
||
推算不了就留空并在提案的这一项上说明。否则扇出检查（fanout_sum）看不到基数，静默失效。
|
||
- A2：在目录页去掉推断或外键来的关系即驳回，保留编号；只有人工新建的关系才真正删除。检查器、关系图
|
||
（助手挑表也用它）、重新起草三处都不能再出现被驳回的关系。
|
||
- A4：码值只补了一部分时不当作完整清单。「码值不在码值表中」只对标了「已列全」的码值报；「已列全」只有人工
|
||
明确确认、或者剖析没截断地看完全表取值时才有。
|
||
- B2：隔天重新剖析、数据没变，目录不升版本（剖析写的备注里只有日期变了不算变化）。
|
||
- B3：计算公式在人工整份提交、模型起草两条路径上也拦下（以前只拦提案）。
|
||
- 一致性：日期列的判定、表名的匹配、拆词各只有一份。
|
||
|
||
数据和名称全部虚构（tests/fixtures/catalog/scenic.py）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import sqlite3
|
||
|
||
import pytest
|
||
|
||
from app.data import catalog, catalog_profile, sqlcheck
|
||
from app.db.base import SessionLocal
|
||
from app.db.models import DataSource
|
||
from tests.fixtures.catalog import scenic, scenic_notes
|
||
from tests.fixtures.catalog.fake_model import FakeModel
|
||
from tests.test_catalog_patch import _schema
|
||
from tests.test_catalog_profile import _notes, _on, _profile, _relation, _table, client, make_source # noqa: F401
|
||
|
||
AT = "2026-10-04T08:00:00+00:00"
|
||
DAY1, DAY2, DAY3 = "2026-10-01T08:00:00+00:00", "2026-10-02T08:00:00+00:00", "2026-10-03T08:00:00+00:00"
|
||
|
||
|
||
async def _checker(sid: str) -> sqlcheck.SqlChecker | None:
|
||
async with SessionLocal() as session:
|
||
row = await session.get(DataSource, sid)
|
||
return (await sqlcheck.load_checkers(session, [row])).get(row.name)
|
||
|
||
|
||
async def _found(sid: str, sql: str) -> list[tuple]:
|
||
checker = await _checker(sid)
|
||
return [(c.code, c.level, c.table, c.column) for c in (checker.check(sql) if checker else [])]
|
||
|
||
|
||
async def _graph(sid: str) -> dict[str, list[catalog.JoinEdge]]:
|
||
"""助手挑表用的关系图：库里的目录原样（含驳回项）加表结构现推。"""
|
||
async with SessionLocal() as session:
|
||
row = await session.get(DataSource, sid)
|
||
entries = await catalog.read_catalog(session, sid)
|
||
return catalog.relation_graph({t: e.notes for t, e in entries.items()}, schema_cache=row.schema_cache)
|
||
|
||
|
||
async def _put(client, sid: str, table: str, notes: dict, version: int):
|
||
return await client.put(f"/api/datasources/{sid}/catalog/{table}", json={"notes": notes, "if_version": version})
|
||
|
||
|
||
def _profile_rel(table: str, columns: list[str], to_table: str, to_columns: list[str], *,
|
||
cardinality: str | None = "many_to_one", coverage: float | None = 1.0,
|
||
status: str = "verified") -> dict:
|
||
return {"id": catalog.relation_id(table, columns, to_table, to_columns), "columns": columns,
|
||
"to_table": to_table, "to_columns": to_columns, "cardinality": cardinality, "coverage": coverage,
|
||
"source": "profile", "status": status,
|
||
"note": "数据剖析（2026-10-01）：子表抽样 8 个不同键值，父表对上 8 个，覆盖率 100%；被指向列是主键。"}
|
||
|
||
|
||
# ==========================================================================
|
||
# A1：提案不抹掉基数和覆盖率
|
||
# ==========================================================================
|
||
|
||
|
||
async def test_patch_confirming_a_profiled_relation_keeps_cardinality_and_coverage(scenic_db):
|
||
tables = (await _schema(scenic_db))["tables"]
|
||
gate = _profile_rel("visits", ["gate_id"], "gates", ["id"])
|
||
notes = {"relations": [gate]}
|
||
value = {"columns": ["gate_id"], "to_table": "gates", "to_columns": ["id"]}
|
||
for path in ("relations.new", f"relations.{gate['id']}"):
|
||
[change] = catalog.plan_patch(notes, [{"path": path, "value": value}], table="visits", tables=tables).changes
|
||
assert change.after["cardinality"] == "many_to_one" and change.after["coverage"] == 1.0
|
||
# 两端、基数、覆盖率都和目录里的一样：保存即确认，来源不变
|
||
assert change.state == "confirm"
|
||
out = catalog.apply_patch(notes, [{"path": "relations.new", "value": value}], table="visits", tables=tables,
|
||
at=AT)
|
||
[rel] = out["relations"]
|
||
assert (rel["source"], rel["status"], rel["cardinality"], rel["coverage"]) == (
|
||
"profile", "confirmed", "many_to_one", 1.0)
|
||
assert rel["note"] == gate["note"]
|
||
|
||
|
||
async def test_patch_may_change_cardinality_but_never_coverage(scenic_db):
|
||
tables = (await _schema(scenic_db))["tables"]
|
||
gate = _profile_rel("visits", ["gate_id"], "gates", ["id"], coverage=0.97)
|
||
value = {"columns": ["gate_id"], "to_table": "gates", "to_columns": ["id"], "cardinality": "one_to_one",
|
||
"coverage": 0.2} # 覆盖率是测量值：模型写了也不认
|
||
[change] = catalog.plan_patch({"relations": [gate]}, [{"path": "relations.new", "value": value}],
|
||
table="visits", tables=tables).changes
|
||
assert change.state == "change"
|
||
assert (change.after["cardinality"], change.after["coverage"]) == ("one_to_one", 0.97)
|
||
out = catalog.apply_patch({"relations": [gate]}, [{"path": "relations.new", "value": value}], table="visits",
|
||
tables=tables, at=AT)
|
||
[rel] = out["relations"]
|
||
assert (rel["source"], rel["status"], rel["cardinality"], rel["coverage"]) == (
|
||
"human", "confirmed", "one_to_one", 0.97)
|
||
assert rel["note"] == gate["note"] # 覆盖率留着，说明它的剖析备注也留着
|
||
|
||
|
||
async def test_new_relation_without_cardinality_is_inferred_from_target_keys(scenic_db):
|
||
tables = (await _schema(scenic_db))["tables"]
|
||
|
||
def plan(table: str, value: dict) -> catalog.PatchChange:
|
||
[change] = catalog.plan_patch({}, [{"path": "relations.new", "value": value}], table=table,
|
||
tables=tables).changes
|
||
return change
|
||
|
||
# 目标是主键、本表这一列不唯一：多对一
|
||
gate = plan("visits", {"columns": ["gate_id"], "to_table": "gates", "to_columns": ["id"]})
|
||
assert (gate.after["cardinality"], gate.after["coverage"], gate.note) == ("many_to_one", None, None)
|
||
assert "note" not in gate.as_dict()
|
||
# 目标是唯一约束、本表这一列也唯一（票号有唯一索引）：一对一
|
||
ticket = plan("visits", {"columns": ["ticket_no"], "to_table": "orders", "to_columns": ["order_no"]})
|
||
assert ticket.after["cardinality"] == "one_to_one"
|
||
# 目标列不是主键也不是唯一约束：推算不了，基数留空，这一项上说明
|
||
park = plan("visits", {"columns": ["park_id"], "to_table": "gates", "to_columns": ["park_id"]})
|
||
assert park.after["cardinality"] is None
|
||
assert park.note and "基数" in park.note and park.as_dict()["note"] == park.note
|
||
# 不知道表结构（尚未探查）：不推算，也说明
|
||
[blind] = catalog.plan_patch({}, [{"path": "relations.new", "value": {
|
||
"columns": ["gate_id"], "to_table": "gates", "to_columns": ["id"]}}], table="visits", tables=None).changes
|
||
assert blind.after["cardinality"] is None and blind.note
|
||
|
||
|
||
async def test_confirming_a_profiled_relation_by_patch_keeps_the_fanout_check(client, make_source, scenic_db):
|
||
"""审查探针 A1：剖析得出多对一、覆盖率 1.0，经对话提案确认这条关系之后，扇出检查照样报 error。"""
|
||
sql = "SELECT COUNT(g.id) AS n FROM gates g JOIN visits v ON v.gate_id = g.id"
|
||
sid = await make_source(scenic_db, options=_on(), draft=["visits", "gates"])
|
||
await _profile(client, sid, ["visits"])
|
||
detail = await _notes(client, sid, "visits")
|
||
rel = _relation(detail["notes"], "gate_id")
|
||
assert (rel["source"], rel["status"], rel["cardinality"], rel["coverage"]) == (
|
||
"profile", "verified", "many_to_one", 1.0)
|
||
# 子表一侧数过、确有重复：基数用数据核实过，一对多确有其事 → error（A6）
|
||
assert rel["cardinality_checked"] is True
|
||
assert ("fanout_sum", "error", "gates", "id") in await _found(sid, sql)
|
||
|
||
changes = [{"path": "relations.new", "value": {"columns": ["gate_id"], "to_table": "gates", "to_columns": ["id"]},
|
||
"reason": "用户说 gate_id 对应 gates.id"}]
|
||
r = await client.post(f"/api/datasources/{sid}/catalog/visits/patch",
|
||
json={"changes": changes, "if_version": detail["version"]})
|
||
assert r.status_code == 200, r.text
|
||
after = _relation(r.json()["notes"], "gate_id")
|
||
assert (after["status"], after["cardinality"], after["coverage"]) == ("confirmed", "many_to_one", 1.0)
|
||
assert ("fanout_sum", "error", "gates", "id") in await _found(sid, sql)
|
||
|
||
|
||
async def test_profiled_relation_without_child_check_only_warns(client, make_source, scenic_db):
|
||
"""A6：子表超过整表统计上限、没核对子表一侧唯不唯一的关系，剖析照样记有确证、按多对一记，但基数不算用数据
|
||
核实过：扇出检查只给 warning（「可能重复计算」），不给 error、不让出具降档。"""
|
||
sql = "SELECT COUNT(g.id) AS n FROM gates g JOIN visits v ON v.gate_id = g.id"
|
||
sid = await make_source(scenic_db, options=_on(max_scan_rows=100), draft=["visits", "gates"])
|
||
await _profile(client, sid, ["visits"])
|
||
rel = _relation((await _notes(client, sid, "visits"))["notes"], "gate_id")
|
||
assert (rel["source"], rel["status"], rel["cardinality"]) == ("profile", "verified", "many_to_one")
|
||
assert not rel.get("cardinality_checked")
|
||
found = await _found(sid, sql)
|
||
assert ("fanout_sum", "warning", "gates", "id") in found
|
||
assert ("fanout_sum", "error", "gates", "id") not in found
|
||
|
||
|
||
# ==========================================================================
|
||
# A2：去掉推断或外键来的关系即驳回
|
||
# ==========================================================================
|
||
|
||
|
||
def _rel(columns, to_table, to_columns, source="name", status=None, table="visits"):
|
||
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 or catalog.initial_status(source)}
|
||
|
||
|
||
def test_human_edit_rejects_removed_inferred_relations_and_deletes_human_ones():
|
||
park, gate = _rel(["park_id"], "parks", ["id"], "fk"), _rel(["gate_id"], "gates", ["id"], "name")
|
||
mine = _rel(["member_id"], "members", ["id"], "human")
|
||
out = catalog.apply_human_edit({"relations": [park, gate, mine]}, {"relations": []}, table="visits", at=AT)
|
||
by_id = {r["id"]: r for r in out["relations"]}
|
||
assert set(by_id) == {park["id"], gate["id"]} # 人工新建的那条真正删掉
|
||
for old in (park, gate):
|
||
assert by_id[old["id"]] == {**old, "status": "rejected", "updated_at": AT} # 编号、来源、两端都留着
|
||
# 已经驳回的再去掉一次：不变
|
||
again = catalog.apply_human_edit(out, {"relations": []}, table="visits", at="2026-10-05T00:00:00+00:00")
|
||
assert again == out
|
||
|
||
|
||
def test_human_edit_moving_an_inferred_relation_rejects_the_old_one():
|
||
gate = _rel(["gate_id"], "gates", ["id"], "name")
|
||
moved = {**gate, "to_table": "parks", "to_columns": ["id"]}
|
||
out = catalog.apply_human_edit({"relations": [gate]}, {"relations": [moved]}, table="visits", at=AT)
|
||
new_id = catalog.relation_id("visits", ["gate_id"], "parks", ["id"])
|
||
by_id = {r["id"]: r for r in out["relations"]}
|
||
assert (by_id[new_id]["source"], by_id[new_id]["status"]) == ("human", "confirmed")
|
||
assert by_id[gate["id"]]["status"] == "rejected" and by_id[gate["id"]]["source"] == "name"
|
||
|
||
|
||
def test_human_edit_rejects_a_removed_human_relation_that_structure_would_bring_back():
|
||
"""人工改过基数的外键关系记成了人工填写；去掉它时若表结构还推得出来，删掉就会被现推回来，只能驳回。"""
|
||
park = {**_rel(["park_id"], "parks", ["id"], "fk"), "cardinality": "one_to_one", "source": "human",
|
||
"status": "confirmed"}
|
||
out = catalog.apply_human_edit({"relations": [park]}, {"relations": []}, table="visits", at=AT,
|
||
inferable={park["id"]})
|
||
assert [(r["id"], r["status"]) for r in out["relations"]] == [(park["id"], "rejected")]
|
||
assert catalog.apply_human_edit({"relations": [park]}, {"relations": []}, table="visits", at=AT) == {}
|
||
|
||
|
||
def test_human_edit_cannot_forge_a_data_checked_cardinality():
|
||
"""A6：「基数用数据核实过」只能由数据剖析写。人工改了关系（记成人工确认），提交里带的 cardinality_checked 不作数；
|
||
值没动的关系保留剖析原来写的。"""
|
||
gate = {**_rel(["gate_id"], "gates", ["id"], "profile", "verified"), "coverage": 1.0, "cardinality_checked": True}
|
||
park = _rel(["park_id"], "parks", ["id"], "name")
|
||
forged = {**park, "cardinality": "one_to_many", "cardinality_checked": True}
|
||
out = catalog.apply_human_edit({"relations": [gate, park]}, {"relations": [gate, forged]}, table="visits", at=AT)
|
||
by_id = {r["id"]: r for r in out["relations"]}
|
||
assert by_id[gate["id"]]["cardinality_checked"] is True
|
||
assert (by_id[park["id"]]["source"], by_id[park["id"]]["status"]) == ("human", "confirmed")
|
||
assert "cardinality_checked" not in by_id[park["id"]]
|
||
|
||
|
||
async def test_removed_or_rejected_relations_stay_out_of_checks_graph_and_redraft(client, make_source, scenic_db):
|
||
sid = await make_source(scenic_db, draft=["visits"])
|
||
detail = await _notes(client, sid, "visits")
|
||
park, gate = _relation(detail["notes"], "park_id"), _relation(detail["notes"], "gate_id")
|
||
ticket = _relation(detail["notes"], "ticket_type_id")
|
||
assert (park["source"], gate["source"], ticket["source"]) == ("fk", "name", "fk")
|
||
gone = {park["id"], gate["id"]}
|
||
|
||
# 目录页编辑时去掉两条（外键来的、命名推断的）：转为驳回
|
||
notes = {**detail["notes"], "relations": [r for r in detail["notes"]["relations"] if r["id"] not in gone]}
|
||
r = await _put(client, sid, "visits", notes, detail["version"])
|
||
assert r.status_code == 200, r.text
|
||
saved = {x["id"]: x for x in r.json()["notes"]["relations"]}
|
||
assert {saved[i]["status"] for i in gone} == {"rejected"}
|
||
# 单项审阅里驳回第三条（界面上非人工来源的关系只给「驳回」）
|
||
r = await client.post(f"/api/datasources/{sid}/catalog/visits/review",
|
||
json={"path": f"relations.{ticket['id']}", "action": "reject",
|
||
"if_version": r.json()["version"]})
|
||
assert r.status_code == 200, r.text
|
||
gone.add(ticket["id"])
|
||
|
||
def absent(edges) -> bool:
|
||
return not any(e.relation_id in gone for e in edges)
|
||
|
||
# 检查器：被驳回的关系不当作「对上了」，也不拿来判断基数
|
||
checker = await _checker(sid)
|
||
for target in ("parks", "gates", "ticket_types"):
|
||
assert checker.edges_between("visits", target) == []
|
||
joined = await _found(sid, "SELECT COUNT(*) FROM visits v JOIN gates g ON v.gate_id = g.id")
|
||
assert ("join_unconfirmed", "warning", "gates", None) in joined
|
||
# 关系图（助手挑表也用它）：不按表结构现推回来
|
||
graph = await _graph(sid)
|
||
assert absent(graph.get("visits", [])) and absent(graph.get("parks", [])) and absent(graph.get("gates", []))
|
||
# 重新起草：仍是驳回
|
||
r = await client.post(f"/api/datasources/{sid}/catalog/draft", json={"tables": ["visits"]})
|
||
assert r.status_code == 200, r.text
|
||
again = {x["id"]: x for x in (await _notes(client, sid, "visits"))["notes"]["relations"]}
|
||
assert {again[i]["status"] for i in gone} == {"rejected"}
|
||
assert absent((await _graph(sid)).get("visits", []))
|
||
|
||
|
||
def test_relation_graph_keeps_structure_relations_of_partially_annotated_tables():
|
||
"""只做过部分注释的表（目录里没有关系）照样按结构补上外键和命名推断；这张表目录里驳回的不补。"""
|
||
cache = {"tables": {
|
||
"visits": {"columns": [{"name": "id", "type": "INTEGER"}, {"name": "park_id", "type": "INTEGER"},
|
||
{"name": "gate_id", "type": "INTEGER"}],
|
||
"primary_key": ["id"], "foreign_keys": [{"columns": ["park_id"], "to_table": "parks",
|
||
"to_columns": ["id"]}]},
|
||
"parks": {"columns": [{"name": "id", "type": "INTEGER"}], "primary_key": ["id"], "foreign_keys": []},
|
||
"gates": {"columns": [{"name": "id", "type": "INTEGER"}], "primary_key": ["id"], "foreign_keys": []},
|
||
}}
|
||
labelled = {"visits": {"label": catalog.make_item("入园记录", "human")}}
|
||
edges = {(e.to_table, e.source) for e in catalog.relation_graph(labelled, schema_cache=cache)["visits"]}
|
||
assert edges == {("parks", "fk"), ("gates", "name")}
|
||
rejected = {"visits": {"relations": [_rel(["park_id"], "parks", ["id"], "fk", "rejected")]}}
|
||
edges = {e.to_table for e in catalog.relation_graph(rejected, schema_cache=cache)["visits"]}
|
||
assert edges == {"gates"}
|
||
|
||
|
||
# ==========================================================================
|
||
# A4：码值「已列全」
|
||
# ==========================================================================
|
||
|
||
|
||
def _codes(value: dict, *, complete: bool | None, status: str = "confirmed") -> dict:
|
||
item = catalog.make_item(value, "human" if status == "confirmed" else "profile", status)
|
||
if complete is not None:
|
||
item["complete"] = complete
|
||
return item
|
||
|
||
|
||
async def test_unknown_code_only_for_codes_marked_complete(scenic_db):
|
||
cache = await _schema(scenic_db)
|
||
sql = "SELECT COUNT(*) FROM visits WHERE status = 3"
|
||
|
||
def found(item: dict) -> list[str]:
|
||
notes = scenic_notes.notes()
|
||
notes["visits"]["columns"]["status"]["codes"] = item
|
||
checker = sqlcheck.SqlChecker(kind="sqlite", schema_cache=cache, notes=notes, source_name="scenic")
|
||
return [c.code for c in checker.check(sql)]
|
||
|
||
assert "unknown_code" not in found(_codes({"1": "有效", "0": "作废"}, complete=None))
|
||
assert "unknown_code" in found(_codes({"1": "有效", "0": "作废"}, complete=True))
|
||
|
||
|
||
async def test_partial_codes_from_a_patch_are_not_treated_as_the_full_list(client, make_source, scenic_db):
|
||
"""审查探针 A4：经提案只写了 status=9 表示作废，之后 status = 1 不该被报「码值不在码值表中」。"""
|
||
sid = await make_source(scenic_db, draft=["orders"])
|
||
detail = await _notes(client, sid, "orders")
|
||
r = await client.post(f"/api/datasources/{sid}/catalog/orders/patch", json={
|
||
"changes": [{"path": "columns.status.codes", "value": {"9": "作废"}}], "if_version": detail["version"]})
|
||
assert r.status_code == 200, r.text
|
||
codes = r.json()["notes"]["columns"]["status"]["codes"]
|
||
assert codes["value"] == {"9": "作废"} and "complete" not in codes
|
||
sql = "SELECT COUNT(*) FROM orders WHERE status = 1"
|
||
assert not any(c[0] == "unknown_code" for c in await _found(sid, sql))
|
||
|
||
# 人工在目录页把码值补全，并勾上「已列出全部取值」：这时才报
|
||
notes = r.json()["notes"]
|
||
notes["columns"]["status"]["codes"] = {**codes, "value": {"1": "已支付", "2": "已退款", "9": "作废"},
|
||
"complete": True}
|
||
r = await _put(client, sid, "orders", notes, r.json()["version"])
|
||
assert r.status_code == 200, r.text
|
||
saved = r.json()["notes"]["columns"]["status"]["codes"]
|
||
assert (saved["source"], saved["status"], saved.get("complete")) == ("human", "confirmed", True)
|
||
assert not any(c[0] == "unknown_code" for c in await _found(sid, sql))
|
||
checker = await _checker(sid)
|
||
[check] = [c for c in checker.check("SELECT COUNT(*) FROM orders WHERE status = 5") if c.code == "unknown_code"]
|
||
assert check.level == "warning"
|
||
# 给模型的话：提示核对取值，不叫它「照目录里的码值改」
|
||
assert "照目录里的码值改" not in check.for_model and "核对" in check.for_model
|
||
|
||
|
||
def test_codes_complete_flag_is_validated_and_counts_as_an_edit():
|
||
base = {"columns": {"status": {"codes": _codes({"1": "", "0": ""}, complete=None, status="proposed")}}}
|
||
assert catalog.validate_notes(base) == []
|
||
assert catalog.validate_notes({"columns": {"status": {"codes": _codes({"1": "有效"}, complete=True)}}}) == []
|
||
bad = catalog.validate_notes({"columns": {"status": {"codes": {**_codes({"1": "有效"}, complete=None),
|
||
"complete": "是"}}}})
|
||
assert bad and "码值" in bad[0]
|
||
wrong_place = catalog.validate_notes({"label": {**catalog.make_item("入园记录", "human"), "complete": True}})
|
||
assert wrong_place and "complete" in wrong_place[0]
|
||
|
||
# 值没动、只勾上「已列全」：算人工改动，记为人工填写、已确认
|
||
ticked = {"columns": {"status": {"codes": {**base["columns"]["status"]["codes"], "complete": True}}}}
|
||
out = catalog.apply_human_edit(base, ticked, table="visits", at=AT)
|
||
codes = out["columns"]["status"]["codes"]
|
||
assert (codes["source"], codes["status"], codes["complete"]) == ("human", "confirmed", True)
|
||
# 取消勾选：不留 complete: false，和没有这个标记写法一样
|
||
unticked = {"columns": {"status": {"codes": {**codes, "complete": False}}}}
|
||
out = catalog.apply_human_edit(out, unticked, table="visits", at=AT)
|
||
assert "complete" not in out["columns"]["status"]["codes"]
|
||
|
||
|
||
async def test_profile_marks_codes_complete_only_when_the_whole_table_was_read(client, make_source, scenic_db):
|
||
whole = await make_source(scenic_db, options=_on(), draft=["visits"])
|
||
await _profile(client, whole, ["visits"])
|
||
codes = (await _notes(client, whole, "visits"))["notes"]["columns"]["status"]["codes"]
|
||
assert "统计全表" in codes["note"] and codes.get("complete") is True
|
||
|
||
# 不做整表统计（只看前若干行）：取值可能没看全，不标
|
||
capped = await make_source(scenic_db, options=_on(max_scan_rows=0, sample_size=10), draft=["visits"])
|
||
await _profile(client, capped, ["visits"])
|
||
codes = (await _notes(client, capped, "visits"))["notes"]["columns"]["status"]["codes"]
|
||
assert "按存储顺序取前 10 行统计" in codes["note"] and "complete" not in codes
|
||
|
||
|
||
# ==========================================================================
|
||
# B2：隔天重新剖析、数据没变，不升版本
|
||
# ==========================================================================
|
||
|
||
|
||
async def test_rerun_on_another_day_with_same_data_keeps_the_version(client, make_source, tmp_path, monkeypatch):
|
||
path = scenic.build(tmp_path / "b2.db")
|
||
sid = await make_source(path, options=_on(), draft=["visits"])
|
||
# 人工确认过的关系：剖析只补覆盖率和基数、换自己的备注，同样不能因为日期变了就改
|
||
detail = await _notes(client, sid, "visits")
|
||
gate = _relation(detail["notes"], "gate_id")
|
||
r = await client.post(f"/api/datasources/{sid}/catalog/visits/review",
|
||
json={"path": f"relations.{gate['id']}", "action": "confirm", "if_version": detail["version"]})
|
||
assert r.status_code == 200, r.text
|
||
|
||
monkeypatch.setattr(catalog, "now_iso", lambda: DAY1)
|
||
first = _table(await _profile(client, sid, ["visits"]), "visits")
|
||
notes_day1 = (await _notes(client, sid, "visits"))["notes"]
|
||
monkeypatch.setattr(catalog, "now_iso", lambda: DAY2)
|
||
second = _table(await _profile(client, sid, ["visits"]), "visits")
|
||
assert (second["added"], second["updated"], second["removed"]) == (0, 0, 0)
|
||
assert second["version"] == first["version"]
|
||
assert (await _notes(client, sid, "visits"))["notes"] == notes_day1 # 备注里还是第一天的日期
|
||
|
||
# 数据变了（多了一种状态），结论跟着变：照常升版本
|
||
db = sqlite3.connect(path)
|
||
try:
|
||
db.execute("INSERT INTO visits VALUES (1501, 'TK9001501', 1, 1, 1, NULL, '2026-09-30 10:00:00', 1, 2)")
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
monkeypatch.setattr(catalog, "now_iso", lambda: DAY3)
|
||
third = _table(await _profile(client, sid, ["visits"]), "visits")
|
||
assert third["version"] == second["version"] + 1 and third["updated"] >= 1
|
||
codes = (await _notes(client, sid, "visits"))["notes"]["columns"]["status"]["codes"]
|
||
assert set(codes["value"]) == {"0", "1", "2"} and codes["updated_at"] == DAY3
|
||
# A7：说明里不写日期（以前写的是 UTC 日期，和界面上的本地时间对不上），时间以这一项的 updated_at 为准
|
||
assert codes["note"].startswith("数据剖析：") and "2026-10-03" not in codes["note"]
|
||
|
||
|
||
#: 升级前剖析写的说法 → 现在的说法（下一轮 C1：大表看的是按存储顺序的前 N 行，不是抽样）。测试里反过来用：把新写法
|
||
#: 改回旧写法，模拟升级前写进目录的备注
|
||
_WORDING = (("覆盖率 ", "抽样覆盖率 ", "，可能不是这条关系"), ("子表取了 ", "子表抽样 ", None),
|
||
("（按存储顺序只看前 ", "（只看前 ", None), ("按存储顺序取前 ", "按前 ", None))
|
||
|
||
|
||
def _legacy(note: str) -> str:
|
||
out = note
|
||
for new, old, tail in _WORDING:
|
||
if tail:
|
||
out = re.sub(re.escape(new) + r"(\S+?)" + re.escape(tail), lambda m: old + m.group(1) + tail, out)
|
||
else:
|
||
out = out.replace(new, old)
|
||
return out
|
||
|
||
|
||
def _profile_notes(notes: dict) -> list[dict]:
|
||
"""目录里剖析写了备注的项（关系、码值、业务日期）。"""
|
||
found = [r for r in notes.get("relations") or [] if catalog.is_profile_note(r.get("note"))]
|
||
for column in (notes.get("columns") or {}).values():
|
||
found += [item for item in column.values() if isinstance(item, dict)
|
||
and catalog.is_profile_note(item.get("note"))]
|
||
bdate = notes.get("business_date")
|
||
return found + ([bdate] if isinstance(bdate, dict) and catalog.is_profile_note(bdate.get("note")) else [])
|
||
|
||
|
||
def test_profile_notes_in_the_old_wording_are_the_same_note():
|
||
"""改过的四处说法：旧写法和新写法一字之外都一样才算同一句；数变了、或者不是剖析写的备注，照旧算不同。"""
|
||
old = "数据剖析：抽样覆盖率 15%，可能不是这条关系；子表抽样 20 个不同键值（只看前 100 行），父表对上 3 个；被指向列是主键。"
|
||
new = "数据剖析：覆盖率 15%，可能不是这条关系；子表取了 20 个不同键值（按存储顺序只看前 100 行），父表对上 3 个；被指向列是主键。"
|
||
assert catalog.same_profile_note(old, new) and catalog.same_profile_note(new, old)
|
||
assert catalog.same_profile_note(old.replace("数据剖析：", "数据剖析（2026-10-01）："), new)
|
||
assert not catalog.same_profile_note(old, new.replace("对上 3 个", "对上 4 个"))
|
||
codes_old = "数据剖析（2026-10-01）：按前 10 行统计，10 行非空值共 2 个取值：1（90%）、0（10%）。"
|
||
codes_new = "数据剖析：按存储顺序取前 10 行统计，10 行非空值共 2 个取值：1（90%）、0（10%）。"
|
||
assert catalog.same_profile_note(codes_old, codes_new)
|
||
assert not catalog.same_profile_note(codes_old, codes_new.replace("前 10 行", "前 20 行"))
|
||
# 起草写的有效行过滤备注：只认接在后面的剖析核对那句
|
||
filter_old = "依据：status 为 1 才是有效记录。剖析核对：按前 100,000 行统计，排除 3% 的行"
|
||
filter_new = "依据：status 为 1 才是有效记录。剖析核对：按存储顺序取前 100,000 行统计，排除 3% 的行"
|
||
assert catalog.same_profile_note(filter_old, filter_new)
|
||
assert not catalog.same_profile_note(filter_old, filter_new.replace("3%", "4%"))
|
||
# 不是剖析写的备注：「抽样覆盖率」是人写的原话，不换
|
||
assert not catalog.same_profile_note("人工备注：抽样覆盖率 15%，可能不是这条关系",
|
||
"人工备注：覆盖率 15%，可能不是这条关系")
|
||
assert not catalog.same_profile_note(new, "覆盖率 15%，可能不是这条关系")
|
||
|
||
|
||
async def test_rerun_after_upgrade_keeps_the_version_when_only_the_wording_changed(client, make_source, tmp_path):
|
||
"""升级前写进目录的剖析备注用的是旧说法（「抽样覆盖率」「子表抽样」「（只看前 N 行）」「按前 N 行统计」）。数据
|
||
没变、升级后第一次重新剖析：结论一字不差、只是措辞改了，不算变化——否则每张有剖析结论的表都升一个版本，发布时
|
||
记下的目录版本对不上，之后每次正式运行都提醒「目录有变化」。"""
|
||
path = scenic.build(tmp_path / "wording.db")
|
||
db = sqlite3.connect(path)
|
||
try:
|
||
db.execute("CREATE TABLE lots (id INTEGER PRIMARY KEY, name TEXT NOT NULL)")
|
||
db.executemany("INSERT INTO lots VALUES (?, ?)", [(i, f"{i} 号停车区") for i in (1, 2, 3)])
|
||
db.executemany("INSERT INTO parking_records VALUES (?, ?, ?, ?, ?, ?)",
|
||
[(i, 1 + i % 20, None, "2026-08-01 09:00:00", None, 10.0) for i in range(1, 101)])
|
||
db.commit()
|
||
finally:
|
||
db.close()
|
||
tables = ["visits", "parking_records"]
|
||
whole = await make_source(path, options=_on(), draft=tables)
|
||
capped = await make_source(path, options=_on(max_scan_rows=0, sample_size=10), draft=tables)
|
||
seen: set[str] = set()
|
||
for sid in (whole, capped):
|
||
await _profile(client, sid, tables)
|
||
for table in tables:
|
||
detail = await _notes(client, sid, table)
|
||
notes = detail["notes"]
|
||
for item in _profile_notes(notes):
|
||
old = _legacy(item["note"])
|
||
seen |= {o for _, o, _ in _WORDING if o in old}
|
||
item["note"] = old
|
||
async with SessionLocal() as session:
|
||
await catalog.write_entry(session, sid, table, notes, if_version=detail["version"], actor=None)
|
||
assert seen == {o for _, o, _ in _WORDING}, "四处旧说法都写进了目录"
|
||
|
||
for sid in (whole, capped):
|
||
before = {t: await _notes(client, sid, t) for t in tables}
|
||
body = await _profile(client, sid, tables)
|
||
for table in tables:
|
||
done = _table(body, table)
|
||
assert (done["added"], done["updated"], done["removed"]) == (0, 0, 0), (table, done)
|
||
assert done["version"] == before[table]["version"]
|
||
assert (await _notes(client, sid, table))["notes"] == before[table]["notes"]
|
||
|
||
|
||
def test_merge_ignores_a_profile_note_that_only_changed_its_date():
|
||
old = {"columns": {"status": {"codes": catalog.make_item(
|
||
{"1": "", "0": ""}, "profile", "proposed", note="数据剖析（2026-10-01）：统计全表，1500 行非空值共 2 个取值。",
|
||
at=DAY1)}}}
|
||
same = {"columns": {"status": {"codes": catalog.make_item(
|
||
{"1": "", "0": ""}, "profile", "proposed", note="数据剖析（2026-10-02）：统计全表，1500 行非空值共 2 个取值。")}}}
|
||
merged, stats = catalog.merge_notes(old, same, covered={"profile"}, at=DAY2)
|
||
assert merged == old and stats == catalog.MergeStats()
|
||
changed = {"columns": {"status": {"codes": catalog.make_item(
|
||
{"1": "", "0": ""}, "profile", "proposed", note="数据剖析（2026-10-02）：统计全表，1600 行非空值共 2 个取值。")}}}
|
||
merged, stats = catalog.merge_notes(old, changed, covered={"profile"}, at=DAY2)
|
||
assert stats.updated == 1 and "2026-10-02" in merged["columns"]["status"]["codes"]["note"]
|
||
|
||
|
||
def test_profile_notes_without_date_stay_compatible_with_dated_ones():
|
||
"""A7：剖析写的说明不再带日期（「数据剖析：…」）。升级前写的「数据剖析（2026-10-01）：…」照样认得是剖析写的；
|
||
结论一样、只是新的不带日期，不算变化——升级后第一次重新剖析不能因此给每张表升一个版本。"""
|
||
dated = "数据剖析（2026-10-01）：统计全表，1500 行非空值共 2 个取值。"
|
||
plain = "数据剖析：统计全表，1500 行非空值共 2 个取值。"
|
||
assert catalog.is_profile_note(dated) and catalog.is_profile_note(plain)
|
||
assert not catalog.is_profile_note("人工备注：数据剖析（2026-10-01）") and not catalog.is_profile_note(None)
|
||
assert catalog.same_profile_note(dated, plain) and catalog.same_profile_note(plain, dated)
|
||
assert catalog.same_profile_note("数据剖析（2026-10-02）：统计全表，1500 行非空值共 2 个取值。", dated)
|
||
assert not catalog.same_profile_note(dated, "数据剖析：统计全表，1600 行非空值共 2 个取值。")
|
||
assert not catalog.same_profile_note(plain, "统计全表，1500 行非空值共 2 个取值。")
|
||
old = {"columns": {"status": {"codes": catalog.make_item({"1": "", "0": ""}, "profile", "proposed", note=dated,
|
||
at=DAY1)}}}
|
||
new = {"columns": {"status": {"codes": catalog.make_item({"1": "", "0": ""}, "profile", "proposed", note=plain)}}}
|
||
merged, stats = catalog.merge_notes(old, new, covered={"profile"}, at=DAY2)
|
||
assert merged == old and stats == catalog.MergeStats()
|
||
|
||
|
||
# ==========================================================================
|
||
# B3：公式在人工编辑、模型起草两条路径上也拦下
|
||
# ==========================================================================
|
||
|
||
|
||
def test_human_edit_rejects_formulas_in_changed_text():
|
||
existing = {"columns": {"amount": {"label": catalog.make_item("实收金额", "llm")}}}
|
||
for submitted in (
|
||
{"columns": {"amount": {"label": catalog.make_item("实收金额", "llm"),
|
||
"meaning": {"value": "客单价=金额/人数"}}}},
|
||
{"description": {"value": "转化率 = 下单数 / 访问数"}},
|
||
{"columns": {"amount": {"label": {"value": "SUM(amount)"}}}},
|
||
):
|
||
with pytest.raises(catalog.CatalogInvalid) as e:
|
||
catalog.apply_human_edit(existing, submitted, table="orders", at=AT)
|
||
assert "口径卡" in str(e.value)
|
||
# 库里早就有的含公式的项，值没动就照常提交：不能因为旧数据挡住这张表的任何写入
|
||
legacy = {"columns": {"amount": {"meaning": catalog.make_item("客单价=金额/人数", "llm")}}}
|
||
out = catalog.apply_human_edit(legacy, {**legacy, "label": {"value": "订单"}}, table="orders", at=AT)
|
||
assert out["columns"]["amount"]["meaning"]["value"] == "客单价=金额/人数"
|
||
assert catalog.validate_notes(legacy) == [] # 结构校验不查公式
|
||
|
||
|
||
async def test_put_with_a_formula_is_422_and_says_where_formulas_go(client, make_source, scenic_db):
|
||
sid = await make_source(scenic_db, draft=["orders"])
|
||
detail = await _notes(client, sid, "orders")
|
||
notes = {**detail["notes"], "columns": {"total_amount": {"meaning": {"value": "客单价=金额/人数"}}}}
|
||
r = await _put(client, sid, "orders", notes, detail["version"])
|
||
assert r.status_code == 422 and "口径卡" in r.json()["detail"]
|
||
assert (await _notes(client, sid, "orders"))["version"] == detail["version"]
|
||
|
||
|
||
async def test_model_draft_drops_formula_items():
|
||
from types import SimpleNamespace
|
||
|
||
meta = {"qualified": "orders", "columns": [{"name": "id", "type": "INTEGER"},
|
||
{"name": "total_amount", "type": "REAL"}],
|
||
"primary_key": ["id"], "foreign_keys": []}
|
||
source = SimpleNamespace(id="src", name="shop", kind="sqlite", origin="manual",
|
||
schema_cache={"tables": {"orders": meta}})
|
||
model = FakeModel({"orders": {"label": "客单价=金额/人数", "grain": "每笔订单一行", "columns": [
|
||
{"name": "total_amount", "label": "订单金额", "meaning": "= SUM(明细金额)", "unit": "元"}]}})
|
||
notes = (await catalog.draft_with_model(model, source, ["orders"]))["orders"].notes
|
||
assert "label" not in notes and notes["grain"]["value"] == "每笔订单一行"
|
||
assert set(notes["columns"]["total_amount"]) == {"label", "unit"}
|
||
|
||
|
||
# ==========================================================================
|
||
# 一致性
|
||
# ==========================================================================
|
||
|
||
|
||
@pytest.mark.parametrize("name, type_, expected", [
|
||
("visit_time", "TEXT", True), ("ordered_at", "TEXT", True), ("joined_on", "VARCHAR(20)", True),
|
||
("createdAt", "", True), ("biz_date", "DATE", True), ("paid", "DATETIME", True), ("loaded", "TIMESTAMP", True),
|
||
("dt", "TEXT", True), ("date", "TEXT", True),
|
||
("created_at", "INTEGER", False), # 整数存的时间戳：取值不是日期，剖析和检查都不当日期列
|
||
("open_time", "TIME", False), # 只有时刻、没有日期
|
||
("on_hand", "INTEGER", False), ("status", "TEXT", False), ("at", "TEXT", False), ("ticket_no", "TEXT", False),
|
||
])
|
||
def test_date_column_rule_is_shared(name, type_, expected):
|
||
assert catalog.is_date_column(name, type_) is expected
|
||
# 剖析认的日期列（业务日期提议）和 SQL 检查认的时间列（按别的时间列统计）是同一套
|
||
assert catalog_profile._is_date_column({"name": name, "type": type_}) is expected
|
||
table = sqlcheck._Table(alias="t", name="t", meta={"columns": [{"name": name, "type": type_}]}, notes={},
|
||
order=0)
|
||
assert table.temporal(name) is expected
|
||
|
||
|
||
def test_table_names_match_by_name_key_everywhere():
|
||
"""表名只把 ASCII 字母转小写（name_key，和 SQLite 比较标识符一致）：非 ASCII 的大小写不同就是两张表。"""
|
||
cache = {"tables": {"CAFÉ_ORDERS": {"columns": []}, "Visits": {"columns": []}}}
|
||
assert catalog.resolve_table_name(cache, "visits") == "Visits"
|
||
assert catalog.resolve_table_name(cache, "main.VISITS") == "Visits"
|
||
assert catalog.resolve_table_name(cache, "cafÉ_orders") == "CAFÉ_ORDERS"
|
||
assert catalog.resolve_table_name(cache, "café_orders") is None
|
||
checker = sqlcheck.SqlChecker(kind="sqlite", schema_cache=cache, notes={"CAFÉ_ORDERS": {
|
||
"label": catalog.make_item("咖啡订单", "human")}})
|
||
assert checker.notes_of("cafÉ_orders")["label"]["value"] == "咖啡订单"
|
||
assert checker.notes_of("café_orders") == {}
|
||
|
||
|
||
def test_word_splitting_has_one_implementation():
|
||
assert catalog.name_words("memberTagId") == ["member", "tag", "id"]
|
||
assert catalog.name_words("MEMBER_TAG") == ["member", "tag"]
|
||
assert not hasattr(catalog_profile, "_words") and not hasattr(catalog_profile, "_WORDS")
|