DataOps-platform/tests/test_wp13_plugin_platform_api.py

"""Flask/RBAC/openapi contract for the closed WP13 plugin gateway."""

from __future__ import annotations

from app.core.plugins.governance import fixture_signature_digest

MANIFEST = {
    "schema_version": 1, "plugin_uid": "fixture-api-quality", "name": "fixture-api-quality", "version": "1.0.0", "api_version": "1",
    "type": "quality", "capabilities": ["evaluate"],
    "permissions": {"network": False, "file": False, "secret": False, "child_process": False},
    "resource": {"timeout_ms": 100, "max_output_bytes": 4096, "max_concurrency": 1, "max_retries": 1},
    "compatibility": {"platform_api": "1"},
    "distribution": {"kind": "builtin_fixture", "fixture_id": "ENGINEERING_EVIDENCE_ONLY", "artifact_digest": "a" * 64},
}
RECORD = {
    "artifact_digest": "a" * 64,
    "signature": {"trust_store_key_id": "local-fixture-key-v1", "signature_digest": fixture_signature_digest("a" * 64)},
    "sbom_digest": "b" * 64, "license_digest": "c" * 64, "vulnerability_digest": "d" * 64, "provenance_digest": "e" * 64,
}


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

    def register(self, **kwargs):
        type(self).calls.append(kwargs)
        return {"state": "draft", "plugin_uid": kwargs["manifest"]["plugin_uid"], "version": "1.0.0"}


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

    _Gateway.calls = []
    monkeypatch.setattr(plugin_platform_routes, "DatabasePluginPlatformService", _Gateway)
    monkeypatch.setattr(
        "app.core.system.auth.load_identity_from_token",
        lambda token, secret: {"id": "01900000-0000-7000-8000-000000000013", "roles": [token]} if token in {"viewer", "admin"} else None,
    )
    app = create_app()
    app.config.update(TESTING=True)
    return app.test_client()


def test_wp13_registry_route_is_closed_rbac_no_store_and_server_principal(monkeypatch):
    client = _client(monkeypatch)
    assert client.post("/api/system/plugins/registry", json={"manifest": MANIFEST, "registry_record": RECORD}, headers={"Authorization": "Bearer viewer"}).status_code == 403
    rejected = client.post("/api/system/plugins/registry", json={"manifest": MANIFEST | {"url": "https://invalid"}, "registry_record": RECORD}, headers={"Authorization": "Bearer admin"})
    assert rejected.status_code == 400 and rejected.headers["Cache-Control"] == "no-store"
    accepted = client.post("/api/system/plugins/registry", json={"manifest": MANIFEST, "registry_record": RECORD}, headers={"Authorization": "Bearer admin"})
    assert accepted.status_code == 201 and accepted.headers["Cache-Control"] == "no-store"
    assert _Gateway.calls[0]["actor_ref"] == "01900000-0000-7000-8000-000000000013"