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).
134 lines
4.3 KiB
Python
134 lines
4.3 KiB
Python
"""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)
|