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).
102 lines
3.5 KiB
Python
102 lines
3.5 KiB
Python
import json
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Literal
|
|
|
|
import sqlglot
|
|
from pydantic import BaseModel
|
|
from sqlglot import exp
|
|
|
|
CTE_TESTS_FILE = "cte_tests.json"
|
|
PROBE = "\nSELECT * FROM __tht_probe__"
|
|
|
|
|
|
class CteError(Exception):
|
|
pass
|
|
|
|
|
|
class CteTestRecord(BaseModel):
|
|
name: str
|
|
ts: datetime
|
|
sql_hash: str
|
|
status: Literal["ok", "error"]
|
|
columns: list[str] = []
|
|
row_sample: int = 0
|
|
execution_ms: int = 0
|
|
warnings: list[str] = []
|
|
error: str | None = None
|
|
|
|
|
|
def has_trailing_select(cte_sql: str) -> bool:
|
|
"""True se il file, parsato da solo, e' gia' una query completa con SELECT
|
|
finale dopo il blocco WITH (violazione della regola: il file deve contenere
|
|
solo WITH ... AS (...)).
|
|
|
|
I file CTE corretti contengono solo 'WITH ... AS (...)': sqlglot.parse_one()
|
|
fallisce su di loro perche' manca la SELECT obbligatoria.
|
|
Un file con SELECT finale parsa invece correttamente come Select con un nodo With.
|
|
I file con errori di sintassi falliscono parse_one e ritornano False (no hint)."""
|
|
try:
|
|
ast = sqlglot.parse_one(cte_sql, read="postgres")
|
|
except sqlglot.errors.ParseError:
|
|
return False
|
|
return ast is not None and ast.find(exp.With) is not None
|
|
|
|
|
|
def last_cte_name(cte_sql: str) -> str:
|
|
"""Nome dell'ultimo CTE di un blocco 'WITH a AS (...), b AS (...)' senza SELECT
|
|
finale. Trucco AV-SQL: si appende una SELECT fittizia per rendere parsabile."""
|
|
if not cte_sql.lstrip().upper().startswith("WITH"):
|
|
raise CteError("il file deve contenere un blocco WITH ... AS (...)")
|
|
try:
|
|
ast = sqlglot.parse_one(cte_sql + PROBE, read="postgres")
|
|
except sqlglot.errors.ParseError as e:
|
|
raise CteError(f"CTE non parsabile: {e}") from e
|
|
with_clause = ast.find(exp.With)
|
|
if with_clause is None or not with_clause.expressions:
|
|
raise CteError("il file deve contenere un blocco WITH ... AS (...)")
|
|
return with_clause.expressions[-1].alias
|
|
|
|
|
|
def build_test_sql(cte_sql: str) -> str:
|
|
"""SQL completo per il test: il blocco WITH + SELECT * dall'ultimo CTE.
|
|
Il LIMIT lo inietta run_controlled."""
|
|
name = last_cte_name(cte_sql)
|
|
return f"{cte_sql.rstrip().rstrip(',')}\nSELECT * FROM {name}"
|
|
|
|
|
|
def _iter_json_objects(text: str):
|
|
"""Estrae gli oggetti JSON di primo livello da `text`, qualunque sia il
|
|
formato: array JSON, JSONL, o misto (array + righe JSONL appese). Salta
|
|
spazi, virgole e parentesi d'array fra un oggetto e il successivo."""
|
|
dec = json.JSONDecoder()
|
|
i, n = 0, len(text)
|
|
while i < n:
|
|
if text[i] in " \t\r\n,[]":
|
|
i += 1
|
|
continue
|
|
obj, end = dec.raw_decode(text, i)
|
|
yield obj
|
|
i = end
|
|
|
|
|
|
def load_cte_tests(session_dir: Path) -> list[CteTestRecord]:
|
|
path = session_dir / CTE_TESTS_FILE
|
|
if not path.exists():
|
|
return []
|
|
text = path.read_text()
|
|
try:
|
|
objs = list(_iter_json_objects(text))
|
|
except json.JSONDecodeError as e:
|
|
raise CteError(f"{CTE_TESTS_FILE} corrotto: {e}") from e
|
|
return [CteTestRecord.model_validate(o) for o in objs]
|
|
|
|
|
|
def append_cte_test(session_dir: Path, record: CteTestRecord) -> None:
|
|
"""Append atomico (una riga JSON per esito): scritture concorrenti sulla
|
|
stessa sessione non si sovrascrivono, a differenza del rewrite dell'array."""
|
|
path = session_dir / CTE_TESTS_FILE
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("a", encoding="utf-8") as f:
|
|
f.write(record.model_dump_json() + "\n")
|