70 lines
2.1 KiB
Python
70 lines
2.1 KiB
Python
|
from approval_engine.api import call
|
||
|
|
from tests.conftest import binding, validity
|
||
|
|
|
||
|
|
|
||
|
|
def test_readyz(app):
|
||
|
|
status, body = call(app, "GET", "/v1/readyz")
|
||
|
|
assert status == 200
|
||
|
|
assert body["status"] == "ok"
|
||
|
|
|
||
|
|
|
||
|
|
def test_create_entry_claim_roundtrip(app):
|
||
|
|
status, created = call(
|
||
|
|
app,
|
||
|
|
"POST",
|
||
|
|
"/v1/approvals",
|
||
|
|
{"binding": binding(), "validity": validity(), "required_count": 1},
|
||
|
|
)
|
||
|
|
assert status == 201
|
||
|
|
aid = created["id"]
|
||
|
|
status, _ = call(app, "POST", f"/v1/approvals/{aid}/entries", {"subject_id": "user:alice"})
|
||
|
|
assert status == 200
|
||
|
|
status, claim = call(app, "GET", f"/v1/approvals/{aid}/claim")
|
||
|
|
assert status == 200
|
||
|
|
assert claim["kind"] == "approval-claim"
|
||
|
|
assert claim["valid_now"] is True
|
||
|
|
assert "effect" not in claim
|
||
|
|
assert "decision" not in claim
|
||
|
|
|
||
|
|
|
||
|
|
def test_no_check_or_authorize_or_consume(app):
|
||
|
|
for path in (
|
||
|
|
"/v1/check",
|
||
|
|
"/authorize",
|
||
|
|
"/v1/approvals/00000000-0000-0000-0000-000000000001/consume",
|
||
|
|
"/v1/approvals/abc/consume",
|
||
|
|
):
|
||
|
|
status, body = call(app, "POST", path, {})
|
||
|
|
assert status == 404
|
||
|
|
assert "consume is not implemented" in body.get("message", "") or body["error"] == "not_found"
|
||
|
|
|
||
|
|
|
||
|
|
def test_claim_after_revoke(app):
|
||
|
|
_, created = call(
|
||
|
|
app,
|
||
|
|
"POST",
|
||
|
|
"/v1/approvals",
|
||
|
|
{"binding": binding(), "validity": validity()},
|
||
|
|
)
|
||
|
|
aid = created["id"]
|
||
|
|
call(app, "POST", f"/v1/approvals/{aid}/entries", {"subject_id": "user:alice"})
|
||
|
|
call(app, "POST", f"/v1/approvals/{aid}/revoke", {})
|
||
|
|
status, claim = call(app, "GET", f"/v1/approvals/{aid}/claim")
|
||
|
|
assert status == 200
|
||
|
|
assert claim["valid_now"] is False
|
||
|
|
assert claim["reason_code"] == "revoked"
|
||
|
|
|
||
|
|
|
||
|
|
def test_store_unavailable_is_503():
|
||
|
|
from approval_engine.api import App
|
||
|
|
from approval_engine.errors import StoreUnavailable
|
||
|
|
from approval_engine.store import Engine
|
||
|
|
|
||
|
|
class Dead(Engine):
|
||
|
|
def outbox_stats(self):
|
||
|
|
raise StoreUnavailable("down")
|
||
|
|
|
||
|
|
status, body = call(App(Dead.__new__(Dead)), "GET", "/v1/readyz")
|
||
|
|
assert status == 503
|
||
|
|
assert body["error"] == "store_unavailable"
|