DataOps-platform/tests/test_trusted_delivery_api.py

from __future__ import annotations

USER_UID = "01900000-0000-7000-8000-000000068801"


class FakeTrustedService:
    preview_calls = 0
    mutation_calls = 0
    def create_policy_version(self, payload, *, actor_uid):
        return {"uid": "policy-1", "status": "active"}

    def evaluate_gateway(self, payload):
        return {"decision": "deny", "reason_code": "default_deny"}

    def create_legal_hold(self, payload, *, actor_uid):
        return {"uid": "hold-1", "status": "active"}

    def activate_policy_version(self, payload, *, actor_uid):
        return {"uid": "policy-1", "status": "active"}

    def rollback_policy_version(self, payload, *, actor_uid):
        return {"uid": "policy-1", "status": "active"}

    def provision(self, payload, *, actor_uid):
        return {"grant_uid": "grant-1", "status": "applied"}

    def release_legal_hold(self, payload, *, actor_uid):
        return {"uid": "hold-1", "status": "released"}

    def execute_reclaim(self, payload, *, actor_uid):
        return {"grant_uid": "grant-1", "status": "reclaimed"}

    def revoke_expired(self, *, as_of, actor_uid):
        type(self).mutation_calls += 1
        return []

    def preview_reclaim(self, *, as_of, actor_uid):
        type(self).preview_calls += 1
        return [{"grant_uid": "grant-1", "status": "eligible"}]


def _client(monkeypatch):
    from app import create_app
    from app.api.system import trusted_delivery

    monkeypatch.setattr(trusted_delivery, "_service", lambda: FakeTrustedService())
    monkeypatch.setattr("app.core.system.auth.load_identity_from_token", lambda token, secret: {"id": USER_UID, "roles": [token]} if token in {"viewer", "editor", "admin"} else None)
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client()


def test_trusted_delivery_api_permissions_closed_schema_and_no_store(monkeypatch):
    client = _client(monkeypatch)
    headers = {"Authorization": "Bearer editor"}
    assert client.post("/api/system/trusted-delivery/policy-versions", json={}, headers={"Authorization": "Bearer viewer"}).status_code == 403
    response = client.post("/api/system/trusted-delivery/policy-versions", json={"unexpected": True}, headers=headers)
    assert response.status_code == 400
    assert response.headers["Cache-Control"] == "no-store"
    assert "unexpected" not in response.get_data(as_text=True)
    decision = client.post("/api/system/trusted-delivery/decisions", json={}, headers=headers)
    assert decision.status_code == 200
    assert decision.headers["Cache-Control"] == "no-store"
    assert client.post("/api/system/trusted-delivery/legal-holds", json={}, headers=headers).status_code == 403
    assert client.post("/api/system/trusted-delivery/legal-holds", json={}, headers={"Authorization": "Bearer admin"}).status_code == 201


def test_trusted_delivery_lifecycle_api_has_separate_operate_manage_boundaries(monkeypatch):
    client = _client(monkeypatch)
    editor = {"Authorization": "Bearer editor"}
    admin = {"Authorization": "Bearer admin"}
    for path in ("/api/system/trusted-delivery/policy-versions/activate", "/api/system/trusted-delivery/policy-versions/rollback", "/api/system/trusted-delivery/provisions"):
        response = client.post(path, json={"unexpected": True}, headers=editor)
        assert response.status_code == 400
        assert response.headers["Cache-Control"] == "no-store"
    for path in ("/api/system/trusted-delivery/legal-holds/release", "/api/system/trusted-delivery/reclaim/execute"):
        assert client.post(path, json={}, headers=editor).status_code == 403
        response = client.post(path, json={"unexpected": True}, headers=admin)
        assert response.status_code == 400
        assert response.headers["Cache-Control"] == "no-store"


def test_reclaim_preview_is_read_only(monkeypatch):
    client = _client(monkeypatch)
    response = client.post("/api/system/trusted-delivery/reclaim/preview", json={}, headers={"Authorization": "Bearer admin"})
    assert response.status_code == 200
    assert FakeTrustedService.preview_calls == 1
    assert FakeTrustedService.mutation_calls == 0
    assert response.headers["Cache-Control"] == "no-store"