311 lines
11 KiB
Python
311 lines
11 KiB
Python
"""Tests for `tht session documents --json` and build_documents."""
|
|
import json
|
|
from datetime import datetime
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.config import DatabaseConfig
|
|
from tht.decisions import DecisionRecord
|
|
from tht.session.models import SessionSnapshot
|
|
from tht.session.store import build_documents, build_snapshot_documents, create_session, new_session_manifest
|
|
from tht.cli.session_cmd import session_app
|
|
|
|
|
|
def _db():
|
|
return DatabaseConfig(
|
|
database="testdb", user="u", password="p", # noqa: S106
|
|
**{"schema": "public"},
|
|
)
|
|
|
|
|
|
def test_manifest_persists_workspace_revision():
|
|
manifest = new_session_manifest(
|
|
"q", _db(), workspace_id="psd-clinical", workspace_revision="a" * 40
|
|
)
|
|
|
|
assert manifest.workspace_id == "psd-clinical"
|
|
assert manifest.workspace_revision == "a" * 40
|
|
|
|
|
|
def test_build_documents_always_has_original_question(tmp_path):
|
|
m = create_session("quante ablazioni nel 2024", _db(), tmp_path)
|
|
docs = build_documents(m, tmp_path / m.id)
|
|
keys = [d["key"] for d in docs]
|
|
assert keys[0] == "question"
|
|
assert docs[0]["format"] == "text"
|
|
assert docs[0]["content"] == "quante ablazioni nel 2024"
|
|
# question.md is written by create_session -> revised_question present
|
|
assert "revised_question" in keys
|
|
|
|
|
|
def test_build_documents_includes_existing_artifacts_only(tmp_path):
|
|
m = create_session("q", _db(), tmp_path)
|
|
sdir = tmp_path / m.id
|
|
(sdir / "sql_final.sql").write_text("SELECT 1")
|
|
(sdir / "schema_linking.json").write_text('{"question":"q","candidates":[]}')
|
|
docs = {d["key"]: d for d in build_documents(m, sdir)}
|
|
assert docs["sql"]["format"] == "sql"
|
|
assert docs["sql"]["content"] == "SELECT 1"
|
|
assert docs["schema_linking"]["format"] == "schema-linking"
|
|
assert "validation_report" not in docs # not written
|
|
|
|
|
|
def _decision(seq: int, type_: str, subject: str, detail: str = "", rationale: str = ""):
|
|
return DecisionRecord(
|
|
seq=seq,
|
|
ts=datetime(2026, 1, 1),
|
|
type=type_,
|
|
subject=subject,
|
|
detail=detail,
|
|
rationale=rationale,
|
|
)
|
|
|
|
|
|
def _rich_session(tmp_path):
|
|
manifest = create_session("domanda originale", _db(), tmp_path)
|
|
session_dir = tmp_path / manifest.id
|
|
question = (
|
|
"# Domanda\n\n"
|
|
"Conta i pazienti con **ablazione**.\n\n"
|
|
"## Assunzioni\n\n"
|
|
"- Considera solo gli adulti\n"
|
|
"- Usa l'ultima procedura disponibile\n"
|
|
)
|
|
report = (
|
|
"# Validation report\n\n"
|
|
"## Parsing e read-only\n\n- parsing: OK\n\n"
|
|
"## Preview / aggregato\n\n"
|
|
"- tempo di esecuzione: 12 ms (limit 10)\n\n"
|
|
"| count |\n|---|\n| 42 |\n\n"
|
|
"## Note del reviewer\n\nRisultato verificato.\n"
|
|
)
|
|
decisions = [
|
|
_decision(1, "concept_clarified", "paziente attivo", "`flag_attivo = TRUE`"),
|
|
_decision(2, "memory_promotion_declined", "paziente attivo", "seq:1"),
|
|
_decision(
|
|
3,
|
|
"concept_clarified",
|
|
"ablazione valida",
|
|
"Procedura con stato **completato**.",
|
|
"Definizione approvata dal reviewer.",
|
|
),
|
|
_decision(4, "memory_promoted", "ablazione valida", "seq:3"),
|
|
_decision(5, "phase_approved", "phase:1"),
|
|
_decision(6, "table_promoted", "fact_procedure"),
|
|
_decision(7, "column_promoted", "fact_procedure.status"),
|
|
_decision(8, "column_excluded", "fact_procedure.note", "Non pertinente"),
|
|
_decision(9, "cte_approved", "eventi_device"),
|
|
_decision(10, "sql_approved", "phase:7"),
|
|
_decision(11, "datamart_requested", "phase:8"),
|
|
_decision(12, "datamart_declined", "phase:8"),
|
|
]
|
|
(session_dir / "question.md").write_text(question)
|
|
(session_dir / "sql_final.sql").write_text("SELECT COUNT(*) FROM fact_procedure")
|
|
(session_dir / "validation_report.md").write_text(report)
|
|
(session_dir / "schema_linking.json").write_text(
|
|
'{"question":"q","candidates":[],"joins":[],"excluded":[],"open_questions":[]}'
|
|
)
|
|
(session_dir / "review_decisions.jsonl").write_text(
|
|
"\n".join(d.model_dump_json() for d in decisions) + "\n"
|
|
)
|
|
return manifest, session_dir, question, report, decisions
|
|
|
|
|
|
def test_build_documents_projects_the_approved_summary_order_and_sections(tmp_path):
|
|
manifest, session_dir, _question, _report, _decisions = _rich_session(tmp_path)
|
|
|
|
docs = build_documents(manifest, session_dir)
|
|
|
|
assert [doc["key"] for doc in docs] == [
|
|
"question",
|
|
"sql",
|
|
"preview",
|
|
"revised_question",
|
|
"assumptions",
|
|
"memories",
|
|
"schema_linking",
|
|
"validation_report",
|
|
"decisions",
|
|
]
|
|
by_key = {doc["key"]: doc for doc in docs}
|
|
assert by_key["preview"]["format"] == "markdown"
|
|
assert "tempo di esecuzione: 12 ms" in by_key["preview"]["content"]
|
|
assert "| 42 |" in by_key["preview"]["content"]
|
|
assert "Parsing e read-only" not in by_key["preview"]["content"]
|
|
assert by_key["revised_question"]["content"] == (
|
|
"# Domanda\n\nConta i pazienti con **ablazione**."
|
|
)
|
|
assert by_key["assumptions"]["content"] == (
|
|
"## Assunzioni\n\n- Considera solo gli adulti\n"
|
|
"- Usa l'ultima procedura disponibile"
|
|
)
|
|
assert "Preview / aggregato" not in by_key["validation_report"]["content"]
|
|
assert "Parsing e read-only" in by_key["validation_report"]["content"]
|
|
assert "Note del reviewer" in by_key["validation_report"]["content"]
|
|
|
|
|
|
def test_build_documents_groups_memories_and_filters_summary_decisions(tmp_path):
|
|
manifest, session_dir, _question, _report, _decisions = _rich_session(tmp_path)
|
|
|
|
by_key = {doc["key"]: doc for doc in build_documents(manifest, session_dir)}
|
|
memories = json.loads(by_key["memories"]["content"])
|
|
|
|
assert memories == [
|
|
{
|
|
"status": "approved",
|
|
"subject": "ablazione valida",
|
|
"detail": "Procedura con stato **completato**.",
|
|
"rationale": "Definizione approvata dal reviewer.",
|
|
},
|
|
{
|
|
"status": "declined",
|
|
"subject": "paziente attivo",
|
|
"detail": "`flag_attivo = TRUE`",
|
|
"rationale": "",
|
|
},
|
|
]
|
|
decisions = by_key["decisions"]["content"]
|
|
assert "column_excluded" in decisions
|
|
for hidden in (
|
|
"concept_clarified",
|
|
"memory_promoted",
|
|
"memory_promotion_declined",
|
|
"phase_approved",
|
|
"table_promoted",
|
|
"column_promoted",
|
|
"cte_approved",
|
|
"sql_approved",
|
|
"datamart_requested",
|
|
"datamart_declined",
|
|
):
|
|
assert hidden not in decisions
|
|
|
|
|
|
def test_build_documents_keeps_validation_report_when_it_has_no_preview(tmp_path):
|
|
manifest = create_session("q", _db(), tmp_path)
|
|
session_dir = tmp_path / manifest.id
|
|
report = "# Validation report\n\n## Parsing\n\n- OK\n"
|
|
(session_dir / "validation_report.md").write_text(report)
|
|
|
|
by_key = {doc["key"]: doc for doc in build_documents(manifest, session_dir)}
|
|
|
|
assert "preview" not in by_key
|
|
assert by_key["validation_report"]["content"] == report.strip()
|
|
|
|
|
|
def test_memory_projection_does_not_repeat_detail_as_marker_rationale(tmp_path):
|
|
manifest = create_session("q", _db(), tmp_path)
|
|
session_dir = tmp_path / manifest.id
|
|
decisions = [
|
|
_decision(1, "concept_clarified", "termine", "Definizione in **Markdown**."),
|
|
_decision(
|
|
2,
|
|
"memory_promoted",
|
|
"termine",
|
|
"seq:1",
|
|
"Definizione in **Markdown**.",
|
|
),
|
|
]
|
|
(session_dir / "review_decisions.jsonl").write_text(
|
|
"\n".join(d.model_dump_json() for d in decisions) + "\n"
|
|
)
|
|
|
|
by_key = {doc["key"]: doc for doc in build_documents(manifest, session_dir)}
|
|
|
|
assert json.loads(by_key["memories"]["content"])[0] == {
|
|
"status": "approved",
|
|
"subject": "termine",
|
|
"detail": "Definizione in **Markdown**.",
|
|
"rationale": "",
|
|
}
|
|
|
|
|
|
def test_memory_projection_excludes_standalone_schema_tables_but_keeps_mappings(tmp_path):
|
|
manifest = create_session("q", _db(), tmp_path)
|
|
session_dir = tmp_path / manifest.id
|
|
decisions = [
|
|
_decision(1, "table_promoted", "fact_procedure", "Tabella dei fatti"),
|
|
_decision(2, "memory_promoted", "fact_procedure", "seq:1"),
|
|
_decision(3, "table_excluded", "dim_time", "Dimensione temporale"),
|
|
_decision(4, "memory_promotion_declined", "dim_time", "Dimensione temporale"),
|
|
_decision(
|
|
5,
|
|
"concept_clarified",
|
|
"ablazione valida",
|
|
"Mappata su `fact_procedure.ablazione_transcatetere = TRUE`.",
|
|
),
|
|
_decision(6, "memory_promoted", "ablazione valida", "seq:5"),
|
|
_decision(
|
|
7,
|
|
"memory_promotion_declined",
|
|
"conteggio annuale",
|
|
"`COUNT(DISTINCT patient_id)` raggruppato per `dim_time.year`.",
|
|
),
|
|
]
|
|
(session_dir / "schema_linking.json").write_text(json.dumps({
|
|
"question": "q",
|
|
"candidates": [{"kind": "table", "name": "fact_procedure"}],
|
|
"joins": [],
|
|
"excluded": [{"kind": "table", "name": "dim_time"}],
|
|
"open_questions": [],
|
|
}))
|
|
(session_dir / "review_decisions.jsonl").write_text(
|
|
"\n".join(d.model_dump_json() for d in decisions) + "\n"
|
|
)
|
|
|
|
by_key = {doc["key"]: doc for doc in build_documents(manifest, session_dir)}
|
|
|
|
assert json.loads(by_key["memories"]["content"]) == [
|
|
{
|
|
"status": "approved",
|
|
"subject": "ablazione valida",
|
|
"detail": "Mappata su `fact_procedure.ablazione_transcatetere = TRUE`.",
|
|
"rationale": "",
|
|
},
|
|
{
|
|
"status": "declined",
|
|
"subject": "conteggio annuale",
|
|
"detail": "`COUNT(DISTINCT patient_id)` raggruppato per `dim_time.year`.",
|
|
"rationale": "",
|
|
},
|
|
]
|
|
|
|
|
|
def test_filesystem_and_snapshot_builders_share_the_same_projection(tmp_path):
|
|
manifest, session_dir, question, report, decisions = _rich_session(tmp_path)
|
|
snapshot = SessionSnapshot(
|
|
manifest=manifest,
|
|
artifacts={
|
|
"question": question,
|
|
"schema_linking": (session_dir / "schema_linking.json").read_text(),
|
|
"sql_final": (session_dir / "sql_final.sql").read_text(),
|
|
"validation_report": report,
|
|
},
|
|
decisions=decisions,
|
|
)
|
|
|
|
assert build_snapshot_documents(snapshot) == build_documents(manifest, session_dir)
|
|
|
|
|
|
def test_cli_documents_json(tmp_path, monkeypatch):
|
|
m = create_session("q", _db(), tmp_path)
|
|
from tht.cli import session_cmd
|
|
|
|
# The CLI resolves its local principal through THT_HOME. Keep this test hermetic instead
|
|
# of changing the developer's real identity directory while exercising JSON output.
|
|
monkeypatch.setenv("THT_HOME", str(tmp_path / "thoth-home"))
|
|
|
|
class FakePaths:
|
|
sessions = tmp_path
|
|
|
|
class FakeCfg:
|
|
paths = FakePaths()
|
|
database = _db()
|
|
|
|
monkeypatch.setattr(session_cmd, "_load_config_or_exit", lambda _: FakeCfg())
|
|
res = CliRunner().invoke(session_app, ["documents", m.id, "--json"])
|
|
assert res.exit_code == 0, res.output
|
|
docs = json.loads(res.output)
|
|
assert isinstance(docs, list)
|
|
assert docs[0]["key"] == "question"
|