test(harness): L1 characterization tests per Onda 0 (sqlcheck, ctetest, execute)
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).
This commit is contained in:
@@ -0,0 +1,133 @@
|
||||
"""L1: tht.ctetest — CTE file structure + test-record append-only ledger.
|
||||
|
||||
The CTE phase (F6) writes one .sql file per CTE block (WITH ... AS (...) only,
|
||||
NO trailing SELECT) and a cte_tests.json ledger of execution outcomes. These tests
|
||||
pin the structural helpers (has_trailing_select / last_cte_name / build_test_sql)
|
||||
and the ledger I/O (load/append, tolerant of JSON-array and JSONL formats).
|
||||
Pure logic, no DB.
|
||||
"""
|
||||
import json
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from tht.ctetest import (
|
||||
CteError,
|
||||
CteTestRecord,
|
||||
append_cte_test,
|
||||
build_test_sql,
|
||||
has_trailing_select,
|
||||
last_cte_name,
|
||||
load_cte_tests,
|
||||
)
|
||||
|
||||
# CTE file contents
|
||||
_CTE_OK = "WITH a AS (SELECT 1), b AS (SELECT * FROM a)"
|
||||
_CTE_WITH_SELECT = "WITH a AS (SELECT 1) SELECT * FROM a"
|
||||
_CTE_SYNTAX_GARBAGE = "WITH a AS (SELECT FROM WHERE"
|
||||
|
||||
|
||||
# --- has_trailing_select (semantics: True = file VIOLATES the rule) ----------
|
||||
|
||||
def test_correct_cte_file_has_no_trailing_select():
|
||||
# a correct CTE file has only WITH ... AS (...), no SELECT -> False
|
||||
assert has_trailing_select(_CTE_OK) is False
|
||||
|
||||
|
||||
def test_file_with_trailing_select_flagged():
|
||||
# a file that already has a SELECT final VIOLATES the rule -> True
|
||||
assert has_trailing_select(_CTE_WITH_SELECT) is True
|
||||
|
||||
|
||||
def test_unparsable_file_returns_false_no_hint():
|
||||
# garbage: no hint (can't tell), returns False
|
||||
assert has_trailing_select(_CTE_SYNTAX_GARBAGE) is False
|
||||
|
||||
|
||||
# --- last_cte_name ----------------------------------------------------------
|
||||
|
||||
def test_last_cte_name_of_two_cte_block():
|
||||
assert last_cte_name(_CTE_OK) == "b"
|
||||
|
||||
|
||||
def test_last_cte_name_single_cte():
|
||||
assert last_cte_name("WITH solo AS (SELECT 1)") == "solo"
|
||||
|
||||
|
||||
def test_last_cte_name_rejects_non_with():
|
||||
with pytest.raises(CteError, match="WITH"):
|
||||
last_cte_name("SELECT 1")
|
||||
|
||||
|
||||
def test_last_cte_name_rejects_unparsable():
|
||||
with pytest.raises(CteError, match="non parsabile"):
|
||||
last_cte_name("WITH a AS (SELECT FROM WHERE")
|
||||
|
||||
|
||||
# --- build_test_sql ---------------------------------------------------------
|
||||
|
||||
def test_build_test_sql_appends_select_from_last_cte():
|
||||
sql = build_test_sql(_CTE_OK)
|
||||
assert sql.endswith("SELECT * FROM b")
|
||||
assert sql.startswith("WITH a AS") # the WITH block preserved
|
||||
|
||||
|
||||
def test_build_test_sql_single_cte():
|
||||
sql = build_test_sql("WITH solo AS (SELECT 1)")
|
||||
assert sql.rstrip().endswith("SELECT * FROM solo")
|
||||
|
||||
|
||||
# --- ledger I/O: load_cte_tests / append_cte_test ---------------------------
|
||||
|
||||
def _record(**kw) -> CteTestRecord:
|
||||
base = dict(
|
||||
name="ablazione_q", ts=datetime(2025, 1, 1, 12, 0), sql_hash="abc123",
|
||||
status="ok", columns=["x"], row_sample=5, execution_ms=42,
|
||||
)
|
||||
base.update(kw)
|
||||
return CteTestRecord(**base)
|
||||
|
||||
|
||||
def test_load_empty_when_no_file(tmp_path):
|
||||
assert load_cte_tests(tmp_path) == []
|
||||
|
||||
|
||||
def test_append_then_load_roundtrip(tmp_path):
|
||||
append_cte_test(tmp_path, _record(name="a", status="ok"))
|
||||
append_cte_test(tmp_path, _record(name="b", status="error", error="boom"))
|
||||
loaded = load_cte_tests(tmp_path)
|
||||
assert len(loaded) == 2
|
||||
assert loaded[0].name == "a" and loaded[0].status == "ok"
|
||||
assert loaded[1].name == "b" and loaded[1].status == "error"
|
||||
assert loaded[1].error == "boom"
|
||||
|
||||
|
||||
def test_load_tolerates_json_array_format(tmp_path):
|
||||
# the ledger may be either JSONL (append-style) or a JSON array
|
||||
path = tmp_path / "cte_tests.json"
|
||||
path.write_text(json.dumps([
|
||||
{"name": "x", "ts": "2025-01-01T00:00:00", "sql_hash": "h", "status": "ok"},
|
||||
]))
|
||||
loaded = load_cte_tests(tmp_path)
|
||||
assert len(loaded) == 1
|
||||
assert loaded[0].name == "x"
|
||||
|
||||
|
||||
def test_load_tolerates_mixed_array_then_jsonl(tmp_path):
|
||||
# defensive: array + appended JSONL rows (the tolerant parser must handle both)
|
||||
path = tmp_path / "cte_tests.json"
|
||||
path.write_text(
|
||||
json.dumps([{"name": "x", "ts": "2025-01-01T00:00:00", "sql_hash": "h", "status": "ok"}])
|
||||
+ "\n"
|
||||
+ json.dumps({"name": "y", "ts": "2025-01-01T00:00:00", "sql_hash": "h2", "status": "ok"})
|
||||
+ "\n"
|
||||
)
|
||||
loaded = load_cte_tests(tmp_path)
|
||||
assert {r.name for r in loaded} == {"x", "y"}
|
||||
|
||||
|
||||
def test_corrupt_ledger_raises(tmp_path):
|
||||
(tmp_path / "cte_tests.json").write_text("{not valid json")
|
||||
with pytest.raises(CteError, match="corrotto"):
|
||||
load_cte_tests(tmp_path)
|
||||
@@ -0,0 +1,60 @@
|
||||
"""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
|
||||
@@ -0,0 +1,164 @@
|
||||
"""L1: tht.sqlcheck.validate_sql — static validation of reviewer-SQL (read-only, no DDL).
|
||||
|
||||
The gate runs generated SQL through validate_sql before it ever touches the DB.
|
||||
A bug here is dangerous (lets a DELETE through, or flags a valid SELECT). These
|
||||
tests pin the read-only contract + forbidden functions + object-existence checks.
|
||||
Pure logic, no DB.
|
||||
"""
|
||||
import pytest
|
||||
|
||||
from tht.mschema.models import (
|
||||
ColumnPhysical,
|
||||
PhysicalSchema,
|
||||
TablePhysical,
|
||||
)
|
||||
from tht.sqlcheck import CheckResult, validate_sql
|
||||
|
||||
|
||||
def _schema() -> PhysicalSchema:
|
||||
"""A tiny known schema for the object-existence checks."""
|
||||
from datetime import datetime
|
||||
|
||||
return PhysicalSchema(
|
||||
database="db",
|
||||
schema="dw",
|
||||
introspected_at=datetime(2025, 1, 1),
|
||||
tables={
|
||||
"dim_pazienti": TablePhysical(
|
||||
columns={
|
||||
"id_paziente": ColumnPhysical(type="bigint", pk=True),
|
||||
"citta": ColumnPhysical(type="varchar(100)"),
|
||||
}
|
||||
),
|
||||
"fct_ricoveri": TablePhysical(
|
||||
columns={
|
||||
"id_ricovero": ColumnPhysical(type="bigint", pk=True),
|
||||
"fk_paziente": ColumnPhysical(type="bigint"),
|
||||
},
|
||||
),
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
# --- parse + single-statement -----------------------------------------------
|
||||
|
||||
def test_valid_simple_select_ok():
|
||||
r = validate_sql("SELECT 1 AS x")
|
||||
assert r.ok
|
||||
assert r.errors == []
|
||||
|
||||
|
||||
def test_unparsable_sql_returns_clear_error():
|
||||
r = validate_sql("SELECT FROM WHERE")
|
||||
assert not r.ok
|
||||
assert any("non parsabile" in e for e in r.errors)
|
||||
|
||||
|
||||
def test_two_statements_rejected():
|
||||
r = validate_sql("SELECT 1; SELECT 2")
|
||||
assert not r.ok
|
||||
assert any("uno solo statement" in e for e in r.errors)
|
||||
|
||||
|
||||
# --- read-only enforcement (the load-bearing contract) -----------------------
|
||||
|
||||
@pytest.mark.parametrize("ddl", [
|
||||
"INSERT INTO t VALUES (1)",
|
||||
"UPDATE t SET x = 1",
|
||||
"DELETE FROM t",
|
||||
"CREATE TABLE t (x int)",
|
||||
"DROP TABLE t",
|
||||
"ALTER TABLE t ADD COLUMN x int",
|
||||
"TRUNCATE t",
|
||||
"GRANT SELECT ON t TO public",
|
||||
])
|
||||
def test_ddl_rejected(ddl):
|
||||
r = validate_sql(ddl)
|
||||
assert not r.ok
|
||||
assert any("read-only" in e or "SELECT" in e for e in r.errors)
|
||||
|
||||
|
||||
def test_with_select_ok():
|
||||
r = validate_sql("WITH cte AS (SELECT 1) SELECT * FROM cte")
|
||||
assert r.ok
|
||||
|
||||
|
||||
def test_union_ok():
|
||||
r = validate_sql("SELECT 1 UNION SELECT 2")
|
||||
assert r.ok
|
||||
|
||||
|
||||
# --- forbidden functions ----------------------------------------------------
|
||||
|
||||
def test_default_forbidden_function_rejected():
|
||||
# dblink is in the default blacklist (side-effect: outbound DB connection)
|
||||
r = validate_sql("SELECT dblink('host=x', 'SELECT 1')")
|
||||
assert not r.ok
|
||||
assert any("dblink" in e for e in r.errors)
|
||||
|
||||
|
||||
def test_custom_forbidden_function_set_used():
|
||||
# caller narrows the blacklist; regexp_match is an Anonymous function whose
|
||||
# .name resolves to "regexp_match", so the matcher catches it.
|
||||
r = validate_sql("SELECT regexp_match(col, 'x')", forbidden_functions={"regexp_match"})
|
||||
assert not r.ok
|
||||
assert any("regexp_match" in e for e in r.errors)
|
||||
|
||||
|
||||
def test_allowed_function_not_flagged():
|
||||
r = validate_sql("SELECT coalesce(NULL, 0)")
|
||||
assert r.ok
|
||||
|
||||
|
||||
def test_specialized_aggregate_not_caught_by_name_matcher():
|
||||
# KNOWN LIMITATION (documents real behavior): sqlglot models count/sum/etc. as
|
||||
# specialized Func nodes whose `.name` is the ARGUMENT ('*' for count(*)), not
|
||||
# the function name. So the name-based matcher in validate_sql does NOT flag
|
||||
# specialized aggregates even when listed. Only Anonymous functions (coalesce,
|
||||
# dblink, set_config...) are matched. If a customer needs to forbid count(), a
|
||||
# separate node-type check would be required -- not in scope here.
|
||||
r = validate_sql("SELECT count(*)", forbidden_functions={"count"})
|
||||
assert r.ok # NOT caught -- this is the documented limitation
|
||||
|
||||
|
||||
# --- object existence (when physical schema given) --------------------------
|
||||
|
||||
def test_nonexistent_table_flagged():
|
||||
r = validate_sql("SELECT * FROM non_esiste", physical=_schema())
|
||||
assert not r.ok
|
||||
assert any("inesistente" in e and "non_esiste" in e for e in r.errors)
|
||||
|
||||
|
||||
def test_existing_table_ok():
|
||||
r = validate_sql("SELECT * FROM dim_pazienti", physical=_schema())
|
||||
assert r.ok
|
||||
|
||||
|
||||
def test_cte_table_not_flagged_as_nonexistent():
|
||||
# a CTE is not a real table; must not be flagged as missing
|
||||
r = validate_sql("WITH cte AS (SELECT 1) SELECT * FROM cte", physical=_schema())
|
||||
assert r.ok
|
||||
|
||||
|
||||
def test_promoted_perimeter_warning():
|
||||
# dim_pazienti is NOT in the promoted perimeter -> warning (not error)
|
||||
r = validate_sql("SELECT * FROM dim_pazienti", physical=_schema(), promoted_tables={"fct_ricoveri"})
|
||||
assert r.ok # warning, not error
|
||||
assert any("fuori dal perimetro" in w for w in r.warnings)
|
||||
|
||||
|
||||
def test_nonexistent_column_flagged():
|
||||
r = validate_sql("SELECT p.colonna_inesistente FROM dim_pazienti p", physical=_schema())
|
||||
assert not r.ok
|
||||
assert any("colonna inesistente" in e for e in r.errors)
|
||||
|
||||
|
||||
def test_existing_column_with_alias_ok():
|
||||
r = validate_sql("SELECT p.citta FROM dim_pazienti p", physical=_schema())
|
||||
assert r.ok
|
||||
|
||||
|
||||
def test_no_physical_schema_skips_object_checks():
|
||||
# without physical, object existence is not checked (trusted context)
|
||||
r = validate_sql("SELECT * FROM anything")
|
||||
assert r.ok
|
||||
@@ -8,7 +8,7 @@ from pydantic import BaseModel
|
||||
from sqlglot import exp
|
||||
|
||||
CTE_TESTS_FILE = "cte_tests.json"
|
||||
PROBE = "\nSELECT * FROM __psd_probe__"
|
||||
PROBE = "\nSELECT * FROM __tht_probe__"
|
||||
|
||||
|
||||
class CteError(Exception):
|
||||
|
||||
Reference in New Issue
Block a user