marcopan
a312746fc8
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.
70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
"""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
|