tegwick
1b8261917e
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
124 lines
4.9 KiB
Python
124 lines
4.9 KiB
Python
"""Offline prototype regression cases; no owner adapter, credential or live use."""
|
|
from dataclasses import FrozenInstanceError, replace
|
|
import pytest
|
|
from warden.case_receipt_state import (
|
|
Outcome, Pins, Refused, StepState, begin, begin_rollback, finish,
|
|
finish_rollback, recover, rollback_objects,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def pins():
|
|
return Pins(*["sha256:" + character * 64 for character in "abcdef"])
|
|
|
|
|
|
@pytest.fixture
|
|
def state(pins):
|
|
return StepState("mint-secret", "railiance-platform", pins, ("case-policy", "case-role"))
|
|
|
|
|
|
def test_records_running_before_dispatch_without_mutating_history(state, pins):
|
|
running = begin(state, pins)
|
|
assert state.receipts == ()
|
|
assert running.last.outcome is Outcome.RUNNING
|
|
with pytest.raises(FrozenInstanceError):
|
|
running.last.attempt = 8
|
|
|
|
|
|
def test_changed_exact_revision_refuses_dispatch(state, pins):
|
|
with pytest.raises(Refused):
|
|
begin(state, replace(pins, adapter="sha256:" + "0" * 64))
|
|
|
|
|
|
def test_lost_receipt_becomes_uncertain_and_cannot_retry(state, pins):
|
|
resumed = recover(begin(state, pins))
|
|
assert resumed.last.outcome is Outcome.UNCERTAIN
|
|
assert len(resumed.receipts) == 2
|
|
with pytest.raises(Refused):
|
|
begin(replace(resumed, retry_safe=True), pins, retry=True)
|
|
|
|
|
|
@pytest.mark.parametrize("outcome", [Outcome.SUCCESS, Outcome.NO_CHANGE])
|
|
def test_verified_success_or_no_change_never_replayed(state, pins, outcome):
|
|
done = finish(begin(state, pins), pins, outcome)
|
|
with pytest.raises(Refused):
|
|
begin(done, pins, retry=True)
|
|
|
|
|
|
def test_retry_requires_verified_no_change_failure_and_safe_adapter(state, pins):
|
|
failed = finish(begin(state, pins), pins, Outcome.FAILED_BEFORE)
|
|
with pytest.raises(Refused):
|
|
begin(failed, pins, retry=True)
|
|
safe = replace(failed, retry_safe=True)
|
|
with pytest.raises(Refused):
|
|
begin(safe, pins)
|
|
retry = begin(safe, pins, retry=True)
|
|
assert retry.last.outcome is Outcome.RUNNING
|
|
assert len(retry.receipts) == 3
|
|
|
|
|
|
def test_failure_after_change_does_not_become_safe_retry_or_rollback(state, pins):
|
|
failed = finish(begin(state, pins), pins, Outcome.FAILED_AFTER, created=("case-policy",))
|
|
with pytest.raises(Refused):
|
|
begin(replace(failed, retry_safe=True), pins, retry=True)
|
|
with pytest.raises(Refused):
|
|
rollback_objects(failed, pins)
|
|
|
|
|
|
def test_rollback_only_created_not_all_reviewed_objects(state, pins):
|
|
done = finish(begin(state, pins), pins, Outcome.SUCCESS, created=("case-role",))
|
|
assert rollback_objects(done, pins) == ("case-role",)
|
|
assert "case-policy" not in rollback_objects(done, pins)
|
|
undo = begin_rollback(done, pins)
|
|
verified = finish_rollback(undo, pins, verified=True)
|
|
assert verified.last.outcome is Outcome.ROLLED_BACK
|
|
with pytest.raises(Refused):
|
|
begin_rollback(verified, pins)
|
|
|
|
|
|
def test_rollback_reverse_order_and_uncertain_undo_blocks_replay(state, pins):
|
|
done = finish(begin(state, pins), pins, Outcome.SUCCESS,
|
|
created=("case-policy", "case-role"))
|
|
assert rollback_objects(done, pins) == ("case-role", "case-policy")
|
|
uncertain = recover(begin_rollback(done, pins))
|
|
assert uncertain.last.outcome is Outcome.ROLLBACK_UNCERTAIN
|
|
with pytest.raises(Refused):
|
|
begin_rollback(uncertain, pins)
|
|
|
|
|
|
def test_unverified_rollback_cannot_claim_success(state, pins):
|
|
done = finish(begin(state, pins), pins, Outcome.SUCCESS, created=("case-role",))
|
|
failed = finish_rollback(begin_rollback(done, pins), pins, verified=False)
|
|
assert failed.last.outcome is Outcome.ROLLBACK_UNCERTAIN
|
|
|
|
|
|
def test_refuses_shared_objects_or_false_no_change_creation(state, pins):
|
|
running = begin(state, pins)
|
|
for outcome, created in [(Outcome.SUCCESS, ("shared-group",)),
|
|
(Outcome.NO_CHANGE, ("case-policy",)),
|
|
(Outcome.FAILED_BEFORE, ("case-policy",)),
|
|
(Outcome.SUCCESS, ("case-policy", "case-policy"))]:
|
|
with pytest.raises(Refused):
|
|
finish(running, pins, outcome, created=created)
|
|
|
|
|
|
def test_no_raw_payload_or_output_receipt_interface(state, pins):
|
|
with pytest.raises(TypeError):
|
|
finish(begin(state, pins), pins, Outcome.SUCCESS, output="sensitive output")
|
|
with pytest.raises(Refused):
|
|
StepState("step", "owner", pins, ("path with spaces or output",))
|
|
with pytest.raises(Refused):
|
|
Pins("not-a-digest", pins.case_type, pins.adapter, pins.catalog, pins.acceptance, pins.run_sheet)
|
|
|
|
|
|
def test_out_of_order_result_and_noop_recovery(state, pins):
|
|
assert recover(state) is state
|
|
with pytest.raises(Refused):
|
|
finish(state, pins, Outcome.SUCCESS)
|
|
with pytest.raises(Refused):
|
|
finish_rollback(state, pins, verified=True)
|
|
|
|
|
|
def test_result_must_be_closed_enum_not_string(state, pins):
|
|
with pytest.raises(Refused):
|
|
finish(begin(state, pins), pins, "verified-success")
|