56 lines
2.3 KiB
Python
56 lines
2.3 KiB
Python
"""L1: build del record solved_question dagli artefatti persistiti della sessione.
|
|
|
|
Il record si costruisce SOLO da cio' che il workflow ha approvato: sql_final.sql
|
|
presente + decisione sql_approved nella vista effective; la domanda e' l'ultima
|
|
question_rewritten (fallback: la domanda del manifest)."""
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from tht.decisions import append_decision
|
|
from tht.session.models import SessionManifest
|
|
from tht.solved import SolvedIndexError, build_solved_record
|
|
|
|
|
|
def _manifest() -> SessionManifest:
|
|
return SessionManifest(
|
|
id="s1", created_at=datetime(2026, 1, 1), question="domanda originale",
|
|
database="db", schema="public",
|
|
)
|
|
|
|
|
|
def test_build_uses_rewritten_question_sql_and_tables(tmp_path):
|
|
(tmp_path / "sql_final.sql").write_text("SELECT 1\n")
|
|
append_decision(tmp_path, type="question_rewritten", subject="domanda",
|
|
detail="domanda riscritta esplicita")
|
|
append_decision(tmp_path, type="sql_approved", subject="phase:7")
|
|
for n in range(1, 8):
|
|
append_decision(tmp_path, type="phase_approved", subject=f"phase:{n}")
|
|
rec = build_solved_record(tmp_path, _manifest(), {"fact_x", "dim_y"})
|
|
assert rec.id == "solved:s1"
|
|
assert rec.content == "domanda riscritta esplicita"
|
|
assert rec.metadata["sql"] == "SELECT 1"
|
|
assert rec.metadata["tables"] == ["dim_y", "fact_x"] # ordinate
|
|
|
|
|
|
def test_build_falls_back_to_manifest_question(tmp_path):
|
|
(tmp_path / "sql_final.sql").write_text("SELECT 1")
|
|
append_decision(tmp_path, type="sql_approved", subject="phase:7")
|
|
for n in range(1, 8):
|
|
append_decision(tmp_path, type="phase_approved", subject=f"phase:{n}")
|
|
rec = build_solved_record(tmp_path, _manifest(), None)
|
|
assert rec.content == "domanda originale"
|
|
assert rec.metadata["tables"] == []
|
|
|
|
|
|
def test_build_requires_sql_file(tmp_path):
|
|
append_decision(tmp_path, type="sql_approved", subject="phase:7")
|
|
with pytest.raises(SolvedIndexError, match="sql_final.sql"):
|
|
build_solved_record(tmp_path, _manifest(), None)
|
|
|
|
|
|
def test_build_requires_sql_approved(tmp_path):
|
|
(tmp_path / "sql_final.sql").write_text("SELECT 1")
|
|
with pytest.raises(SolvedIndexError, match="sql_approved"):
|
|
build_solved_record(tmp_path, _manifest(), None)
|