DataOps-platform/tests/test_domain_template_api.py

from __future__ import annotations

ACTOR_UID = "01900000-0000-7000-8000-000000000101"


class FakeService:
    def __init__(self):
        self.calls = []

    def list_templates(self):
        self.calls.append(("list",))
        return [{"template_code": "spare_parts", "current_version": 1}]

    def get_template(self, template_code):
        self.calls.append(("get", template_code))
        return {"template_code": template_code, "current_version": 1}

    def dry_run(self, definition):
        self.calls.append(("dry_run", definition))
        return {"valid": True, "diff": {"added_object_types": ["spare_part"]}}

    def import_template(self, definition, *, actor_uid):
        self.calls.append(("import", definition, actor_uid))
        return {"template_code": definition["template_code"], "current_version": 1}

    def rollback(self, template_code, *, target_version, actor_uid):
        self.calls.append(("rollback", template_code, target_version, actor_uid))
        return {"template_code": template_code, "current_version": 2}

    def list_imports(self, template_code):
        self.calls.append(("audits", template_code))
        return [{"operation": "import", "template_code": template_code}]


def _headers(role):
    return {"Authorization": f"Bearer {role}"}


def test_domain_template_api_is_readable_and_admin_managed(monkeypatch):
    from app import create_app
    from app.api.meta_data import domain_templates

    service = FakeService()
    monkeypatch.setattr(domain_templates, "_service", lambda: service)
    monkeypatch.setattr(
        "app.core.system.auth.load_identity_from_token",
        lambda token, secret: (
            {"id": ACTOR_UID, "username": token, "roles": [token]}
            if token in {"viewer", "editor", "admin"}
            else None
        ),
    )
    app = create_app()
    app.config.update(TESTING=True)
    client = app.test_client()

    listed = client.get("/api/meta/domain-templates", headers=_headers("viewer"))
    assert listed.status_code == 200

    preview = client.post(
        "/api/meta/domain-templates/dry-run",
        json={"template_code": "spare_parts"},
        headers=_headers("editor"),
    )
    assert preview.status_code == 200

    forbidden = client.post(
        "/api/meta/domain-templates/import",
        json={"template_code": "spare_parts"},
        headers=_headers("editor"),
    )
    assert forbidden.status_code == 403

    imported = client.post(
        "/api/meta/domain-templates/import",
        json={"template_code": "spare_parts"},
        headers=_headers("admin"),
    )
    assert imported.status_code == 200
    assert service.calls[-1][2] == ACTOR_UID

    rolled_back = client.post(
        "/api/meta/domain-templates/spare_parts/rollback",
        json={"target_version": 1},
        headers=_headers("admin"),
    )
    assert rolled_back.status_code == 200


def test_domain_template_paths_have_dedicated_permissions():
    from app.core.system.permissions import (
        DOMAIN_TEMPLATES_MANAGE,
        DOMAIN_TEMPLATES_PREVIEW,
        DOMAIN_TEMPLATES_READ,
        permission_for_request,
    )

    assert permission_for_request("/api/meta/domain-templates", "GET") == (
        DOMAIN_TEMPLATES_READ,
    )
    assert permission_for_request("/api/meta/domain-templates/dry-run", "POST") == (
        DOMAIN_TEMPLATES_PREVIEW,
    )
    assert permission_for_request("/api/meta/domain-templates/import", "POST") == (
        DOMAIN_TEMPLATES_MANAGE,
    )