DataOps-platform/tests/test_wp12_metering_showback_api.py

"""Closed Flask/OpenAPI/RBAC contract for P3-WP12 Showback endpoints."""

from __future__ import annotations

from pathlib import Path

import yaml

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


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

    def record_for_principal(self, *, principal_id, body):
        type(self).calls.append({"record": principal_id, "body": body})
        return {"event_uid": body["event_uid"], "persisted_before_ack": True}

    def showback_for_principal(self, *, principal_id, window, kind):
        type(self).calls.append({"read": principal_id, "window": window, "kind": kind})
        return {"window": window, "source_micros": 1250000, "allocated_micros": 1250000, "difference_micros": 0}

    def publish_allocation_for_principal(self, *, principal_id, body):
        type(self).calls.append({"allocation": principal_id, "body": body})
        return {"rule_uid": body["rule_uid"], "rule_version": body["rule_version"], "persisted_before_ack": True}

    def evaluate_budget_for_principal(self, *, principal_id, body):
        type(self).calls.append({"budget": principal_id, "body": body})
        return {"budget_uid": body["budget_uid"], "alert_created": True, "provider": "disabled"}

    def allocation_replay_for_principal(self, *, principal_id, window, rule_uid, rule_version):
        type(self).calls.append({"allocation_replay": principal_id, "window": window, "rule_uid": rule_uid, "rule_version": rule_version})
        return {"rule_uid": rule_uid, "rule_version": rule_version, "source_micros": 1250000, "allocated_micros": 1250000, "difference_micros": 0}


EVENT = {
    "schema_version": 1,
    "event_uid": "api-event-001",
    "event_kind": "query",
    "occurred_at": "2026-08-18T00:00:00Z",
    "window_start": "2026-08-18T00:00:00Z",
    "window_end": "2026-08-18T00:05:00Z",
    "quantity": "1.250000",
    "unit": "gb",
    "idempotency_key": "api-idempotency-001",
    "evidence": {"digest": "a" * 64, "reference": "local-fixture://wp12/v1"},
    "mapping": {"department": "engineering", "project": "local-engineering", "cost_center": "local-fixture"},
}

ALLOCATION = {
    "schema_version": 1,
    "rule_uid": "api-allocation-001",
    "rule_version": 1,
    "effective_start": "2026-08-01T00:00:00Z",
    "effective_end": "2026-09-01T00:00:00Z",
    "mapping": {"department": "engineering", "project": "local-engineering", "cost_center": "local-fixture"},
    "allocations": [{"target": "local-engineering", "weight_micros": 1000000}],
}

BUDGET = {
    "schema_version": 1,
    "budget_uid": "api-budget-001",
    "window": "2026-08",
    "mapping": {"department": "engineering", "project": "local-engineering", "cost_center": "local-fixture"},
    "limit_micros": 2000000,
    "threshold_micros": 1000000,
}


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

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


def test_wp12_routes_are_rbac_closed_no_store_and_server_derived(monkeypatch):
    client = _client(monkeypatch)
    assert client.post("/api/system/metering/events", json=EVENT, headers={"Authorization": "Bearer viewer"}).status_code == 403
    rejected = client.post("/api/system/metering/events", json=EVENT | {"tenant_ref": "spoof"}, headers={"Authorization": "Bearer admin"})
    assert rejected.status_code == 400 and rejected.headers["Cache-Control"] == "no-store"
    created = client.post("/api/system/metering/events", json=EVENT, headers={"Authorization": "Bearer admin"})
    assert created.status_code == 201 and created.headers["Cache-Control"] == "no-store"
    assert _MeteringService.calls[0]["record"] == USER_UID
    shown = client.get("/api/system/metering/showback?window=2026-08", headers={"Authorization": "Bearer viewer"})
    assert shown.status_code == 200 and shown.headers["Cache-Control"] == "no-store"
    replay = client.get("/api/system/metering/allocation-replay?window=2026-08&rule_uid=api-allocation-001&rule_version=1", headers={"Authorization": "Bearer viewer"})
    assert replay.status_code == 200 and replay.headers["Cache-Control"] == "no-store"
    assert client.post("/api/system/metering/chargeback", json={}, headers={"Authorization": "Bearer admin"}).status_code == 404


def test_wp12_control_routes_are_closed_rbac_and_no_store(monkeypatch):
    client = _client(monkeypatch)
    assert client.post("/api/system/metering/allocations", json=ALLOCATION, headers={"Authorization": "Bearer viewer"}).status_code == 403
    rejected = client.post("/api/system/metering/allocations", json=ALLOCATION | {"tenant_ref": "spoof"}, headers={"Authorization": "Bearer admin"})
    assert rejected.status_code == 400 and rejected.headers["Cache-Control"] == "no-store"
    created = client.post("/api/system/metering/allocations", json=ALLOCATION, headers={"Authorization": "Bearer admin"})
    assert created.status_code == 201 and created.headers["Cache-Control"] == "no-store"
    budget = client.post("/api/system/metering/budgets", json=BUDGET, headers={"Authorization": "Bearer admin"})
    assert budget.status_code == 201 and budget.headers["Cache-Control"] == "no-store"
    assert _MeteringService.calls[0]["allocation"] == USER_UID
    assert _MeteringService.calls[1]["budget"] == USER_UID


def test_wp12_openapi_is_closed_permissioned_and_no_store():
    spec = yaml.safe_load(Path("docs/architecture/OPENAPI.yaml").read_text())
    event = spec["paths"]["/api/system/metering/events"]["post"]
    showback = spec["paths"]["/api/system/metering/showback"]["get"]
    allocation = spec["paths"]["/api/system/metering/allocations"]["post"]
    budget = spec["paths"]["/api/system/metering/budgets"]["post"]
    replay = spec["paths"]["/api/system/metering/allocation-replay"]["get"]
    assert event["x-required-permission"] == "metering:manage"
    assert showback["x-required-permission"] == "metering:read"
    assert event["requestBody"]["content"]["application/json"]["schema"]["additionalProperties"] is False
    assert event["responses"]["201"]["headers"]["Cache-Control"]["schema"]["const"] == "no-store"
    assert event["requestBody"]["required"] is True
    assert event["requestBody"]["content"]["application/json"]["schema"]["properties"]["mapping"]["additionalProperties"] is False
    assert event["requestBody"]["content"]["application/json"]["schema"]["properties"]["evidence"]["required"] == ["digest", "reference"]
    assert allocation["x-required-permission"] == "metering:manage"
    assert allocation["requestBody"]["content"]["application/json"]["schema"]["additionalProperties"] is False
    assert allocation["requestBody"]["required"] is True
    assert allocation["requestBody"]["content"]["application/json"]["schema"]["properties"]["allocations"]["items"]["additionalProperties"] is False
    assert budget["responses"]["201"]["headers"]["Cache-Control"]["schema"]["const"] == "no-store"
    assert replay["x-required-permission"] == "metering:read"
    assert {item["name"] for item in showback["parameters"]} == {"window"}
    assert all(item["required"] is True for item in replay["parameters"])