"""基于数据目录的 SQL 检查（data/sqlcheck.py）：七条规则的正例、反例，级别随依据的状态变化，以及解析不了的写法。 用合成的景区业务库（tests/fixtures/catalog/scenic.py）和它的一份目录（scenic_notes.py）。检查只做有把握的判断： 解析不了、遇到不支持的写法、列对不上表，一律跳过不报——它的 error 会把图打回去改、挡住受管级别的发布、让出具降档， 误报的代价比漏报大。 """ from __future__ import annotations from types import SimpleNamespace import pytest from app.data.engine import engines from app.data.introspect import introspect from app.data.sqlcheck import CHECK_CODES, SqlChecker, check_sql, dialect_of from tests.fixtures.catalog import scenic_notes from tests.fixtures.catalog.scenic_notes import ORDER_ITEMS_TO_ORDERS _CACHE: dict[str, dict] = {} @pytest.fixture async def cache(scenic_db): """景区库的表结构缓存（真实探查一次：带外键和唯一约束，关系现推要用）。每个测试进程只探查一次。""" if scenic_db not in _CACHE: source = SimpleNamespace(id="sqlcheck-scenic", name="scenic", kind="sqlite", database=scenic_db, host=None, port=None, username=None, password=None, options={}, readonly=True, description="", schema_cache={}, origin="manual") try: _CACHE[scenic_db] = await introspect(source) finally: await engines.invalidate(source.id) return _CACHE[scenic_db] def run(cache, sql, status="confirmed", override=None, kind="sqlite"): checker = SqlChecker(kind=kind, schema_cache=cache, notes=scenic_notes.notes(status, override), source_name="scenic") return checker.check(sql) def found(cache, sql, status="confirmed", override=None, kind="sqlite"): """[(code, level, table, column)]，按检查给出的顺序。""" return [(c.code, c.level, c.table, c.column) for c in run(cache, sql, status, override, kind)] def codes(cache, sql, status="confirmed", override=None): return {c.code for c in run(cache, sql, status, override)} # -------------------------------------------------------------------------- # 结果的形状 # -------------------------------------------------------------------------- async def test_result_shape_and_wording(cache): sql = ("SELECT o.channel_id, SUM(o.total_amount) AS amt FROM orders o " "JOIN order_items i ON i.order_id = o.id WHERE o.status = 1 GROUP BY o.channel_id") [check] = run(cache, sql) out = check.as_dict() assert set(out) == {"code", "level", "message", "for_model", "table", "column", "relation_id", "sql_excerpt"} assert out["code"] == "fanout_sum" and out["level"] == "error" assert out["table"] == "orders" and out["column"] == "total_amount" assert out["relation_id"] == ORDER_ITEMS_TO_ORDERS assert "SUM(o.total_amount)" in out["sql_excerpt"] assert out["message"].startswith("「订单」关联「订单明细」是一对多，对「订单」的「订单金额」求和会重复计算。") # 给模型的写 SQL 里的真名和改法；给人看的不出现工具名和对模型的指令 assert "orders" in out["for_model"] and "order_items" in out["for_model"] assert "db_schema__" not in out["message"] and "db_" not in out["message"] # 没有的可选字段不出现 single = run(cache, "SELECT AVG(discount_rate) FROM member_levels")[0].as_dict() assert "relation_id" not in single and single["column"] == "discount_rate" assert set(CHECK_CODES) == {"fanout_sum", "stock_summed", "join_unconfirmed", "ratio_aggregated", "missing_valid_filter", "unknown_code", "wrong_date_column"} def test_dialects(): assert dialect_of("sqlite") == "sqlite" assert dialect_of("postgres") == dialect_of("postgresql") == "postgres" assert dialect_of("mysql") == dialect_of("mariadb") == "mysql" assert dialect_of("oracle") == "oracle" assert dialect_of("mssql") is None and dialect_of(None) is None # -------------------------------------------------------------------------- # fanout_sum：沿一对多关联之后对「一」那一侧的度量列求和、计数 # -------------------------------------------------------------------------- FANOUT = [ "SELECT SUM(o.total_amount) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1", # 先写明细、后连订单：订单照样随明细重复 "SELECT SUM(o.total_amount) FROM order_items i JOIN orders o ON o.id = i.order_id WHERE o.status = 1", # 包一层 COALESCE、CASE 里只用订单的列：还是订单的数 "SELECT SUM(COALESCE(o.total_amount, 0)) FROM orders o LEFT JOIN order_items i ON o.id = i.order_id " "WHERE o.status = 1", "SELECT SUM(CASE WHEN o.status = 1 THEN o.total_amount END) FROM orders o JOIN order_items i " "ON i.order_id = o.id AND i.qty > 0", # 逗号连接、条件写在 WHERE 里 "SELECT SUM(o.total_amount) FROM orders o, order_items i WHERE i.order_id = o.id AND o.status = 1", ] @pytest.mark.parametrize("sql", FANOUT) async def test_fanout_sum_positive(cache, sql): assert ("fanout_sum", "error", "orders", "total_amount") in found(cache, sql) async def test_fanout_count_of_the_one_side_key(cache): sql = "SELECT COUNT(o.id) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1" [check] = run(cache, sql) assert (check.code, check.level, check.table, check.column) == ("fanout_sum", "error", "orders", "id") assert "计数会重复计算" in check.message async def test_fanout_through_two_hops_with_an_inferred_foreign_key(cache): """支付记录没写目录：订单 → 支付记录的关系按外键现推（verified），明细经订单连支付记录，明细的金额照样重复。""" sql = ("SELECT SUM(i.amount) FROM order_items i JOIN orders o ON o.id = i.order_id " "JOIN payments p ON p.order_id = o.id WHERE o.status = 1") [check] = [c for c in run(cache, sql) if c.code == "fanout_sum"] assert (check.level, check.table, check.column) == ("error", "order_items", "amount") assert "「订单」关联「payments」是一对多，「订单明细」的每一行会随之重复" in check.message assert "去掉与「payments」的关联" in check.message FANOUT_NEGATIVE = [ # 对「多」那一侧求和：本来就在明细粒度上 "SELECT SUM(i.amount) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1", "SELECT SUM(i.amount) FROM order_items i JOIN orders o ON o.id = i.order_id WHERE o.status = 1", # 不关联 "SELECT SUM(o.total_amount) FROM orders o WHERE o.status = 1", # 去重计数、去重求和：写的人已经在处理重复 "SELECT COUNT(DISTINCT o.id) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1", # 两边的列一起算：按明细逐行算出来的数 "SELECT SUM(i.qty * o.total_amount) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1", # 按明细的主键分组：每组只有一行明细，订单不重复 "SELECT i.id, SUM(o.total_amount) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1 " "GROUP BY i.id", # 明细先在子查询里按订单汇总再连：一对一 "SELECT SUM(o.total_amount), SUM(x.n) FROM orders o JOIN (SELECT order_id, COUNT(*) AS n FROM order_items " "GROUP BY order_id) x ON x.order_id = o.id WHERE o.status = 1", # 计数所有行：说不清是要数订单还是数明细 "SELECT COUNT(*) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1", # 窗口函数：不是分组聚合 "SELECT o.id, SUM(o.total_amount) OVER (PARTITION BY o.channel_id) FROM orders o " "JOIN order_items i ON i.order_id = o.id WHERE o.status = 1", # 反连接：只留没有明细的订单，每笔订单至多一行 "SELECT SUM(o.total_amount) FROM orders o LEFT JOIN order_items i ON i.order_id = o.id " "WHERE i.id IS NULL AND o.status = 1", ] @pytest.mark.parametrize("sql", FANOUT_NEGATIVE) async def test_fanout_sum_negative(cache, sql): assert "fanout_sum" not in codes(cache, sql) async def test_fanout_level_follows_the_status_of_its_basis(cache): sql = FANOUT[0] assert ("fanout_sum", "error", "orders", "total_amount") in found(cache, sql, "verified") assert ("fanout_sum", "info", "orders", "total_amount") in found(cache, sql, "proposed") # 关系确认了、度量类型只是推断：依据里有推断就降为 info assert ("fanout_sum", "info", "orders", "total_amount") in found( cache, sql, override={("orders", "columns.total_amount.measure"): "proposed"}) # 关系被驳回：不知道是一对多，不报重复计算；关联本身对不上目录，单独提示 rejected = found(cache, sql, override={("order_items", f"relations.{ORDER_ITEMS_TO_ORDERS}"): "rejected"}) assert not [f for f in rejected if f[0] == "fanout_sum"] assert ("join_unconfirmed", "warning", "order_items", None) in rejected # 度量类型被驳回：不知道订单金额能不能加，不报 assert "fanout_sum" not in codes(cache, sql, override={("orders", "columns.total_amount.measure"): "rejected"}) # -------------------------------------------------------------------------- # stock_summed：存量跨时间求和 # -------------------------------------------------------------------------- STOCK = [ "SELECT SUM(on_hand) FROM inventory_snapshots", 'SELECT "storeID", SUM(on_hand) FROM inventory_snapshots GROUP BY "storeID"', "SELECT SUM(s.on_hand) FROM inventory_snapshots s WHERE s.snapshot_date BETWEEN '2026-09-01' AND '2026-09-30'", ] @pytest.mark.parametrize("sql", STOCK) async def test_stock_summed_positive(cache, sql): assert ("stock_summed", "error", "inventory_snapshots", "on_hand") in found(cache, sql) STOCK_NEGATIVE = [ "SELECT snapshot_date, SUM(on_hand) FROM inventory_snapshots GROUP BY snapshot_date", "SELECT SUM(on_hand) FROM inventory_snapshots WHERE snapshot_date = '2026-09-30'", "SELECT substr(snapshot_date, 1, 7) AS m, SUM(on_hand) FROM inventory_snapshots GROUP BY substr(snapshot_date, 1, 7)", "SELECT AVG(on_hand) FROM inventory_snapshots", "SELECT MAX(on_hand) FROM inventory_snapshots", "SELECT snapshot_date, SUM(on_hand) OVER (PARTITION BY snapshot_date) FROM inventory_snapshots", # 经日期维度表分组：维度表的日期和快照日期相等 "SELECT d.biz_date, SUM(s.on_hand) FROM inventory_snapshots s JOIN dim_dates d ON d.biz_date = s.snapshot_date " "GROUP BY d.biz_date", ] @pytest.mark.parametrize("sql", STOCK_NEGATIVE) async def test_stock_summed_negative(cache, sql): assert "stock_summed" not in codes(cache, sql) async def test_stock_level_follows_status(cache): sql = STOCK[0] assert ("stock_summed", "error", "inventory_snapshots", "on_hand") in found(cache, sql, "verified") assert ("stock_summed", "info", "inventory_snapshots", "on_hand") in found(cache, sql, "proposed") assert "stock_summed" not in codes(cache, sql, "rejected") # 业务日期被驳回：不知道按哪一列算时间，不报 assert "stock_summed" not in codes(cache, sql, override={("inventory_snapshots", "business_date"): "rejected"}) # -------------------------------------------------------------------------- # join_unconfirmed：关联条件对不上目录里的关系 # -------------------------------------------------------------------------- async def test_join_unconfirmed_positive(cache): # 列连错了：订单的主键对明细的主键 wrong = found(cache, "SELECT SUM(i.amount) FROM orders o JOIN order_items i ON i.id = o.id WHERE o.status = 1") assert ("join_unconfirmed", "warning", "order_items", None) in wrong [check] = [c for c in run(cache, "SELECT o.id FROM orders o JOIN members m ON m.id = o.channel_id " "WHERE o.status = 1") if c.code == "join_unconfirmed"] assert check.level == "warning" and "o.channel_id" in check.sql_excerpt assert check.table == "members" and "「订单」与「members」的关联条件" in check.message # 只对上命名推断出来的关系（入园记录.member_id → 会员）：提示一句，级别 info proposed = found(cache, "SELECT m.name FROM visits v JOIN members m ON m.id = v.member_id WHERE v.status = 1") assert ("join_unconfirmed", "info", "members", None) in proposed JOIN_NEGATIVE = [ "SELECT SUM(i.amount) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1", # 多写的过滤条件不影响对关系 "SELECT SUM(i.amount) FROM orders o JOIN order_items i ON i.order_id = o.id AND i.qty > 0 WHERE o.status = 1", # 没写目录的表之间按外键现推（verified） "SELECT SUM(p.paid_amount) FROM orders o JOIN payments p ON p.order_id = o.id WHERE o.status = 1", # 两张表都没写目录：没有可对照的目录，不判 "SELECT COUNT(*) FROM tour_groups t JOIN agencies a ON a.channel_id = t.agency_id", # 子查询、CTE 当一边：不知道它的粒度，不判 "WITH x AS (SELECT order_id, SUM(amount) AS a FROM order_items GROUP BY order_id) " "SELECT SUM(x.a) FROM orders o JOIN x ON x.order_id = o.id WHERE o.status = 1", # 自己连自己：目录不记这种关系 "SELECT c.name FROM categories c JOIN categories p ON p.id = c.\"parentId\"", # 没有等值条件 "SELECT COUNT(*) FROM orders o JOIN order_items i ON i.amount > o.total_amount WHERE o.status = 1", # OR 连起来的条件：说不清按哪条连 "SELECT COUNT(*) FROM orders o JOIN order_items i ON i.order_id = o.id OR i.id = o.id WHERE o.status = 1", ] @pytest.mark.parametrize("sql", JOIN_NEGATIVE) async def test_join_unconfirmed_negative(cache, sql): assert "join_unconfirmed" not in codes(cache, sql) async def test_sibling_join_through_a_shared_parent_is_only_a_note(cache): """两张表各自按 member_id 指向会员（都是外键），彼此按 member_id 相连：目录里没有这两张表之间的关系， 但这是经共同父表推得出来的连法，不是「连错了列」。只提示一句：关联后行数可能成倍增加。""" sql = "SELECT COUNT(*) FROM orders o JOIN member_tag_links t ON t.member_id = o.member_id WHERE o.status = 1" [check] = [c for c in run(cache, sql) if c.code == "join_unconfirmed"] assert check.level == "info" assert "会员" in check.message or "members" in check.message assert "成倍" in check.message async def test_join_level_follows_status(cache): sql = "SELECT SUM(i.amount) FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1" assert "join_unconfirmed" not in codes(cache, sql, "confirmed") assert "join_unconfirmed" not in codes(cache, sql, "verified") # 结果记在 JOIN 引入的那张表上 assert ("join_unconfirmed", "info", "order_items", None) in found(cache, sql, "proposed") assert ("join_unconfirmed", "warning", "order_items", None) in found(cache, sql, "rejected") # -------------------------------------------------------------------------- # ratio_aggregated：对比率求和、求平均 # -------------------------------------------------------------------------- @pytest.mark.parametrize("sql", [ "SELECT AVG(discount_rate) FROM member_levels", "SELECT SUM(discount_rate) FROM member_levels", "SELECT ROUND(AVG(l.discount_rate), 2) FROM member_levels l", ]) async def test_ratio_aggregated_positive(cache, sql): assert found(cache, sql) == [("ratio_aggregated", "warning", "member_levels", "discount_rate")] @pytest.mark.parametrize("sql", [ "SELECT level_name, discount_rate FROM member_levels", "SELECT MAX(discount_rate) FROM member_levels", # 加权：比率乘上数量再加，是正确的算法 "SELECT SUM(m.id * l.discount_rate) FROM members m JOIN member_levels l ON l.id = m.member_level_id", ]) async def test_ratio_aggregated_negative(cache, sql): assert "ratio_aggregated" not in codes(cache, sql) async def test_ratio_level_follows_status(cache): sql = "SELECT AVG(discount_rate) FROM member_levels" assert found(cache, sql, "verified") == [("ratio_aggregated", "warning", "member_levels", "discount_rate")] assert found(cache, sql, "proposed") == [("ratio_aggregated", "info", "member_levels", "discount_rate")] assert found(cache, sql, "rejected") == [] # -------------------------------------------------------------------------- # missing_valid_filter：没按有效记录条件筛 # -------------------------------------------------------------------------- @pytest.mark.parametrize("sql", [ "SELECT COUNT(*) FROM visits", "SELECT SUM(visitor_count) FROM visits WHERE park_id = 1", "SELECT v.park_id, SUM(v.visitor_count) FROM visits v GROUP BY v.park_id", ]) async def test_missing_valid_filter_positive(cache, sql): [check] = run(cache, sql) assert (check.code, check.level, check.table) == ("missing_valid_filter", "warning", "visits") assert "有效记录条件「status = 1」" in check.message @pytest.mark.parametrize("sql", [ "SELECT COUNT(*) FROM visits WHERE status = 1", "SELECT COUNT(*) FROM visits v WHERE v.status IN (0, 1)", # 按状态分组：写的人就是要看每种状态 "SELECT status, COUNT(*) FROM visits GROUP BY status", # 条件写在聚合里 "SELECT SUM(CASE WHEN status = 1 THEN visitor_count ELSE 0 END) FROM visits", # CTE 里筛过 "WITH ok AS (SELECT * FROM visits WHERE status = 1) SELECT COUNT(*) FROM ok", # 子查询原样透出、外层再筛 "SELECT COUNT(*) FROM (SELECT * FROM visits) x WHERE x.status = 1", ]) async def test_missing_valid_filter_negative(cache, sql): assert "missing_valid_filter" not in codes(cache, sql) async def test_valid_filter_level_follows_status(cache): sql = "SELECT COUNT(*) FROM visits" assert found(cache, sql, "verified") == [("missing_valid_filter", "warning", "visits", None)] assert found(cache, sql, "proposed") == [("missing_valid_filter", "info", "visits", None)] assert found(cache, sql, "rejected") == [] # -------------------------------------------------------------------------- # unknown_code：写了码值表里没有的值 # -------------------------------------------------------------------------- async def test_unknown_code_positive(cache): [check] = run(cache, "SELECT COUNT(*) FROM visits WHERE status = 3") assert (check.code, check.level, check.table, check.column) == ("unknown_code", "warning", "visits", "status") assert "「3」" in check.message and "1=有效、0=作废" in check.message [listed] = run(cache, "SELECT COUNT(*) FROM visits WHERE status IN (1, 5, 7)") assert "「5」、「7」" in listed.message and "「1」" not in listed.message assert found(cache, "SELECT COUNT(*) FROM channels WHERE channel_type = '线上'") == [ ("unknown_code", "warning", "channels", "channel_type")] @pytest.mark.parametrize("sql", [ "SELECT COUNT(*) FROM visits WHERE status = 1", "SELECT COUNT(*) FROM visits WHERE status = '1'", "SELECT COUNT(*) FROM visits WHERE 0 = status OR status = 1", "SELECT COUNT(*) FROM visits WHERE status IN (0, 1)", "SELECT COUNT(*) FROM visits WHERE status = 1.0", "SELECT COUNT(*) FROM channels WHERE channel_type = '分销'", # 模板和参数：值要到运行时才知道 "SELECT COUNT(*) FROM visits WHERE status = '{{ input.status }}'", "SELECT COUNT(*) FROM visits WHERE status = {{ input.status }}", "SELECT COUNT(*) FROM visits WHERE status = ?", "SELECT COUNT(*) FROM visits WHERE status = :status", ]) async def test_unknown_code_negative(cache, sql): assert "unknown_code" not in codes(cache, sql) async def test_unknown_code_level_follows_status(cache): sql = "SELECT COUNT(*) FROM channels WHERE channel_type = '线上'" assert found(cache, sql, "verified") == [("unknown_code", "warning", "channels", "channel_type")] assert found(cache, sql, "proposed") == [("unknown_code", "info", "channels", "channel_type")] assert found(cache, sql, "rejected") == [] # -------------------------------------------------------------------------- # wrong_date_column：按另一个时间列分组、筛选 # -------------------------------------------------------------------------- @pytest.mark.parametrize("sql", [ "SELECT substr(entered_at, 1, 10) AS d, SUM(fee) FROM parking_records GROUP BY substr(entered_at, 1, 10)", "SELECT SUM(p.fee) FROM parking_records p WHERE p.entered_at >= '2026-09-01'", ]) async def test_wrong_date_column_positive(cache, sql): [check] = run(cache, sql) assert (check.code, check.level, check.table, check.column) == ( "wrong_date_column", "warning", "parking_records", "entered_at") assert "业务日期按「出场时间」计" in check.message and "「入场时间」" in check.message @pytest.mark.parametrize("sql", [ "SELECT substr(exited_at, 1, 10) AS d, SUM(fee) FROM parking_records GROUP BY substr(exited_at, 1, 10)", "SELECT SUM(fee) FROM parking_records WHERE exited_at >= '2026-09-01' AND entered_at IS NOT NULL", # 两个时间都用上了：写的人知道两者的区别 "SELECT SUM(fee) FROM parking_records WHERE exited_at >= '2026-09-01' AND entered_at < '2026-09-30'", "SELECT entered_at, exited_at, fee FROM parking_records", ]) async def test_wrong_date_column_negative(cache, sql): assert "wrong_date_column" not in codes(cache, sql) async def test_wrong_date_level_follows_status(cache): sql = "SELECT SUM(p.fee) FROM parking_records p WHERE p.entered_at >= '2026-09-01'" assert found(cache, sql, "verified") == [("wrong_date_column", "warning", "parking_records", "entered_at")] assert found(cache, sql, "proposed") == [("wrong_date_column", "info", "parking_records", "entered_at")] assert found(cache, sql, "rejected") == [] # -------------------------------------------------------------------------- # 复杂写法：照样分析得出，或者跳过；不报错、不误报 # -------------------------------------------------------------------------- CLEAN_COMPLEX = [ # CTE + 窗口函数 + 子查询 """WITH daily AS ( SELECT substr(o.ordered_at, 1, 10) AS d, SUM(i.amount) AS amt FROM orders o JOIN order_items i ON i.order_id = o.id WHERE o.status = 1 GROUP BY substr(o.ordered_at, 1, 10)) SELECT d, amt, SUM(amt) OVER (ORDER BY d) AS running, (SELECT COUNT(*) FROM visits v WHERE v.status = 1) AS visits FROM daily ORDER BY d""", # UNION "SELECT 'web' AS ch, SUM(total_amount) FROM orders WHERE status = 1 AND channel_id = 1 " "UNION ALL SELECT 'box', SUM(total_amount) FROM orders WHERE status = 1 AND channel_id = 3", # 相关子查询、EXISTS "SELECT o.id FROM orders o WHERE o.status = 1 AND EXISTS (SELECT 1 FROM order_items i WHERE i.order_id = o.id)", # 引号括起来的驼峰标识符 'SELECT t.tag_name, COUNT(*) FROM member_tag_links l JOIN "memberTags" t ON t.id = l."memberTagId" ' "GROUP BY t.tag_name", ] @pytest.mark.parametrize("sql", CLEAN_COMPLEX) async def test_complex_but_clean_sql_reports_nothing(cache, sql): assert run(cache, sql) == [] async def test_problems_inside_cte_and_union_are_found(cache): cte = ("WITH t AS (SELECT o.channel_id, SUM(o.total_amount) AS amt FROM orders o " "JOIN order_items i ON i.order_id = o.id WHERE o.status = 1 GROUP BY o.channel_id) SELECT * FROM t") assert ("fanout_sum", "error", "orders", "total_amount") in found(cache, cte) union = ("SELECT SUM(on_hand) FROM inventory_snapshots UNION ALL " "SELECT SUM(on_hand) FROM inventory_snapshots WHERE snapshot_date = '2026-09-30'") assert found(cache, union) == [("stock_summed", "error", "inventory_snapshots", "on_hand")] @pytest.mark.parametrize("sql", [ "SELECT FROM WHERE", "SELEC total_amount FROM orders", "SELECT SUM(o.total_amount FROM orders o JOIN order_items i ON i.order_id = o.id", "SELECT 1; SELECT SUM(on_hand) FROM inventory_snapshots", "DELETE FROM visits", "", " ", # 方言特性：SQLite 读不了 Oracle 的 q 引号 "SELECT SUM(on_hand) FROM inventory_snapshots WHERE q'[x]' = 'x'", # 模板里的控制语句：拼出来是什么样说不清 "SELECT SUM(on_hand) FROM inventory_snapshots {% if x %}WHERE snapshot_date = '2026-09-30'{% endif %}", # 结构里没有的表、别名对不上的列 "SELECT SUM(x.on_hand) FROM inventory_snapshots s", "SELECT SUM(on_hand) FROM not_a_table", ]) async def test_unparseable_or_unsupported_sql_is_skipped(cache, sql): assert run(cache, sql) == [] async def test_other_dialects(cache): pg = "SELECT s.on_hand::numeric FROM inventory_snapshots s WHERE s.snapshot_date = $1" assert run(cache, pg, kind="postgres") == [] assert found(cache, "SELECT SUM(s.on_hand)::numeric FROM inventory_snapshots s", kind="postgres") == [ ("stock_summed", "error", "inventory_snapshots", "on_hand")] mysql = "SELECT `v`.`status`, COUNT(*) FROM `visits` `v` WHERE `v`.`status` = 3 GROUP BY `v`.`status`" assert found(cache, mysql, kind="mysql") == [("unknown_code", "warning", "visits", "status")] # 能读就照常分析，读不了就跳过：只要求不抛 assert isinstance(run(cache, "SELECT SUM(on_hand) FROM inventory_snapshots WHERE q'[it''s]' = 'x'", kind="oracle"), list) assert run(cache, "SELECT SUM(on_hand) FROM inventory_snapshots", kind="mssql") == [] async def test_no_catalog_means_no_checks(cache): sql = "SELECT SUM(o.total_amount) FROM orders o JOIN order_items i ON i.id = o.id" assert SqlChecker(kind="sqlite", schema_cache=cache, notes={}).check(sql) == [] assert check_sql(sql, kind="sqlite", schema_cache=cache, notes=None) == [] # 没探查过结构：对不上表，一样不判 assert check_sql(sql, kind="sqlite", schema_cache={}, notes=scenic_notes.notes()) == [] async def test_duplicates_are_folded_and_errors_come_first(cache): sql = ("SELECT SUM(o.total_amount), SUM(o.total_amount) * 2, AVG(l.discount_rate) " "FROM orders o JOIN order_items i ON i.order_id = o.id JOIN members m ON m.id = o.member_id " "JOIN member_levels l ON l.id = m.member_level_id WHERE o.status = 1") out = found(cache, sql) assert [f for f in out if f[0] == "fanout_sum"] == [("fanout_sum", "error", "orders", "total_amount")] levels = [f[1] for f in out] assert levels == sorted(levels, key=["error", "warning", "info"].index) async def test_checker_builds_relation_graph_lazily_and_reuses_it(cache, monkeypatch): from app.data import catalog calls = [] real = catalog.relation_graph monkeypatch.setattr(catalog, "relation_graph", lambda *a, **k: calls.append(1) or real(*a, **k)) checker = SqlChecker(kind="sqlite", schema_cache=cache, notes=scenic_notes.notes()) checker.check("SELECT AVG(discount_rate) FROM member_levels") assert calls == [] # 单表查询用不到关系图 checker.check(FANOUT[0]) checker.check(FANOUT[1]) assert calls == [1]