81 lines
2.6 KiB
Python
81 lines
2.6 KiB
Python
|
from datetime import datetime, timezone
|
||
|
|
|
||
|
|
from approval_engine.errors import Conflict, DuplicateApprover
|
||
|
|
from approval_engine.store import Engine
|
||
|
|
from tests.conftest import approve, binding, validity
|
||
|
|
|
||
|
|
|
||
|
|
def test_requested_until_threshold(engine):
|
||
|
|
obj = engine.create(binding(), validity(), required_count=2)
|
||
|
|
assert obj.status == "requested"
|
||
|
|
obj = engine.add_entry(obj.id, "user:alice")
|
||
|
|
assert obj.status == "requested"
|
||
|
|
claim = engine.claim(obj.id)
|
||
|
|
assert claim["valid_now"] is False
|
||
|
|
assert claim["reason_code"] == "requested"
|
||
|
|
|
||
|
|
|
||
|
|
def test_threshold_approves_and_valid_now(engine):
|
||
|
|
obj = approve(engine, required=2)
|
||
|
|
assert obj.status == "approved"
|
||
|
|
assert len(obj.entries) == 2
|
||
|
|
claim = engine.claim(obj.id)
|
||
|
|
assert claim["state"] == "valid"
|
||
|
|
assert claim["valid_now"] is True
|
||
|
|
|
||
|
|
|
||
|
|
def test_duplicate_approver_fails_closed(engine):
|
||
|
|
obj = engine.create(binding(), validity(), required_count=2)
|
||
|
|
engine.add_entry(obj.id, "user:alice")
|
||
|
|
try:
|
||
|
|
engine.add_entry(obj.id, "user:alice")
|
||
|
|
raise AssertionError("duplicate must fail")
|
||
|
|
except DuplicateApprover:
|
||
|
|
pass
|
||
|
|
obj = engine.get(obj.id)
|
||
|
|
assert len(obj.entries) == 1
|
||
|
|
assert obj.status == "requested"
|
||
|
|
|
||
|
|
|
||
|
|
def test_revoke_without_holder_and_next_claim(engine):
|
||
|
|
obj = approve(engine)
|
||
|
|
engine.revoke(obj.id)
|
||
|
|
claim = engine.claim(obj.id)
|
||
|
|
assert claim["valid_now"] is False
|
||
|
|
assert claim["state"] == "revoked"
|
||
|
|
assert claim["reason_code"] == "revoked"
|
||
|
|
|
||
|
|
|
||
|
|
def test_expiry_on_observation():
|
||
|
|
jumping = {"t": datetime(2026, 8, 29, 12, 0, tzinfo=timezone.utc)}
|
||
|
|
eng = Engine(":memory:", clock=lambda: jumping["t"])
|
||
|
|
obj = eng.create(binding(), validity(), required_count=1)
|
||
|
|
obj = eng.add_entry(obj.id, "user:alice")
|
||
|
|
assert eng.claim(obj.id)["valid_now"] is True
|
||
|
|
jumping["t"] = datetime(2026, 8, 29, 16, 0, tzinfo=timezone.utc)
|
||
|
|
claim = eng.claim(obj.id)
|
||
|
|
assert claim["state"] == "expired"
|
||
|
|
assert claim["valid_now"] is False
|
||
|
|
eng.close()
|
||
|
|
|
||
|
|
|
||
|
|
def test_not_yet_valid():
|
||
|
|
future = datetime(2026, 8, 29, 10, 0, tzinfo=timezone.utc)
|
||
|
|
eng = Engine(":memory:", clock=lambda: future)
|
||
|
|
obj = eng.create(binding(), validity(), required_count=1)
|
||
|
|
obj = eng.add_entry(obj.id, "user:alice")
|
||
|
|
claim = eng.claim(obj.id)
|
||
|
|
assert claim["valid_now"] is False
|
||
|
|
assert claim["reason_code"] == "not_yet_valid"
|
||
|
|
eng.close()
|
||
|
|
|
||
|
|
|
||
|
|
def test_cannot_revoke_twice(engine):
|
||
|
|
obj = approve(engine)
|
||
|
|
engine.revoke(obj.id)
|
||
|
|
try:
|
||
|
|
engine.revoke(obj.id)
|
||
|
|
raise AssertionError("second revoke must conflict")
|
||
|
|
except Conflict:
|
||
|
|
pass
|