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_supersede_endpoint_reissues_an_expired_approval(app):
    expired = {
        "not_before": "2026-08-29T09:00:00+00:00",
        "expires_at": "2026-08-29T10:00:00+00:00",  # before the frozen test clock
    }
    _, created = call(app, "POST", "/v1/approvals", {"binding": binding(), "validity": expired})
    aid = created["id"]
    status, fetched = call(app, "GET", f"/v1/approvals/{aid}")
    assert status == 200
    assert fetched["status"] == "expired"

    fresh = {"not_before": "2026-08-29T13:00:00+00:00", "expires_at": "2026-08-29T18:00:00+00:00"}
    status, result = call(app, "POST", f"/v1/approvals/{aid}/supersede", {"validity": fresh})
    assert status == 200
    successor_id = result["successor_id"]

    status, original = call(app, "GET", f"/v1/approvals/{aid}")
    assert original["status"] == "superseded"
    assert original["superseded_by"] == successor_id

    status, successor = call(app, "GET", f"/v1/approvals/{successor_id}")
    assert successor["status"] == "requested"
    assert successor["validity"] == fresh


def test_supersede_endpoint_refuses_expired_without_validity(app):
    expired = {
        "not_before": "2026-08-29T09:00:00+00:00",
        "expires_at": "2026-08-29T10:00:00+00:00",
    }
    _, created = call(app, "POST", "/v1/approvals", {"binding": binding(), "validity": expired})
    aid = created["id"]
    call(app, "GET", f"/v1/approvals/{aid}")  # persist the expiry
    status, body = call(app, "POST", f"/v1/approvals/{aid}/supersede", {})
    assert status == 409
    assert body["error"] == "conflict"


def test_store_unavailable_is_503():
    from approval_engine.api import App
    from approval_engine.auth import Identity, StaticTokenAuthenticator
    from approval_engine.errors import StoreUnavailable
    from approval_engine.store import Engine

    class Dead(Engine):
        def storage_status(self):
            return {"schema_current": True, "persistent": True}

        def outbox_stats(self):
            raise StoreUnavailable("down")

    identity = Identity(
        subject="test",
        issuer="test",
        audiences=("approval-engine",),
        principal_type="service",
        tenant="test",
        roles=frozenset(),
        scopes=frozenset(),
        assurance={},
        evidence_ref="test",
    )
    app = App(Dead.__new__(Dead), StaticTokenAuthenticator({"test-token": identity}))
    status, body = call(app, "GET", "/v1/readyz")
    assert status == 503
    assert body["error"] == "store_unavailable"