294 lines
10 KiB
Python
294 lines
10 KiB
Python
import hashlib
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path
|
|
|
|
import typer
|
|
|
|
from tht.cli.config_cmd import CONFIG_OPT
|
|
from tht.cli.schema_cmd import _load_config_or_exit
|
|
from tht.cli.session_cmd import load_session_or_exit, load_snapshot_or_exit, session_repository
|
|
from tht.cli.sql_cmd import promoted_tables_for, require_action
|
|
|
|
_RULE6_HINT = (
|
|
" — il file del CTE deve contenere SOLO il blocco WITH ... AS (...), "
|
|
"senza SELECT finale: la SELECT la aggiunge tht in fase di test."
|
|
)
|
|
|
|
cte_app = typer.Typer(help="Test controllato dei CTE proposti (Agent View Generation)")
|
|
|
|
|
|
@cte_app.command("save")
|
|
def save_cmd(
|
|
session: str = typer.Option(..., "--session"),
|
|
name: str = typer.Option(..., "--name"),
|
|
file: str = typer.Option(..., "--file", help="File SQL, oppure '-' per stdin."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Persist one CTE SQL block through the configured session repository."""
|
|
import sys
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session)
|
|
raw = sys.stdin.read() if file == "-" else Path(file).read_text()
|
|
try:
|
|
session_repository(cfg).write_artifact(session, f"cte_sql:{name}", raw)
|
|
except ValueError as exc:
|
|
typer.secho(f"ERRORE: {exc}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1) from None
|
|
typer.secho(f"OK: CTE {name} salvato.", fg=typer.colors.GREEN)
|
|
|
|
|
|
@cte_app.command("test")
|
|
def test_cmd(
|
|
name: str = typer.Argument(..., help="Nome del CTE (file sessions/<id>/ctes/<nome>.sql)."),
|
|
session: str = typer.Option(..., "--session"),
|
|
json_out: bool = typer.Option(False, "--json", help="Emette l'esito come JSON (per Pi)."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Valida e testa un CTE: wrap su ultimo CTE + esecuzione nelle 4 reti."""
|
|
from rich.console import Console
|
|
from rich.table import Table
|
|
|
|
from tht.cli.sql_cmd import do_run
|
|
from tht.ctetest import (
|
|
CteError,
|
|
CteTestRecord,
|
|
_jsonable,
|
|
append_cte_test_snapshot,
|
|
build_test_sql,
|
|
has_trailing_select,
|
|
)
|
|
from tht.execute import ExecutionError
|
|
from tht.execute.warnings import runtime_warnings
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
require_action(cfg, "cte_test")
|
|
load_session_or_exit(cfg, session)
|
|
snapshot = load_snapshot_or_exit(cfg, session)
|
|
|
|
from tht.cli.phase_cmd import require_phase_or_exit
|
|
from tht.phase import next_cte
|
|
|
|
require_phase_or_exit(cfg, session, 6)
|
|
nxt = next_cte(snapshot)
|
|
if nxt is not None and name != nxt:
|
|
typer.secho(
|
|
f"ERRORE: ordine CTE. Ora tocca a '{nxt}' (il primo CTE del piano non "
|
|
f"ancora approvato), non a '{name}'. Testa '{nxt}', poi falla approvare "
|
|
f"dal reviewer (decisione cte_approved); solo allora potrai testare il "
|
|
f"CTE successivo.",
|
|
fg=typer.colors.RED, err=True,
|
|
)
|
|
raise typer.Exit(code=5)
|
|
|
|
cte_sql = snapshot.artifacts.get(f"cte_sql:{name}")
|
|
if cte_sql is None:
|
|
typer.secho(f"ERRORE: file CTE non trovato: {name}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
sql_hash = hashlib.sha256(cte_sql.encode()).hexdigest()
|
|
|
|
def _record_error(message: str) -> None:
|
|
record = CteTestRecord(
|
|
name=name, ts=datetime.now(UTC), sql_hash=sql_hash,
|
|
status="error", error=message,
|
|
)
|
|
append_cte_test_snapshot(session_repository(cfg), snapshot, record)
|
|
if json_out:
|
|
typer.echo(record.model_dump_json())
|
|
else:
|
|
typer.secho(f"ERRORE: {message}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
|
|
try:
|
|
test_sql = build_test_sql(cte_sql)
|
|
except CteError as e:
|
|
msg = str(e)
|
|
if "non parsabile" in msg and has_trailing_select(cte_sql):
|
|
msg += _RULE6_HINT
|
|
_record_error(msg)
|
|
|
|
# validazione statica (esce con messaggi propri se invalida; qui la
|
|
# intercettiamo per registrare comunque l'esito nel cte_tests.json)
|
|
from tht.cli.sql_cmd import _load_physical_or_exit
|
|
from tht.sqlcheck import validate_sql
|
|
|
|
check = validate_sql(
|
|
test_sql,
|
|
physical=_load_physical_or_exit(cfg),
|
|
promoted_tables=promoted_tables_for(cfg, session),
|
|
forbidden_functions=set(cfg.execution.forbidden_functions),
|
|
)
|
|
for w in check.warnings:
|
|
typer.secho(f" warning: {w}", fg=typer.colors.YELLOW)
|
|
if not check.ok:
|
|
msg = "; ".join(check.errors)
|
|
if any("statement" in e for e in check.errors):
|
|
msg += _RULE6_HINT
|
|
_record_error(msg)
|
|
|
|
try:
|
|
result = do_run(cfg, test_sql, limit=cfg.execution.max_preview_rows)
|
|
except ExecutionError as e:
|
|
_record_error(str(e))
|
|
|
|
warnings = runtime_warnings(result, cfg.execution)
|
|
record = CteTestRecord(
|
|
name=name, ts=datetime.now(UTC), sql_hash=sql_hash, status="ok",
|
|
columns=result.columns, row_sample=len(result.rows),
|
|
execution_ms=result.execution_ms, warnings=warnings,
|
|
preview_rows=[[_jsonable(cell) for cell in row] for row in result.rows],
|
|
)
|
|
append_cte_test_snapshot(session_repository(cfg), snapshot, record)
|
|
|
|
if json_out:
|
|
typer.echo(record.model_dump_json())
|
|
return
|
|
|
|
typer.echo(f"CTE {name} — {result.execution_ms} ms")
|
|
table = Table()
|
|
for col in result.columns:
|
|
table.add_column(col)
|
|
for row in result.rows:
|
|
table.add_row(*[str(v) for v in row])
|
|
Console().print(table)
|
|
for w in warnings:
|
|
typer.secho(f" warning: {w}", fg=typer.colors.YELLOW)
|
|
typer.secho("OK: esito registrato in cte_tests.json", fg=typer.colors.GREEN)
|
|
|
|
|
|
CTE_PLAN_DOC_FILE = "cte_plan_doc.json"
|
|
|
|
|
|
@cte_app.command("plan")
|
|
def plan_cmd(
|
|
session: str = typer.Option(..., "--session"),
|
|
name: list[str] = typer.Option(..., "--name", help="Nome CTE (ripetibile, in ordine)."),
|
|
doc: str = typer.Option(
|
|
None, "--doc",
|
|
help="Path al JSON con la doc della catena (payload A, campo ctes[]), oppure '-' per stdin."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Persiste il piano CTE ordinato (sessions/<id>/cte_plan.json).
|
|
|
|
Con --doc, valida e persiste anche sessions/<id>/cte_plan_doc.json (i nomi in
|
|
doc["ctes"] devono combaciare con --name, stesso ordine); cte_plan.json resta
|
|
invariato (list[str])."""
|
|
import json
|
|
import sys
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session)
|
|
from tht.cli.phase_cmd import require_phase_or_exit
|
|
require_phase_or_exit(cfg, session, 6)
|
|
|
|
doc_data = None
|
|
if doc is not None:
|
|
raw = sys.stdin.read() if doc == "-" else Path(doc).read_text()
|
|
try:
|
|
doc_data = json.loads(raw)
|
|
except json.JSONDecodeError as e:
|
|
typer.secho(f"ERRORE: --doc JSON non valido: {e}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
doc_names = [c["name"] for c in doc_data["ctes"]]
|
|
if doc_names != name:
|
|
typer.secho(
|
|
f"ERRORE: i nomi in --doc ({doc_names}) non combaciano con --name ({name}).",
|
|
fg=typer.colors.RED, err=True,
|
|
)
|
|
raise typer.Exit(code=1)
|
|
|
|
repository = session_repository(cfg)
|
|
repository.write_artifact(session, "cte_plan", json.dumps(name, ensure_ascii=False))
|
|
if doc_data is not None:
|
|
repository.write_artifact(session, "cte_plan_doc", json.dumps(doc_data, ensure_ascii=False))
|
|
typer.secho(f"OK: piano CTE salvato ({len(name)} CTE).",
|
|
fg=typer.colors.GREEN)
|
|
|
|
|
|
@cte_app.command("next")
|
|
def next_cmd(
|
|
session: str = typer.Option(..., "--session"),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Primo CTE del piano non ancora approvato (stdout pulito; vuoto se nessuno).
|
|
|
|
Sorgente unica dell'ordine di approvazione CTE (F6): il gate lo usa per
|
|
registrare cte_approved sul NOME del CTE (non su 'phase:6')."""
|
|
from tht.phase import next_cte
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session)
|
|
nxt = next_cte(load_snapshot_or_exit(cfg, session))
|
|
if nxt:
|
|
typer.echo(nxt)
|
|
|
|
|
|
@cte_app.command("info")
|
|
def info_cmd(
|
|
name: str = typer.Argument(..., help="Nome del CTE (deve comparire in cte_plan.json)."),
|
|
session: str = typer.Option(..., "--session"),
|
|
json_out: bool = typer.Option(False, "--json", help="Emetti JSON puro su stdout."),
|
|
config: Path = CONFIG_OPT,
|
|
) -> None:
|
|
"""Dettaglio READ-ONLY di un CTE: posizione nel piano, SQL persistito, doc e ultimo test.
|
|
|
|
Fonte dati deterministica per il gate JS (WS2) che costruisce il payload cte_result v2."""
|
|
import json as _json
|
|
|
|
from tht.phase import CTE_PLAN_FILE, approved_ctes, cte_plan
|
|
|
|
cfg = _load_config_or_exit(config)
|
|
load_session_or_exit(cfg, session)
|
|
snapshot = load_snapshot_or_exit(cfg, session)
|
|
|
|
plan = cte_plan(snapshot)
|
|
if not plan:
|
|
typer.secho(f"ERRORE: {CTE_PLAN_FILE} assente o vuoto per la sessione '{session}'.",
|
|
fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
if name not in plan:
|
|
typer.secho(f"ERRORE: '{name}' non e' nel piano CTE ({plan}).",
|
|
fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
|
|
cte_sql = snapshot.artifacts.get(f"cte_sql:{name}")
|
|
if cte_sql is None:
|
|
typer.secho(f"ERRORE: file CTE non trovato: {name}", fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
|
|
doc = None
|
|
raw_doc = snapshot.artifacts.get("cte_plan_doc")
|
|
if raw_doc:
|
|
full_doc = _json.loads(raw_doc)
|
|
for c in full_doc.get("ctes", []):
|
|
if c.get("name") == name:
|
|
doc = {k: v for k, v in c.items() if k != "name"}
|
|
break
|
|
|
|
from tht.ctetest import CteError, load_cte_tests_text
|
|
|
|
try:
|
|
records = [r for r in load_cte_tests_text(snapshot.artifacts.get("cte_tests", "")) if r.name == name]
|
|
except CteError as e:
|
|
typer.secho(f"ERRORE: impossibile leggere cte_tests.json: {e}",
|
|
fg=typer.colors.RED, err=True)
|
|
raise typer.Exit(code=1)
|
|
last_test = _json.loads(records[-1].model_dump_json()) if records else None
|
|
|
|
payload = {
|
|
"name": name,
|
|
"index": plan.index(name) + 1,
|
|
"total": len(plan),
|
|
"plan": plan,
|
|
"sql": cte_sql,
|
|
"approved": name in approved_ctes(snapshot),
|
|
"doc": doc,
|
|
"last_test": last_test,
|
|
}
|
|
|
|
if json_out:
|
|
typer.echo(_json.dumps(payload, ensure_ascii=False))
|
|
return
|
|
typer.echo(f"{name}: {payload['index']}/{payload['total']} — "
|
|
f"approvato={payload['approved']}")
|