approval-engine/tests/test_api.py

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(app):
    for path in (
        "/v1/check",
        "/authorize",
    ):
        status, body = call(app, "POST", path, {})
        assert status == 404
        assert body["error"] == "not_found"


def test_consume_endpoint_is_a_mutation_not_a_decision(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"})
    digest = "sha256:" + "12" * 32
    status, result = call(
        app,
        "POST",
        f"/v1/approvals/{aid}/consume",
        {"request_digest": digest, "decision_id": "decision:123"},
    )
    assert status == 200
    assert result == {
        "approval_id": aid,
        "status": "consumed",
        "request_digest": digest,
        "decision_id": "decision:123",
        "consumed_at": "2026-08-29T12:00:00+00:00",
        "idempotent": False,
    }
    assert not ({"effect", "decision", "allow", "deny"} & set(result))


def test_consume_requires_request_digest(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"})
    status, body = call(app, "POST", f"/v1/approvals/{aid}/consume", {})
    assert status == 422
    assert body["error"] == "unprocessable"

    status, body = call(
        app,
        "POST",
        f"/v1/approvals/{aid}/consume",
        {"request_digest": ["not", "a", "digest"]},
    )
    assert status == 422
    assert body["error"] == "unprocessable"


def test_consume_endpoint_rejects_different_digest_replay(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"})
    first = "sha256:" + "45" * 32
    second = "sha256:" + "67" * 32
    assert call(
        app, "POST", f"/v1/approvals/{aid}/consume", {"request_digest": first}
    )[0] == 200
    status, body = call(
        app, "POST", f"/v1/approvals/{aid}/consume", {"request_digest": second}
    )
    assert status == 409
    assert body["error"] == "conflict"


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"