DataOps-platform/tests/test_trusted_delivery_subscription_api.py

from __future__ import annotations

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


class FakeSubscriptionService:
    def create_subscription(self, payload, *, actor_uid):
        return {"uid": "01900000-0000-7000-8000-000000069901", "status": "draft"}

    def pause_subscription(self, uid, *, actor_uid):
        return {"uid": uid, "status": "paused"}

    def activate_subscription(self, uid, *, actor_uid):
        return {"uid": uid, "status": "active"}

    def resume_subscription(self, uid, *, actor_uid):
        return {"uid": uid, "status": "active"}

    def terminate_subscription(self, uid, *, actor_uid):
        return {"uid": uid, "status": "terminated"}

    def compensate_dead_letter(self, uid, *, reason_code, receipt_code, actor_uid):
        return {"uid": uid, "status": "compensated"}

    def report_anomaly(self, payload, *, actor_uid):
        return {"kind": "purpose_breach"}


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

    monkeypatch.setattr(trusted_delivery, "_subscription_service", lambda: FakeSubscriptionService())
    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_subscription_api_enforces_rbac_closed_bodies_and_no_store(monkeypatch):
    client = _client(monkeypatch)
    editor = {"Authorization": "Bearer editor"}
    admin = {"Authorization": "Bearer admin"}
    assert client.post("/api/system/trusted-delivery/subscriptions", json={}, headers={"Authorization": "Bearer viewer"}).status_code == 403
    response = client.post("/api/system/trusted-delivery/subscriptions", json={"unexpected": True}, headers=editor)
    assert response.status_code == 400
    assert response.headers["Cache-Control"] == "no-store"
    assert client.post("/api/system/trusted-delivery/subscriptions/01900000-0000-7000-8000-000000069901/pause", json={}, headers=editor).status_code == 200
    assert client.post("/api/system/trusted-delivery/subscriptions/deliveries/01900000-0000-7000-8000-000000069902/compensate", json={}, headers=editor).status_code == 403
    response = client.post("/api/system/trusted-delivery/subscriptions/deliveries/01900000-0000-7000-8000-000000069902/compensate", json={"unexpected": True}, headers=admin)
    assert response.status_code == 400
    assert response.headers["Cache-Control"] == "no-store"


def test_subscription_activate_is_operate_protected_and_no_store(monkeypatch):
    client = _client(monkeypatch)
    uid = "01900000-0000-7000-8000-000000069901"
    assert client.post(f"/api/system/trusted-delivery/subscriptions/{uid}/activate", json={}, headers={"Authorization": "Bearer viewer"}).status_code == 403
    response = client.post(f"/api/system/trusted-delivery/subscriptions/{uid}/activate", json={}, headers={"Authorization": "Bearer editor"})
    assert response.status_code == 200
    assert response.headers["Cache-Control"] == "no-store"