marcopan
9b4f6b9804
WS1 of review-gates-v2: gives the JS gate (WS2) deterministic data to build the cte_result v2 payload. - CteTestRecord gains optional preview_rows (JSON-coerced, truncated cells); test_cmd populates it from the bounded result rows. - New read-only `tht cte info <name> --session <id> [--json]`: plan index/total, persisted .sql, cte_plan_doc.json entry (if any), last CteTestRecord. Exits 1 with a clean stderr message on missing session/plan/name/sql. - `tht cte plan --doc -` validates a chain-doc JSON (ctes[].name must match --name, same order) and writes it to cte_plan_doc.json; cte_plan.json stays a plain list[str] (load-bearing for tht.phase.next_cte). --doc is optional.
120 lines
4.2 KiB
Python
120 lines
4.2 KiB
Python
"""L1: `tht cte plan --doc -` — persists the cte_plan_doc.json chain documentation.
|
|
|
|
cte_plan.json (list[str], load-bearing for tht.phase.next_cte) stays untouched; the
|
|
doc (payload-A shape: {"ctes": [{"name", "purpose", "rationale", "depends_on", ...}]})
|
|
is validated (names must match --name, same order) and written verbatim to a
|
|
separate sessions/<id>/cte_plan_doc.json. --doc is optional; without it, behavior
|
|
is unchanged (existing test_cte_next.py covers the no-doc path).
|
|
"""
|
|
import json
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli.cte_cmd import cte_app
|
|
from tht.config import DatabaseConfig
|
|
from tht.decisions import append_decision
|
|
from tht.session.store import create_session
|
|
|
|
|
|
def _db():
|
|
return DatabaseConfig(database="testdb", user="u", password="p", **{"schema": "public"}) # noqa: S106
|
|
|
|
|
|
def _patch_cfg(monkeypatch, tmp_path):
|
|
from tht.cli import cte_cmd
|
|
|
|
class FakePaths:
|
|
sessions = tmp_path
|
|
|
|
class FakeCfg:
|
|
paths = FakePaths()
|
|
database = _db()
|
|
|
|
monkeypatch.setattr(cte_cmd, "_load_config_or_exit", lambda _: FakeCfg())
|
|
|
|
|
|
def _advance_to_phase_6(sdir):
|
|
"""plan_cmd requires phase >= 6: approve phases 1..5 in order."""
|
|
for n in range(1, 6):
|
|
append_decision(sdir, type="phase_approved", subject=f"phase:{n}")
|
|
|
|
|
|
def test_plan_with_matching_doc_writes_cte_plan_doc_json(tmp_path, monkeypatch):
|
|
m = create_session("q", _db(), tmp_path)
|
|
_advance_to_phase_6(tmp_path / m.id)
|
|
_patch_cfg(monkeypatch, tmp_path)
|
|
doc = {
|
|
"question": "domanda riscritta",
|
|
"strategy": "razionale",
|
|
"ctes": [
|
|
{"name": "a", "purpose": "p_a", "rationale": "r_a", "depends_on": []},
|
|
{"name": "b", "purpose": "p_b", "rationale": "r_b", "depends_on": ["a"]},
|
|
],
|
|
}
|
|
res = CliRunner().invoke(
|
|
cte_app,
|
|
["plan", "--session", m.id, "--name", "a", "--name", "b", "--doc", "-"],
|
|
input=json.dumps(doc),
|
|
)
|
|
assert res.exit_code == 0, res.output
|
|
|
|
plan_path = tmp_path / m.id / "cte_plan.json"
|
|
assert json.loads(plan_path.read_text()) == ["a", "b"]
|
|
|
|
doc_path = tmp_path / m.id / "cte_plan_doc.json"
|
|
assert json.loads(doc_path.read_text()) == doc
|
|
|
|
|
|
def test_plan_without_doc_unchanged(tmp_path, monkeypatch):
|
|
m = create_session("q", _db(), tmp_path)
|
|
_advance_to_phase_6(tmp_path / m.id)
|
|
_patch_cfg(monkeypatch, tmp_path)
|
|
res = CliRunner().invoke(cte_app, ["plan", "--session", m.id, "--name", "a"])
|
|
assert res.exit_code == 0, res.output
|
|
assert json.loads((tmp_path / m.id / "cte_plan.json").read_text()) == ["a"]
|
|
assert not (tmp_path / m.id / "cte_plan_doc.json").exists()
|
|
|
|
|
|
def test_plan_doc_name_mismatch_rejected(tmp_path, monkeypatch):
|
|
m = create_session("q", _db(), tmp_path)
|
|
_advance_to_phase_6(tmp_path / m.id)
|
|
_patch_cfg(monkeypatch, tmp_path)
|
|
doc = {"ctes": [{"name": "a", "purpose": "p"}, {"name": "zzz", "purpose": "p2"}]}
|
|
res = CliRunner().invoke(
|
|
cte_app,
|
|
["plan", "--session", m.id, "--name", "a", "--name", "b", "--doc", "-"],
|
|
input=json.dumps(doc),
|
|
)
|
|
assert res.exit_code == 1
|
|
assert res.stdout == ""
|
|
# neither file should be written on a rejected doc
|
|
assert not (tmp_path / m.id / "cte_plan.json").exists()
|
|
assert not (tmp_path / m.id / "cte_plan_doc.json").exists()
|
|
|
|
|
|
def test_plan_doc_wrong_order_rejected(tmp_path, monkeypatch):
|
|
m = create_session("q", _db(), tmp_path)
|
|
_advance_to_phase_6(tmp_path / m.id)
|
|
_patch_cfg(monkeypatch, tmp_path)
|
|
doc = {"ctes": [{"name": "b", "purpose": "p"}, {"name": "a", "purpose": "p2"}]}
|
|
res = CliRunner().invoke(
|
|
cte_app,
|
|
["plan", "--session", m.id, "--name", "a", "--name", "b", "--doc", "-"],
|
|
input=json.dumps(doc),
|
|
)
|
|
assert res.exit_code == 1
|
|
assert not (tmp_path / m.id / "cte_plan_doc.json").exists()
|
|
|
|
|
|
def test_plan_doc_invalid_json_rejected(tmp_path, monkeypatch):
|
|
m = create_session("q", _db(), tmp_path)
|
|
_advance_to_phase_6(tmp_path / m.id)
|
|
_patch_cfg(monkeypatch, tmp_path)
|
|
res = CliRunner().invoke(
|
|
cte_app,
|
|
["plan", "--session", m.id, "--name", "a", "--doc", "-"],
|
|
input="{not valid json",
|
|
)
|
|
assert res.exit_code == 1
|
|
assert not (tmp_path / m.id / "cte_plan_doc.json").exists()
|