DataOps-platform/tests/test_wp10_tenant_api.py

from __future__ import annotations

from pathlib import Path

import yaml

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


class _ControlService:
    calls: list[dict] = []

    def issue_quota_claim(self, *, principal_id, host, request):
        type(self).calls.append({"principal_id": principal_id, "host": host, "request": request})
        return {"claim_uid": "claim-1", "tenant_id": "tenant-alpha", "status": "issued"}

    def provision_private(self, *, principal_id, host, request):
        type(self).calls.append({"principal_id": principal_id, "host": host, "request": request})
        return {"tenant_id": "tenant-alpha", "state": "active"}

    def issue_lifecycle_approval(self, *, reviewer_id, request):
        type(self).calls.append({"reviewer_id": reviewer_id, "request": request})
        return {"approval_ref": "01900000-0000-7000-8000-000000068999"}

    def lifecycle_transition(self, *, principal_id, host, operation, request):
        type(self).calls.append({"principal_id": principal_id, "host": host, "operation": operation, "request": request})
        return {"tenant_id": "tenant-alpha", "state": "frozen", "lease_fence": 1}

    def read(self, *, principal_id, host, kind):
        type(self).calls.append({"principal_id": principal_id, "host": host, "kind": kind})
        return {"tenant_id": "tenant-alpha", kind: []}


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

    _ControlService.calls = []
    monkeypatch.setenv("TRUSTED_TENANT_ROUTE_HOST", "private.example")
    monkeypatch.setattr(tenant_routes, "_service", lambda: _ControlService())
    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_wp10_quota_claim_route_is_authenticated_server_derived_closed_and_no_store(monkeypatch):
    client = _client(monkeypatch)
    editor = {"Authorization": "Bearer editor", "Host": "private.example"}
    assert client.post("/api/system/tenant/quota-claims", json={}, headers={"Authorization": "Bearer viewer"}).status_code == 403
    rejected = client.post("/api/system/tenant/quota-claims", json={"tenant_id": "tenant-beta", "quota_name": "records", "amount": "1", "idempotency_key": "reserve-1"}, headers=editor)
    assert rejected.status_code == 400
    assert rejected.headers["Cache-Control"] == "no-store"
    created = client.post("/api/system/tenant/quota-claims", json={"quota_name": "records", "amount": "1", "idempotency_key": "reserve-1"}, headers=editor)
    assert created.status_code == 201
    assert created.headers["Cache-Control"] == "no-store"
    assert _ControlService.calls == [{
        "principal_id": USER_UID,
        "host": "private.example",
        "request": {"quota_name": "records", "amount": "1", "idempotency_key": "reserve-1"},
    }]


def test_wp10_private_provision_route_is_admin_only_closed_and_no_store(monkeypatch):
    client = _client(monkeypatch)
    headers = {"Authorization": "Bearer admin", "Host": "private.example"}
    assert client.post("/api/system/tenant/provisions", json={"idempotency_key": "provision-1"}, headers={"Authorization": "Bearer editor"}).status_code == 403
    rejected = client.post("/api/system/tenant/provisions", json={"tenant_id": "tenant-beta", "idempotency_key": "provision-1"}, headers=headers)
    assert rejected.status_code == 400
    created = client.post("/api/system/tenant/provisions", json={"idempotency_key": "provision-1"}, headers=headers)
    assert created.status_code == 201
    assert created.headers["Cache-Control"] == "no-store"


def test_wp10_approval_route_is_admin_only_closed_and_no_store(monkeypatch):
    client = _client(monkeypatch)
    admin = {"Authorization": "Bearer admin", "Host": "private.example"}
    assert client.post("/api/system/tenant/approvals", json={"operation": "begin_recovery", "expected_fence": "1", "idempotency_key": "recover-1"}, headers={"Authorization": "Bearer editor"}).status_code == 403
    rejected = client.post("/api/system/tenant/approvals", json={"tenant_id": "tenant-beta", "operation": "begin_recovery", "expected_fence": "1", "idempotency_key": "recover-1"}, headers=admin)
    assert rejected.status_code == 400
    created = client.post("/api/system/tenant/approvals", json={"operation": "begin_recovery", "expected_fence": "1", "idempotency_key": "recover-1"}, headers=admin)
    assert created.status_code == 201
    assert created.headers["Cache-Control"] == "no-store"


def test_wp10_approval_openapi_contract_is_permissioned_required_and_closed():
    operation = yaml.safe_load(Path("docs/architecture/OPENAPI.yaml").read_text())["paths"]["/api/system/tenant/approvals"]["post"]
    assert operation["x-required-permission"] == "identity:manage"
    assert operation["requestBody"]["required"] is True
    schema = operation["requestBody"]["content"]["application/json"]["schema"]
    assert schema["additionalProperties"] is False
    assert set(schema["required"]) == {"operation", "expected_fence", "idempotency_key"}
    assert "201" in operation["responses"]
    assert operation["responses"]["201"]["headers"]["Cache-Control"]["schema"]["const"] == "no-store"
    assert operation["responses"]["default"]["headers"]["Cache-Control"]["schema"]["const"] == "no-store"


def test_wp10_lifecycle_and_read_routes_are_rbac_closed_and_no_store(monkeypatch):
    client = _client(monkeypatch)
    admin = {"Authorization": "Bearer admin", "Host": "private.example"}
    editor = {"Authorization": "Bearer editor", "Host": "private.example"}
    assert client.post("/api/system/tenant/lifecycle/freeze", json={"expected_fence": "0", "idempotency_key": "freeze-1"}, headers=editor).status_code == 403
    rejected = client.post("/api/system/tenant/lifecycle/freeze", json={"tenant_id": "tenant-beta", "expected_fence": "0", "idempotency_key": "freeze-1"}, headers=admin)
    assert rejected.status_code == 400
    changed = client.post("/api/system/tenant/lifecycle/freeze", json={"expected_fence": "0", "idempotency_key": "freeze-1"}, headers=admin)
    assert changed.status_code == 200
    assert changed.headers["Cache-Control"] == "no-store"
    assert client.get("/api/system/tenant/status", headers=admin).status_code == 200
    assert client.get("/api/system/tenant/audit", headers=admin).headers["Cache-Control"] == "no-store"
    assert client.get("/api/system/tenant/manifests", headers=admin).status_code == 200