feat(cte): add 'tht cte next' — first unapproved plan CTE
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,55 @@
|
||||
"""L1: `tht cte next` — the first plan CTE not yet approved (drives the F6 gate)."""
|
||||
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 test_cte_next_first_unapproved(tmp_path, monkeypatch):
|
||||
m = create_session("q", _db(), tmp_path)
|
||||
(tmp_path / m.id / "cte_plan.json").write_text(json.dumps(["a", "b", "c"]))
|
||||
_patch_cfg(monkeypatch, tmp_path)
|
||||
res = CliRunner().invoke(cte_app, ["next", "--session", m.id])
|
||||
assert res.exit_code == 0, res.output
|
||||
assert res.output.strip() == "a"
|
||||
|
||||
|
||||
def test_cte_next_after_one_approval(tmp_path, monkeypatch):
|
||||
m = create_session("q", _db(), tmp_path)
|
||||
(tmp_path / m.id / "cte_plan.json").write_text(json.dumps(["a", "b", "c"]))
|
||||
append_decision(tmp_path / m.id, type="cte_approved", subject="a")
|
||||
_patch_cfg(monkeypatch, tmp_path)
|
||||
res = CliRunner().invoke(cte_app, ["next", "--session", m.id])
|
||||
assert res.exit_code == 0, res.output
|
||||
assert res.output.strip() == "b"
|
||||
|
||||
|
||||
def test_cte_next_empty_when_all_approved(tmp_path, monkeypatch):
|
||||
m = create_session("q", _db(), tmp_path)
|
||||
(tmp_path / m.id / "cte_plan.json").write_text(json.dumps(["a"]))
|
||||
append_decision(tmp_path / m.id, type="cte_approved", subject="a")
|
||||
_patch_cfg(monkeypatch, tmp_path)
|
||||
res = CliRunner().invoke(cte_app, ["next", "--session", m.id])
|
||||
assert res.exit_code == 0, res.output
|
||||
assert res.output.strip() == ""
|
||||
@@ -150,6 +150,24 @@ def plan_cmd(
|
||||
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(session_dir(cfg, session))
|
||||
if nxt:
|
||||
typer.echo(nxt)
|
||||
|
||||
|
||||
@cte_app.command("list")
|
||||
def list_cmd(
|
||||
session: str = typer.Option(..., "--session"),
|
||||
|
||||
Reference in New Issue
Block a user