DataOps-platform/tests/test_trusted_delivery_controls_api.py

from __future__ import annotations

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


class FakeControlsService:
    def active_profile_summary(self, profile_id):
        return {"profile_id": profile_id, "status": "active"}

    def evaluate_release_gate(self, payload, *, actor_uid):
        return {"decision_code": "RELEASE_GATE_APPROVED"}

    def activate_profile(self, payload, *, actor_uid, operation="activate"):
        return {"profile_id": payload.get("profile_id"), "status": "active", "operation": operation}

    def approve_capability(self, payload, *, actor_uid):
        return {"status": "approved"}

    def evidence_summary(self):
        return [{"evidence_ref": "trusted-evidence:api", "decision_code": "SAFE"}]

    def approve_destruction(self, payload, *, actor_uid):
        return {"status": "approved"}


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

    monkeypatch.setattr(trusted_delivery_controls, "_service", lambda: FakeControlsService())
    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_controls_api_rbac_closed_schema_no_store_and_safe_viewer_summary(monkeypatch):
    client = _client(monkeypatch)
    viewer = {"Authorization": "Bearer viewer"}
    editor = {"Authorization": "Bearer editor"}
    admin = {"Authorization": "Bearer admin"}
    summary = client.get("/api/system/trusted-delivery/controls/profiles/finance/active", headers=viewer)
    assert summary.status_code == 200
    assert summary.headers["Cache-Control"] == "no-store"
    assert client.post("/api/system/trusted-delivery/controls/release-gates/evaluate", json={}, headers=viewer).status_code == 403
    response = client.post("/api/system/trusted-delivery/controls/release-gates/evaluate", json={"raw_scan": "secret"}, headers=editor)
    assert response.status_code == 400
    assert response.headers["Cache-Control"] == "no-store"
    assert "raw_scan" not in response.get_data(as_text=True)
    assert client.post("/api/system/trusted-delivery/controls/profiles/activate", json={}, headers=editor).status_code == 403
    for path in ("/api/system/trusted-delivery/controls/profiles/activate", "/api/system/trusted-delivery/controls/profiles/rollback", "/api/system/trusted-delivery/controls/destruction-approvals"):
        response = client.post(path, json={"unexpected": True}, headers=admin)
        assert response.status_code == 400
        assert response.headers["Cache-Control"] == "no-store"