From 31552782c03e9779a7a4be8e6d9e8fcc57827170 Mon Sep 17 00:00:00 2001 From: mptyl Date: Sat, 27 Jun 2026 13:04:34 +0200 Subject: [PATCH] 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). --- harness/tests/test_ctetest.py | 133 +++++++++++++++++ harness/tests/test_execute_inject_limit.py | 60 ++++++++ harness/tests/test_sqlcheck.py | 164 +++++++++++++++++++++ harness/tht/ctetest.py | 2 +- 4 files changed, 358 insertions(+), 1 deletion(-) create mode 100644 harness/tests/test_ctetest.py create mode 100644 harness/tests/test_execute_inject_limit.py create mode 100644 harness/tests/test_sqlcheck.py diff --git a/harness/tests/test_ctetest.py b/harness/tests/test_ctetest.py new file mode 100644 index 00000000..3c98c422 --- /dev/null +++ b/harness/tests/test_ctetest.py @@ -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) diff --git a/harness/tests/test_execute_inject_limit.py b/harness/tests/test_execute_inject_limit.py new file mode 100644 index 00000000..00c905be --- /dev/null +++ b/harness/tests/test_execute_inject_limit.py @@ -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 diff --git a/harness/tests/test_sqlcheck.py b/harness/tests/test_sqlcheck.py new file mode 100644 index 00000000..e6e44832 --- /dev/null +++ b/harness/tests/test_sqlcheck.py @@ -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 diff --git a/harness/tht/ctetest.py b/harness/tht/ctetest.py index 0c9e5ee3..40bdb63d 100644 --- a/harness/tht/ctetest.py +++ b/harness/tht/ctetest.py @@ -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):