94 lines
3.0 KiB
Python
94 lines
3.0 KiB
Python
import json
|
|
|
|
import pytest
|
|
from typer.testing import CliRunner
|
|
|
|
from tht.cli.decision_cmd import decision_app
|
|
from tht.decisions import append_decision, append_decisions, list_decisions
|
|
from tht.phase import current_phase
|
|
|
|
|
|
def _walk_to_phase(session, target):
|
|
while current_phase(session) < target:
|
|
append_decision(session, type="phase_approved", subject=f"phase:{current_phase(session)}")
|
|
|
|
|
|
def _configure_command(monkeypatch, sessions):
|
|
import tht.cli.decision_cmd as mod
|
|
|
|
class _Cfg:
|
|
class paths:
|
|
pass
|
|
|
|
_Cfg.paths.sessions = sessions
|
|
monkeypatch.setattr(mod, "_load_config_or_exit", lambda _c: _Cfg())
|
|
monkeypatch.setattr(mod, "load_session_or_exit", lambda _cfg, _s: None)
|
|
monkeypatch.setattr(mod, "session_dir", lambda _cfg, sid: sessions / sid)
|
|
|
|
|
|
def test_add_join_set_rejects_the_whole_batch_when_one_item_is_invalid(tmp_path, monkeypatch):
|
|
session_id = "s1"
|
|
session = tmp_path / session_id
|
|
session.mkdir()
|
|
_walk_to_phase(session, 4)
|
|
_configure_command(monkeypatch, tmp_path)
|
|
|
|
payload = [
|
|
{"type": "join_modified", "subject": "a-b", "detail": "a.id = b.a_id"},
|
|
{"type": "not_a_decision", "subject": "b-c", "detail": "b.id = c.b_id"},
|
|
]
|
|
result = CliRunner().invoke(
|
|
decision_app,
|
|
["add-join-set", "--session", session_id, "--doc", "-"],
|
|
input=json.dumps(payload),
|
|
)
|
|
|
|
assert result.exit_code != 0
|
|
assert not [decision for decision in list_decisions(session) if decision.type == "join_modified"]
|
|
|
|
|
|
def test_add_join_set_appends_the_complete_valid_batch(tmp_path, monkeypatch):
|
|
session_id = "s1"
|
|
session = tmp_path / session_id
|
|
session.mkdir()
|
|
_walk_to_phase(session, 4)
|
|
_configure_command(monkeypatch, tmp_path)
|
|
|
|
payload = [
|
|
{"type": "join_modified", "subject": "a-b", "detail": "a.id = b.a_id"},
|
|
{"type": "join_modified", "subject": "b-c", "detail": "b.id = c.b_id"},
|
|
]
|
|
result = CliRunner().invoke(
|
|
decision_app,
|
|
["add-join-set", "--session", session_id, "--doc", "-"],
|
|
input=json.dumps(payload),
|
|
)
|
|
|
|
assert result.exit_code == 0, result.output
|
|
joins = [decision for decision in list_decisions(session) if decision.type == "join_modified"]
|
|
assert [decision.subject for decision in joins] == ["a-b", "b-c"]
|
|
|
|
|
|
def test_atomic_replace_failure_keeps_the_original_ledger(tmp_path, monkeypatch):
|
|
session = tmp_path / "s1"
|
|
session.mkdir()
|
|
append_decision(session, type="phase_approved", subject="phase:1")
|
|
original = (session / "review_decisions.jsonl").read_text()
|
|
|
|
import tht.decisions as mod
|
|
|
|
def fail_replace(_source, _target):
|
|
raise OSError("disk failure")
|
|
|
|
monkeypatch.setattr(mod.os, "replace", fail_replace)
|
|
with pytest.raises(OSError, match="disk failure"):
|
|
append_decisions(
|
|
session,
|
|
[
|
|
{"type": "join_modified", "subject": "a-b"},
|
|
{"type": "join_modified", "subject": "b-c"},
|
|
],
|
|
)
|
|
|
|
assert (session / "review_decisions.jsonl").read_text() == original
|