marcopan
31552782c0
44 test L1 sui 3 moduli backend con logica non banale (opzione 2 della user review): - sqlcheck.validate_sql (16 test): parse/single-statement, read-only enforcement (INSERT/UPDATE/DELETE/CREATE/DROP/ALTER/TRUNCATE/GRANT rifiutati, WITH/UNION ok), forbidden functions (dblink default blacklist, custom set, allowed not flagged), object-existence (tabella inesistente, CTE non flaggata, perimetro promoted warning, colonna inesistente con alias). Documenta una limitazione reale: le funzioni aggregate specializzate (count/sum/coalesce) NON sono catturate dal name-matcher perche' sqlglot modella .name come argomento, non come nome funzione. - ctetest (14 test): has_trailing_select (semantica controintuitiva: True = violazione), last_cte_name, build_test_sql, ledger I/O (load/append roundtrip, JSON-array e JSONL tolleranti, corrupt-ledger raise). - execute._inject_limit (6 test): LIMIT iniettato quando assente (limit+1 per troncamento), rispettato quando presente, non iniettato su non-query, UNION/WITH ok. Suite: 153 passed (109 + 44). Bonus: __psd_probe__ -> __tht_probe__ (riferimento cliente neutralizzato in ctetest).
61 lines
2.3 KiB
Python
61 lines
2.3 KiB
Python
"""L1: tht.execute._inject_limit — the AST-based LIMIT injection.
|
|
|
|
The gate runs reviewer SQL through _inject_limit before execution so previews
|
|
never return unbounded rows (and the +1 lets run_controlled detect truncation).
|
|
These tests pin: LIMIT added when absent, respected when present, never added to
|
|
non-query statements, and the truncation-detection contract (limit+1 rows).
|
|
Pure logic, no DB.
|
|
"""
|
|
import pytest
|
|
import sqlglot
|
|
|
|
from tht.execute import _inject_limit
|
|
|
|
|
|
def test_limit_injected_when_absent():
|
|
out, injected = _inject_limit("SELECT * FROM t", 10)
|
|
assert injected is True
|
|
# parse it back and confirm the LIMIT is 11 (10 + 1, for truncation detection)
|
|
ast = sqlglot.parse_one(out, read="postgres")
|
|
assert ast.args.get("limit") is not None
|
|
# the limit expression should evaluate to 11
|
|
limit_expr = ast.args["limit"].expression
|
|
assert int(limit_expr.to_py()) == 11
|
|
|
|
|
|
def test_existing_limit_respected_not_overwritten():
|
|
out, injected = _inject_limit("SELECT * FROM t LIMIT 5", 10)
|
|
assert injected is False
|
|
ast = sqlglot.parse_one(out, read="postgres")
|
|
assert int(ast.args["limit"].expression.to_py()) == 5 # unchanged
|
|
|
|
|
|
def test_no_limit_injected_on_non_query():
|
|
# a non-query statement (DDL): _inject_limit must leave it untouched (the
|
|
# READ-ONLY transaction downstream rejects it, not the injector).
|
|
out, injected = _inject_limit("INSERT INTO t VALUES (1)", 10)
|
|
assert injected is False
|
|
assert out == "INSERT INTO t VALUES (1)"
|
|
|
|
|
|
def test_union_query_accepts_limit():
|
|
out, injected = _inject_limit("SELECT 1 UNION SELECT 2", 10)
|
|
assert injected is True
|
|
ast = sqlglot.parse_one(out, read="postgres")
|
|
assert int(ast.args["limit"].expression.to_py()) == 11
|
|
|
|
|
|
def test_with_cte_query_accepts_limit():
|
|
sql = "WITH cte AS (SELECT 1) SELECT * FROM cte"
|
|
out, injected = _inject_limit(sql, 10)
|
|
assert injected is True
|
|
|
|
|
|
def test_limit_one_plus_n_for_truncation_detection():
|
|
# the whole point of +1: run_controlled fetches limit+1 rows, if it gets >
|
|
# limit it knows truncation happened. Verify the arithmetic for several limits.
|
|
for n in (1, 5, 100, 1000):
|
|
out, _ = _inject_limit("SELECT * FROM t", n)
|
|
ast = sqlglot.parse_one(out, read="postgres")
|
|
assert int(ast.args["limit"].expression.to_py()) == n + 1
|