From 4ba4a21b2c784c17bc90d303fdf3b2501192cb03 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 14:26:04 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=A7=E5=BA=93=E7=9A=84=20look=20=E9=A2=84?= =?UTF-8?q?=E7=AD=9B=E4=B8=8E=E6=8C=89=E9=9C=80=E5=8A=A0=E5=85=A5=E3=80=81?= =?UTF-8?q?=E5=89=96=E6=9E=90=E4=BA=8B=E5=AE=9E=E5=9B=9B=E5=A4=84=E4=B8=80?= =?UTF-8?q?=E8=87=B4=E3=80=81=E6=9C=89=E6=95=88=E8=AE=B0=E5=BD=95=E6=9D=A1?= =?UTF-8?q?=E4=BB=B6=E6=8C=89=20quote=5Fname=20=E5=8A=A0=E5=BC=95=E5=8F=B7?= =?UTF-8?q?=E7=9A=84=E6=B5=8B=E8=AF=95=EF=BC=9B=E5=AF=B9=E8=B1=A1=E7=B4=A2?= =?UTF-8?q?=E5=BC=95=E6=89=B9=E9=87=8F=E6=9B=B4=E6=96=B0=E6=94=B9=E8=B5=B0?= =?UTF-8?q?=E6=8C=89=E8=A1=A8=E7=9A=84=20executemany?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 为什么：A2 和遗留 e 的行为要有脚本化的验收，且能证明测试抓得住回归（去掉遮罩重滤、去掉 quote_name 强制引号后对应用例都失败， 已回跑确认）。 - tests/test_look_large.py：3,000 张表的合成库，探查范围 300 张 + 5 张有目录的表，其余只在对象索引里（注释写进索引，模拟 MySQL / PG 的表注释）。10 条脚本化需求挑中 ≥ 9 条、挑选提示词 < 20,000 字（挑选用按字面重合打分的假模型，只看得到交给它的 那份目录）；挑中范围外的表按需加入、记 added_to_scope 和留痕；点名的每次查看最多 4 张、超了照实说；db_schema__ 的 keyword、 只列前 50 张、每次运行最多 8 张；查询照实列「未按目录检查的表」；目录搜索和「加入范围」；剖析的关系指向范围外时不说 「表结构里没有」。 - tests/test_profile_facts.py：look、运行时 db_schema、/schema、表助手四处同一段剖析事实和固定提示；剖析之后加遮罩，那一列 的范围不出现；没有剖析结果时表详情一字不差；上传表格快照不一致（含没记快照的）不给；剖析任务开始一张表时记下快照； 「上次剖析」取真正开始过的那一次。 - tests/test_valid_filter_quote.py：存进目录的写法和剖析核对查询都按 quote_name 给 Oracle 小写列加引号。 - source_objects.refresh_index：注释、类型、行数估算有变的行改用按表的 executemany（ORM 的按主键批量更新要求参数里写主键 列名，3,000 张表的库第二次刷新索引时报错）。 - db_schema__ 不传 table 时说清没列出的里有几张在探查范围外。 --- backend/app/data/source_objects.py | 10 +- backend/app/tools/datasource.py | 7 +- backend/tests/test_look_large.py | 238 +++++++++++++++++++++++ backend/tests/test_profile_facts.py | 178 +++++++++++++++++ backend/tests/test_valid_filter_quote.py | 39 ++++ 5 files changed, 466 insertions(+), 6 deletions(-) create mode 100644 backend/tests/test_look_large.py create mode 100644 backend/tests/test_profile_facts.py create mode 100644 backend/tests/test_valid_filter_quote.py diff --git a/backend/app/data/source_objects.py b/backend/app/data/source_objects.py index e69a3ce4..ad8f8a8d 100644 --- a/backend/app/data/source_objects.py +++ b/backend/app/data/source_objects.py @@ -480,11 +480,13 @@ async def refresh_index(session: AsyncSession, source_id: str, schema: str, resu for chunk in _chunks(new_rows): await session.execute(insert(_SO), chunk) if changed: - stmt = (update(_SO).where(_SO.id == bindparam("_id")) - .values(kind=bindparam("kind"), comment=bindparam("comment"), est_rows=bindparam("est_rows")) - .execution_options(synchronize_session=False)) + # 按表（不按 ORM 实体）的 executemany：ORM 的按主键批量更新要求参数里写主键列的名字 + table = SourceObject.__table__ + stmt = (update(table).where(table.c.id == bindparam("_id")) + .values(kind=bindparam("kind"), comment=bindparam("comment"), est_rows=bindparam("est_rows"))) + conn = await session.connection() for chunk in _chunks(changed): - await session.execute(stmt, chunk) + await conn.execute(stmt, chunk) still = [r.id for name, r in existing.items() if name in seen] for chunk in _chunks(still): await session.execute(update(_SO).where(_SO.id.in_(chunk)).values(last_seen_at=now) diff --git a/backend/app/tools/datasource.py b/backend/app/tools/datasource.py index ecfafc9f..86a2a9e4 100644 --- a/backend/app/tools/datasource.py +++ b/backend/app/tools/datasource.py @@ -854,7 +854,8 @@ def _schema_tables(table: str | list[str] | None) -> list[str]: #: db_schema 不传 table 时最多列几张（几千张表的库整份列出来没法看，也塞不进上下文）、按 keyword 检索给几个 SCHEMA_LIST_LIMIT = 50 SCHEMA_KEYWORD_LIMIT = 30 -SCHEMA_MORE_PROMPT = "…另有 {n} 张没有列出：用 keyword 按要查的内容检索，或者 table 直接点名" +SCHEMA_MORE_PROMPT = "…另有 {n} 张没有列出{outside}：用 keyword 按要查的内容检索，或者 table 直接点名" +SCHEMA_MORE_OUTSIDE = "（其中 {n} 张在探查范围外，点名时按需加入）" SCHEMA_KEYWORD_HEAD = "数据源「{source}」里和「{keyword}」相关的表（按相关程度排）：" SCHEMA_KEYWORD_NONE = "数据源「{source}」里没有检索到和「{keyword}」相关的表：换个说法，或者 table 直接点名" SCHEMA_OUTSIDE_MARK = "（探查范围外：table 点名它会按需加入）" @@ -908,7 +909,9 @@ def _make_schema_tool(source: Any, *, fixed: bool = False, lines = [f" {n}" + ("（按原样导入、未经规整，不能直接对列求和）" if n in raw else "") + (REPORTED_TOTAL_MARK if n in totals else "") for n in shown] if total > len(shown): - lines.append(" " + SCHEMA_MORE_PROMPT.format(n=total - len(shown))) + beyond = total - len(tables) + lines.append(" " + SCHEMA_MORE_PROMPT.format( + n=total - len(shown), outside=SCHEMA_MORE_OUTSIDE.format(n=beyond) if beyond > 0 else "")) return head + "：\n" + "\n".join(lines) return StructuredTool( diff --git a/backend/tests/test_look_large.py b/backend/tests/test_look_large.py new file mode 100644 index 00000000..aaad0f73 --- /dev/null +++ b/backend/tests/test_look_large.py @@ -0,0 +1,238 @@ +"""大库下的 look、db_schema、目录搜索（docs/design/next-round/03-数据接入与目录维护.md A2）。 + +3,000 张表的合成库（tests/fixtures/catalog/large.py）：探查范围 ord_*、sal_* 加上 5 张有数据目录的目标表，其余在范围外， +对象索引里只有名字和注释。这组守的是： + +- look 的挑表在候选超过 PREFILTER_LIMIT 时只给预筛结果（BM25）、已用到的表、一跳关联：10 条脚本化需求至少 9 条挑中， + 挑选提示词不到 20,000 字；挑中范围外的表按需加入，look 记录写 added_to_scope； +- 点名范围外的表按需加入、留痕（谁、哪一轮、哪一次查看、需求原文）；每次查看最多 ON_DEMAND_PER_LOOK 张，超了照实说； +- 运行时 db_schema__：keyword 用同一个检索；不传 table 只列前 50 张；点名范围外的按需加入，每次运行最多 ON_DEMAND_PER_RUN 张； +- 查询照实列「未按目录检查的表」； +- 目录：GET /catalog/search 找得到范围外的表，POST /catalog/scope/add 加进来（记进规则的显式列出）。 + +挑选模型用一个按需求和目录行的字面重合打分的假模型（不调真模型）：它只能看到交给它的那份目录，预筛漏掉的它也挑不到。 +""" +from __future__ import annotations + +import re +import uuid + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import select + +from app.data import catalog +from app.db.base import SessionLocal +from app.db.models import DataSource +from app.db.models_sources import SourceObject +from app.look.base import ON_DEMAND_PER_LOOK, ON_DEMAND_PER_RUN, PICK_LIMIT +from app.look.pick import PICK_HUMAN_PROMPT, PICK_SYSTEM_PROMPT, PICK_USED_PROMPT +from app.look.service import LookService +from app.look.tables import TableKind +from app.main import app +from tests.fixtures.catalog import large +from tests.fixtures.sources import drop_source, introspect_now + + +class OverlapPicker: + """假的挑选模型：按需求里的二字片段在目录行里出现的次数给每行打分，挑前 PICK_LIMIT 行。记下每次的提示词字数。""" + + def __init__(self) -> None: + self.prompts: list[int] = [] + self.calls = 0 + + async def pick(self, *, need: str, what: str, line: str, index: str, used: str = ""): + human = PICK_HUMAN_PROMPT.format(need=need, used=PICK_USED_PROMPT.format(items=used) if used else "", + what=what, line=line, index=index) + self.prompts.append(len(PICK_SYSTEM_PROMPT) + len(human)) + self.calls += 1 + grams = {need[i:i + 2] for i in range(len(need) - 1) if re.match(r"[一-鿿]{2}", need[i:i + 2])} + source, scored = None, [] + for row in index.splitlines(): + if m := re.match(r"数据源「(.+?)」", row): + source = m.group(1) + continue + if not row.strip() or "｜" not in row and not re.match(r"^[\w.]+", row): + continue + name = row.split("｜", 1)[0].strip() + score = sum(1 for g in grams if g in row) + if score: + scored.append((score, source, name)) + scored.sort(key=lambda x: -x[0]) + return [(s, n) for _score, s, n in scored[:PICK_LIMIT]] + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +@pytest.fixture(scope="module") +def big_db(tmp_path_factory) -> str: + return str(large.build(tmp_path_factory.mktemp("large-look") / "large.db")) + + +@pytest.fixture +async def big(client, big_db): + """探查范围 ord_*、sal_*（300 张）+ 5 张有目录的目标表（pinned）；对象索引里写上全部注释（SQLite 没有表注释）。""" + name = f"large_{uuid.uuid4().hex[:8]}" + r = await client.post("/api/datasources", json={"name": name, "kind": "sqlite", "database": big_db, + "readonly": True}) + sid = r.json()["id"] + async with SessionLocal() as session: + for t in large.TARGETS: + if t.cataloged: + await catalog.write_entry(session, sid, t.name, {"label": {"value": t.comment, "source": "human", + "status": "confirmed"}}, + if_version=0, actor=None) + r = await client.put(f"/api/datasources/{sid}/introspect-scope", + json={"scope": {"mode": "rules", "include": ["ord_*", "sal_*"]}}) + assert r.status_code == 200 + out = await introspect_now(client, sid) + assert out["table_count"] == 305 + # 探查之后再写注释：刷新对象索引会照 SQLite 的字典（没有注释）改回去 + notes = large.comments() + async with SessionLocal() as session: + for obj in (await session.execute(select(SourceObject).where(SourceObject.source_id == sid))).scalars(): + obj.comment = notes.get(obj.name) + await session.commit() + yield sid, name + await drop_source(sid) + + +async def _row(sid: str) -> DataSource: + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + session.expunge(row) + return row + + +async def test_prefiltered_pick_hits_at_least_9_of_10_with_a_short_prompt(big): + sid, name = big + picker = OverlapPicker() + kind = await TableKind.load([await _row(sid)], picker=picker, trail={"actor": "tester", "conversation_id": "c1"}) + looks = LookService([kind]) + hits = 0 + added: list[str] = [] + for target in large.TARGETS: + result = await looks.look({"op": "look", "need": target.need}) + objects = (result.record or {}).get("objects") or [] + picked = {n for o in objects for n in o.get("picked", [])} + if target.name in picked: + hits += 1 + for o in objects: + assert o.get("prefiltered", 0) <= 120 + 40 + added += o.get("added_to_scope", []) + assert hits >= 9, f"挑中 {hits}/10" + assert picker.prompts and max(picker.prompts) < 20_000, f"挑选提示词 {max(picker.prompts)} 字" + out_of_scope = {t.name for t in large.TARGETS if not t.cataloged} + assert out_of_scope & set(added), "挑中范围外的表时按需加入，记在 added_to_scope" + rows = {r.name: r for r in (await _objects(sid)) if r.name in out_of_scope and r.in_scope} + assert rows and all(r.added_via == "look" and r.added_by == "tester" and r.added_conversation_id == "c1" + and r.added_need and r.added_ref for r in rows.values()) + + +async def _objects(sid: str) -> list[SourceObject]: + async with SessionLocal() as session: + return list((await session.execute(select(SourceObject).where(SourceObject.source_id == sid))).scalars()) + + +async def test_naming_out_of_scope_tables_adds_them_with_a_per_look_cap(big): + sid, name = big + kind = await TableKind.load([await _row(sid)], picker=OverlapPicker(), trail={"actor": "tester"}) + looks = LookService([kind]) + wanted = ["shift_roster", "complaint_ticket", "waybill_track", "pur_hdr_east", "pur_line_east", "pur_log_east"] + result = await looks.look({"op": "look", "need": "看这几张表的字段", + "targets": [{"kind": "table", "source": name, "name": n} for n in wanted]}) + (obj,) = result.record["objects"] + assert len(obj["added_to_scope"]) == ON_DEMAND_PER_LOOK == 4 + assert set(obj["named"]) == set(obj["added_to_scope"]) + assert "按需加入探查范围" in result.text and f"一次查看最多按需加入 {ON_DEMAND_PER_LOOK} 张" in result.text + assert sorted(obj["missing"]) == sorted(wanted[4:]), "超了上限的照实写没找到、说清原因" + assert "biz_date" in result.text, "加进来的表当场给字段" + row = await _row(sid) + assert all(n in row.schema_cache["tables"] for n in obj["added_to_scope"]), "落盘，下一轮、运行时都看得到" + # 下一次查看重新算上限 + again = await looks.look({"op": "look", "need": "再看两张", + "targets": [{"kind": "table", "source": name, "name": n} for n in wanted[4:]]}) + assert len(again.record["objects"][0]["added_to_scope"]) == 2 + + +async def test_db_schema_keyword_list_limit_and_per_run_cap(big): + from app.tools.datasource import build_datasource_tools + from app.tools.registry import ToolContext, Versions + + sid, name = big + async with SessionLocal() as session: + tools = await build_datasource_tools([f"db_schema__{name}", f"db_query__{name}"], + ToolContext(run_id="run-cap-1", node_id="n1", + data_versions=Versions.CURRENT), session) + schema = next(t for t in tools if t.name.startswith("db_schema__")) + query = next(t for t in tools if t.name.startswith("db_query__")) + listed = await schema.coroutine() + assert listed.count("\n ") == 51 and "另有 2950 张没有列出（其中 2695 张在探查范围外" in listed and "keyword" in listed + found = await schema.coroutine(keyword="员工排班") + assert "shift_roster｜员工排班表（探查范围外：table 点名它会按需加入）" in found + + names = [f"pur_{e}_east" for e in ("hdr", "line", "log", "snap", "daily", "monthly", "cfg", "map", "hist")] + text = await schema.coroutine(table=names) + assert text.count("按需加入了探查范围") == ON_DEMAND_PER_RUN == 8 + assert f"这次运行最多按需加入 {ON_DEMAND_PER_RUN} 张" in text, "第 9 张照实说上限用完了" + added = [r for r in await _objects(sid) if r.added_run_id == "run-cap-1"] + assert len(added) == 8 and all(r.added_via == "db_schema" and r.added_ref == "n1" for r in added) + # 查询工具看到的是同一份表结构：加进来的表不再列成「未按目录检查的表」，范围外的照实列 + import json + + out = json.loads(await query.coroutine(sql="SELECT a.id FROM pur_hdr_east a JOIN pur_tmp_east b ON a.id = b.id")) + assert out.get("unchecked_tables") == ["pur_tmp_east"] + # 删掉了、从来没有的分开说 + assert "没有 no_such_table" in await schema.coroutine(table="no_such_table") + + +async def test_catalog_search_and_add_to_scope(client, big): + sid, name = big + r = await client.get(f"/api/datasources/{sid}/catalog/search", params={"q": "投诉"}) + body = r.json() + assert body["indexed"] is True and body["total"] >= 1 + hit = next(i for i in body["items"] if i["name"] == "complaint_ticket") + assert hit["in_scope"] is False and hit["absent_reason"] == "out_of_scope" and hit["comment"] == "客户投诉工单" + r = await client.get(f"/api/datasources/{sid}/catalog/search", params={"q": "ORD_HDR"}) + assert r.json()["items"][0]["name"].startswith("ord_hdr") and r.json()["items"][0]["in_scope"] is True + + r = await client.post(f"/api/datasources/{sid}/catalog/scope/add", + json={"tables": ["complaint_ticket", "nope_table", "ord_hdr_east"]}) + assert r.status_code == 200, r.text + out = r.json() + assert out["added"] == ["complaint_ticket"] and out["already"] == ["ord_hdr_east"] + assert [x["name"] for x in out["refused"]] == ["nope_table"] + row = await _row(sid) + assert "complaint_ticket" in row.schema_cache["tables"] + assert "complaint_ticket" in row.options["introspect_scope"]["tables"], "记进规则的显式列出，下次探查也带上" + listed = (await client.get(f"/api/datasources/{sid}/catalog")).json() + assert any(t["table_name"] == "complaint_ticket" for t in listed["tables"]) + scope = (await client.get(f"/api/datasources/{sid}/introspect-scope")).json() + trail = next(t for t in scope["on_demand"] if t["name"] == "complaint_ticket") + assert trail["via"] == "catalog" and trail["scope_reason"] == "explicit" + + +async def test_out_of_scope_parent_is_not_called_missing(big): + from app.data import catalog_profile + + sid, _ = big + row = await _row(sid) + from app.data import catalog_profile_jobs + + async with SessionLocal() as session: + outside = await catalog_profile_jobs.outside_of(session, row) + tables = row.schema_cache["tables"] + rel = {"id": "r1", "columns": ["ord_id"], "to_table": "pur_hdr_west", "to_columns": ["id"], "status": "proposed", + "source": "name"} + checks = catalog_profile.plan_checks("ord_hdr_east", tables["ord_hdr_east"], {"relations": [rel]}, + tables=tables, masked=set(), outside=outside) + (relation,) = [c for c in checks if c.kind == "relation"] + assert relation.skip[0] == "out_of_scope" and "探查范围外" in relation.skip[1] and "重新探查" not in relation.skip[1] + rel["to_table"] = "really_missing" + checks = catalog_profile.plan_checks("ord_hdr_east", tables["ord_hdr_east"], {"relations": [rel]}, + tables=tables, masked=set(), outside=outside) + assert [c for c in checks if c.kind == "relation"][0].skip[0] == "missing" diff --git a/backend/tests/test_profile_facts.py b/backend/tests/test_profile_facts.py new file mode 100644 index 00000000..5594c2f5 --- /dev/null +++ b/backend/tests/test_profile_facts.py @@ -0,0 +1,178 @@ +"""剖析事实进 look 的表详情（docs/design/next-round/05-已知遗留.md (e)）。 + +剖析记下的行数估算和依据、各日期列的范围以前只在 catalog_profile_tasks 里，助手写 SQL 时不知道表有多大、数据到哪天。现在 +经 schema_view.profile_lines 成一节，look、运行时 db_schema、/schema、表助手四处同一段文字，附「只供写查询参考……不能 +作为回答里的数」。这组守的是： + +- 四处文字一致；没有剖析结果时表详情和以前一字不差； +- 遮罩按渲染那一刻重滤：剖析之后才给某一列加遮罩，那一列的范围不出现； +- 上传表格只在剖析时的快照和这次读的快照一致时给，不一致（含升级前没记快照的）写「剖析早于当前版本，不给」； +- 剖析任务开始一张表时记下上传表格的快照； +- 「上次剖析」取真正开始过的最近一次（同表清单）。 +""" +from __future__ import annotations + +import sqlite3 +import uuid +from datetime import datetime, timezone +from types import SimpleNamespace + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.data import catalog_profile_jobs, schema_view +from app.data.schema_view import PROFILE_HINT_PROMPT, PROFILE_STALE_PROMPT +from app.db.base import SessionLocal, new_id +from app.db.models import CatalogProfileJob, CatalogProfileTask, DataSource +from app.main import app +from tests.fixtures.sources import drop_source, introspect_now + +FACT_LINE = "数据剖析（2026-10-01）：约 12,345 行（数据库统计估算）；order_date 从 2024-01-01 到 2026-09-30；paid_at 从 2024-01-02 到 2026-09-29 18:00:01" + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +@pytest.fixture +async def shop(client, tmp_path): + path = tmp_path / "shop.db" + db = sqlite3.connect(path) + db.execute("CREATE TABLE orders (id INTEGER PRIMARY KEY, order_date TEXT, paid_at TEXT, amount REAL)") + db.execute("CREATE TABLE stores (id INTEGER PRIMARY KEY, name TEXT)") + db.commit() + db.close() + name = f"facts_{uuid.uuid4().hex[:8]}" + r = await client.post("/api/datasources", json={"name": name, "kind": "sqlite", "database": str(path)}) + sid = r.json()["id"] + await introspect_now(client, sid) + yield sid, name + await drop_source(sid) + + +async def _profiled(sid: str, *, started: bool = True, snapshot: str | None = None, rows: int = 12345, + at: str = "2026-10-01T08:00:00+00:00") -> None: + """记一次剖析：orders 有行数估算和两个日期列的范围。""" + when = datetime.fromisoformat(at) + async with SessionLocal() as session: + job = CatalogProfileJob(id=new_id(), source_id=sid, status="succeeded", scope="selected", + created_at=when, finished_at=when) + session.add(job) + await session.flush() + session.add(CatalogProfileTask( + job_id=job.id, seq=0, table_name="orders", status="ok" if started else "cancelled", + row_estimate={"rows": rows, "method": "stats", "at_least": None} if started else None, + date_ranges=[{"column": "order_date", "min": "2024-01-01", "max": "2026-09-30"}, + {"column": "paid_at", "min": "2024-01-02 00:00:00", "max": "2026-09-29 18:00:01"}] + if started else None, + snapshot_id=snapshot, started_at=when if started else None, finished_at=when)) + await session.commit() + + +async def _row(sid: str) -> DataSource: + async with SessionLocal() as session: + row = await session.get(DataSource, sid) + session.expunge(row) + return row + + +async def _four_places(client, sid: str, name: str) -> dict[str, str]: + from app.data import catalog_assistant + from app.look.service import LookService + from app.look.tables import TableKind + from app.tools.datasource import build_datasource_tools + from app.tools.registry import ToolContext, Versions + + row = await _row(sid) + looks = LookService([await TableKind.load([row])]) + look = (await looks.look({"op": "look", "need": "订单", "targets": [ + {"kind": "table", "source": name, "name": "orders"}]})).text + async with SessionLocal() as session: + (tool,) = await build_datasource_tools([f"db_schema__{name}"], ToolContext(data_versions=Versions.CURRENT), + session) + assistant = (await catalog_assistant.prepare(session, row, row, "orders")).system + tool_text = await tool.coroutine(table="orders") + api = (await client.get(f"/api/datasources/{sid}/schema", params={"table": "orders"})).json()["detail"] + return {"look": look, "db_schema": tool_text, "/schema": api, "assistant": assistant} + + +async def test_the_same_profile_facts_in_four_places(client, shop): + sid, name = shop + before = (await client.get(f"/api/datasources/{sid}/schema", params={"table": "orders"})).json()["detail"] + await _profiled(sid) + places = await _four_places(client, sid, name) + for where, text in places.items(): + assert FACT_LINE in text, f"{where} 里没有那一段剖析事实：{text[-400:]}" + assert PROFILE_HINT_PROMPT in text, f"{where} 里没有「只供写查询参考」" + after = places["/schema"] + assert after.startswith(before) and after != before, "剖析事实是追加的一节，表详情原有的部分不动" + + # 剖析之后才给 paid_at 加遮罩：渲染那一刻重滤，它的范围不出现 + r = await client.patch(f"/api/datasources/{sid}", json={"options": {"mask_columns": "paid_at"}}) + assert r.status_code == 200, r.text + for where, text in (await _four_places(client, sid, name)).items(): + assert "paid_at 从" not in text, f"{where} 给了遮罩列的范围" + assert "order_date 从 2024-01-01 到 2026-09-30" in text, where + + +async def test_without_profile_results_the_detail_is_unchanged(shop): + sid, _ = shop + row = await _row(sid) + plain = schema_view.table_detail(row, "orders") + async with SessionLocal() as session: + facts = await catalog_profile_jobs.last_facts(session, sid) + assert facts == {} + assert schema_view.table_detail(row, "orders", profile=facts) == plain + assert schema_view.table_detail(row, "orders", profile={"stores": {"row_estimate": {"rows": 1}}}) == plain + + +async def test_last_facts_follow_the_last_profile_that_really_started(shop): + sid, _ = shop + await _profiled(sid, rows=500, at="2026-09-01T08:00:00+00:00") + await _profiled(sid, started=False, at="2026-10-02T08:00:00+00:00") + async with SessionLocal() as session: + facts = await catalog_profile_jobs.last_facts(session, sid) + assert await catalog_profile_jobs.last_facts(session, sid, []) == {} + assert facts["orders"]["row_estimate"]["rows"] == 500 and facts["orders"]["at"].startswith("2026-09-01"), \ + "没开始就停了的那一项不挡掉上一次真正剖析的结果" + + +def test_upload_sources_only_get_facts_of_the_current_snapshot(): + facts = {"at": "2026-10-01T00:00:00+00:00", "snapshot_id": "snap-a", + "row_estimate": {"rows": 3, "method": "count", "at_least": None}, "date_ranges": []} + assert schema_view.profile_lines(facts, set(), "snap-a", upload=True)[0].endswith("3 行（逐行数出）") + assert schema_view.profile_lines(facts, set(), "snap-b", upload=True) == [" " + PROFILE_STALE_PROMPT] + assert schema_view.profile_lines({**facts, "snapshot_id": None}, set(), "snap-a", upload=True) \ + == [" " + PROFILE_STALE_PROMPT], "升级前没记快照的剖析，上传表格在重新剖析之前不给" + at_least = {**facts, "row_estimate": {"rows": None, "method": "count", "at_least": 100000}} + assert "至少 100,000 行（数到上限为止）" in schema_view.profile_lines(at_least, set(), None)[0] + unknown = {**facts, "row_estimate": {"rows": None, "method": "unknown", "at_least": None}} + assert schema_view.profile_lines(unknown, set(), None) == [], "什么都没有就不出这一节" + + view = SimpleNamespace(name="up", origin="upload", snapshot_id="snap-b", options={}, kind="sqlite", + schema_cache={"tables": {"t": {"qualified": "t", "columns": [{"name": "id", "type": "INTEGER"}]}}}) + text = schema_view.table_detail(view, "t", profile={"t": facts}) + assert PROFILE_STALE_PROMPT in text and "3 行" not in text + + +async def test_profile_job_records_the_snapshot_of_an_upload_table(shop): + sid, _ = shop + from app.data.catalog_profile_jobs import _ProfileRun + + async with SessionLocal() as session: + job = CatalogProfileJob(id=new_id(), source_id=sid, status="running", scope="selected", + created_at=datetime.now(timezone.utc)) + session.add(job) + await session.flush() + session.add(CatalogProfileTask(job_id=job.id, seq=0, table_name="orders", status="pending")) + await session.commit() + run = _ProfileRun(job.id) + run.cx = SimpleNamespace(source=SimpleNamespace(snapshot_id="snap-z")) + async with SessionLocal() as session: + await run._start_table(session, 0) + async with SessionLocal() as session: + task = (await session.get(CatalogProfileJob, job.id)) and (await session.execute( + __import__("sqlalchemy").select(CatalogProfileTask).where(CatalogProfileTask.job_id == job.id))).scalar_one() + assert task.snapshot_id == "snap-z" and task.status == "running" diff --git a/backend/tests/test_valid_filter_quote.py b/backend/tests/test_valid_filter_quote.py new file mode 100644 index 00000000..31d50303 --- /dev/null +++ b/backend/tests/test_valid_filter_quote.py @@ -0,0 +1,39 @@ +"""有效记录条件按 quote_name 强制加引号（第 1 波留下的一处，next-round 03 C1 ③）。 + +Oracle 库里本来就是小写的列（建表时写了 "status"）探查时记了 quote_name。拼有效记录条件时只按 SqlDialect.quote（要加才加） +写成不带引号的 status，Oracle 转成大写 STATUS，找的是另一列。存进目录的写法（stored_valid_filter）和剖析核对查询 +（check_valid_filter）都要按表结构里的 quote_name 加引号——两处是同一条。 +""" +from __future__ import annotations + +from types import SimpleNamespace + +from app.data import catalog_profile, sqlcheck + +META = {"qualified": "ANALYTICS.visits", "schema": "ANALYTICS", "quote_name": True, + "columns": [{"name": "status", "type": "VARCHAR2(10 CHAR)", "nullable": True, "quote_name": True}, + {"name": "amount", "type": "NUMBER(12, 2)", "nullable": True}]} + + +def test_stored_filter_quotes_columns_recorded_with_quote_name(): + parsed = sqlcheck.parse_valid_filter("status <> 'X' AND amount > 0", META, kind="oracle", table="visits") + assert parsed.problem is None + assert sqlcheck.stored_valid_filter(parsed, "oracle", META) == "\"status\" <> 'X' AND amount > 0" + # 不给表结构时照旧按方言要加才加（老的调用方） + assert sqlcheck.stored_valid_filter(parsed, "oracle") == "status <> 'X' AND amount > 0" + + +async def test_the_profile_check_quotes_the_same_way(monkeypatch): + seen: list[str] = [] + + async def fake(source, sql, **kw): + seen.append(sql) + return SimpleNamespace(rows=[[10, 2]]) + + monkeypatch.setattr(catalog_profile, "run_single_query", fake) + source = SimpleNamespace(kind="oracle", enabled=True, options={catalog_profile.PROFILE_OPTION: {"enabled": True}}) + found = await catalog_profile.check_valid_filter(source, "visits", META, "status <> 'X'") + assert found.rows == 10 and len(seen) == 1 + sql = seen[0] + assert "CASE WHEN \"status\" <> 'X' THEN 0" in sql, sql + assert "SELECT \"status\" FROM \"ANALYTICS\".\"visits\"" in sql, "取的列、表名也按 quote_name 加引号"