75 lines
2.8 KiB
Python
75 lines
2.8 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
|
|
|
|
session_id = "2026-07-16-120000"
|
|
s = tmp_path / session_id
|
|
s.mkdir()
|
|
(s / "session_manifest.yaml").write_text(
|
|
f"id: {session_id}\nquestion: q\ndatabase: d\nschema: s\n"
|
|
"created_at: 2026-01-01T00:00:00+00:00\nstatus: open\n"
|
|
)
|
|
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 session_id
|
|
|
|
|
|
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
|