feat(harness): require_phase_or_exit + phase advance/reopen/show (Onda 1.2)
require_phase_or_exit: guard riscritto vs Workflow (load_workflow().phase_name invece della costante PHASE_NAMES drift). Exit 1 se la sessione e' sotto soglia. Usato da cte/decision/datamart cmd. Comandi phase (portati + adattati al modello ThothII, non copia cieca): - advance: persiste phase_approved; --auto exit 6 se la fase non e' completa (contratto col gate) - reopen: persiste phase_reopened + teardown_to_phase degli artefatti oltre il target - show: stato sessione (fase corrente, ultime decisioni) session_dir helper tenuto qui (mirror di session_cmd) per evitare circular import. _cfg() fa fallback a THT_WORKSPACE env finche' _load_config_or_exit (Onda 1.4) non sara' portato. L1: 4 test require_phase_or_exit (allow at/above, exit below, message con nome fase dal workflow). Suite: 161 passed.
This commit is contained in:
@@ -0,0 +1,69 @@
|
||||
"""L1: require_phase_or_exit — the phase-gate guard rewritten vs Workflow.
|
||||
|
||||
Used by cte_cmd/decision_cmd/datamart_cmd to refuse an action when the session
|
||||
hasn't reached a phase yet. The rewrite replaces the old PHASE_NAMES constant
|
||||
with load_workflow().phase_name(). Exit code 1 (typer.Exit) when below threshold.
|
||||
"""
|
||||
import pytest
|
||||
import typer
|
||||
|
||||
from tht.cli.phase_cmd import require_phase_or_exit
|
||||
|
||||
|
||||
def _make_session(tmp_path, current_phase_num: int) -> str:
|
||||
"""Build a session dir whose current_phase reads as current_phase_num.
|
||||
current_phase is derived from the ledger fold (cur = last approved + 1), so
|
||||
approving phases 1..N-1 puts the session at phase N."""
|
||||
import json
|
||||
|
||||
s = tmp_path / "sess"
|
||||
s.mkdir()
|
||||
decisions = [
|
||||
{"seq": n, "type": "phase_approved", "subject": f"phase:{n}", "ts": "2025-01-01T00:00:00"}
|
||||
for n in range(1, current_phase_num)
|
||||
]
|
||||
(s / "review_decisions.jsonl").write_text(
|
||||
"\n".join(json.dumps(d) for d in decisions) + ("\n" if decisions else "")
|
||||
)
|
||||
return "sess"
|
||||
|
||||
|
||||
class _StubConfig:
|
||||
"""Minimal config: paths.sessions points at tmp_path so session_dir resolves there."""
|
||||
|
||||
def __init__(self, tmp_path):
|
||||
from tht.config import PathsConfig
|
||||
|
||||
self.paths = PathsConfig(sessions=tmp_path)
|
||||
|
||||
|
||||
def test_require_phase_or_exit_allows_when_at_or_above(tmp_path):
|
||||
cfg = _StubConfig(tmp_path)
|
||||
session = _make_session(tmp_path, current_phase_num=5) # at phase 5
|
||||
# require phase 4 -> allowed (cur 5 >= 4), no exit
|
||||
require_phase_or_exit(cfg, session, min_phase=4)
|
||||
|
||||
|
||||
def test_require_phase_or_exit_allows_when_exactly_at(tmp_path):
|
||||
cfg = _StubConfig(tmp_path)
|
||||
session = _make_session(tmp_path, current_phase_num=4) # at phase 4
|
||||
require_phase_or_exit(cfg, session, min_phase=4) # boundary: allowed
|
||||
|
||||
|
||||
def test_require_phase_or_exit_exits_when_below(tmp_path):
|
||||
cfg = _StubConfig(tmp_path)
|
||||
session = _make_session(tmp_path, current_phase_num=3) # at phase 3
|
||||
with pytest.raises(typer.Exit) as exc_info:
|
||||
require_phase_or_exit(cfg, session, min_phase=5) # require 5 -> exit 1
|
||||
assert exc_info.value.exit_code == 1
|
||||
|
||||
|
||||
def test_require_phase_or_exit_message_uses_workflow_phase_name(tmp_path, capsys):
|
||||
cfg = _StubConfig(tmp_path)
|
||||
session = _make_session(tmp_path, current_phase_num=3)
|
||||
with pytest.raises(typer.Exit):
|
||||
require_phase_or_exit(cfg, session, min_phase=6) # phase 6 = cte
|
||||
captured = capsys.readouterr()
|
||||
# the message must use the workflow-derived name (not a hardcoded constant)
|
||||
assert "cte" in captured.err.lower()
|
||||
assert "Fase 6" in captured.err
|
||||
Reference in New Issue
Block a user