Wait within decision validity bounds and preserve failed preflight evidence
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a09cbb-87c6-7900-a145-4ce53ba9f1a6
This commit is contained in:
parent
1c9fd5279c
commit
5aaa49a2ec
7 changed files with 647 additions and 1 deletions
33
tests/test_decision_start_wait.py
Normal file
33
tests/test_decision_start_wait.py
Normal file
|
|
@ -0,0 +1,33 @@
|
|||
from datetime import datetime, timedelta, timezone
|
||||
import pytest
|
||||
from secrets_engine.authorization import wait_for_decision_start
|
||||
from secrets_engine.errors import DecisionError
|
||||
|
||||
@pytest.mark.parametrize("offset,expected", [(-1, []), (0, []), (0.25, [0.25]), (2, [2])])
|
||||
def test_waits_until_start_without_changing_the_envelope(offset, expected):
|
||||
now = datetime(2026, 9, 14, tzinfo=timezone.utc)
|
||||
envelope = {"lifetime": {"not_before": (now + timedelta(seconds=offset)).isoformat()}}
|
||||
calls = []
|
||||
wait_for_decision_start(envelope, clock=lambda: now, sleeper=calls.append)
|
||||
assert calls == expected
|
||||
assert envelope["lifetime"]["not_before"] == (now + timedelta(seconds=offset)).isoformat()
|
||||
|
||||
def test_large_skew_refuses_without_sleep():
|
||||
now = datetime(2026, 9, 14, tzinfo=timezone.utc)
|
||||
with pytest.raises(DecisionError, match="bounded wait"):
|
||||
wait_for_decision_start({"lifetime": {"not_before": (now + timedelta(seconds=3)).isoformat()}},
|
||||
clock=lambda: now, sleeper=lambda _: pytest.fail("must not sleep"))
|
||||
|
||||
def test_wait_does_not_waive_expiration_or_start_validation():
|
||||
from secrets_engine.authorization import validate_decision_envelope
|
||||
from tests.test_action_authorization import _envelope, _request
|
||||
request = _request()
|
||||
envelope = _envelope(request)
|
||||
now = datetime(2026, 9, 14, tzinfo=timezone.utc)
|
||||
envelope["lifetime"] = {"not_before": (now + timedelta(seconds=1)).isoformat(),
|
||||
"expires_at": (now + timedelta(seconds=2)).isoformat()}
|
||||
wait_for_decision_start(envelope, clock=lambda: now, sleeper=lambda _: None)
|
||||
with pytest.raises(DecisionError, match="has not started"):
|
||||
validate_decision_envelope(envelope, request, accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"}, accepted_policy_versions={"v2"}, now=now)
|
||||
with pytest.raises(DecisionError, match="expired"):
|
||||
validate_decision_envelope(envelope, request, accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"}, accepted_policy_versions={"v2"}, now=now + timedelta(seconds=3))
|
||||
Loading…
Add table
Add a link
Reference in a new issue