Release v0.1.5

src/govoplan_admin/backend/governance.py

@@ -0,0 +1,169 @@
+from __future__ import annotations
+
+from sqlalchemy.orm import Session
+
+from govoplan_admin.backend.db.models import GovernanceTemplate, GovernanceTemplateAssignment
+from govoplan_core.admin.common import AdminConflictError, AdminValidationError, slugify
+from govoplan_core.core.access import (
+    CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER,
+    AccessGovernanceMaterializer,
+    GovernanceTemplateMaterialization,
+)
+from govoplan_core.core.runtime import get_registry
+from govoplan_core.security.permissions import validate_tenant_permissions
+from govoplan_tenancy.backend.db.models import Tenant
+
+TEMPLATE_KINDS = {"group", "role"}
+ASSIGNMENT_MODES = {"available", "required"}
+
+
+def _governance_materializer() -> AccessGovernanceMaterializer:
+    registry = get_registry()
+    if registry is None or not registry.has_capability(CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER):
+        raise AdminValidationError("Access governance materializer capability is not configured.")
+    capability = registry.require_capability(CAPABILITY_ACCESS_GOVERNANCE_MATERIALIZER)
+    if not isinstance(capability, AccessGovernanceMaterializer):
+        raise AdminValidationError("Access governance materializer capability is invalid.")
+    return capability
+
+
+def _materialization(
+    item: GovernanceTemplate,
+    *,
+    tenant_id: str,
+    required: bool = False,
+) -> GovernanceTemplateMaterialization:
+    return GovernanceTemplateMaterialization(
+        template_id=item.id,
+        kind=item.kind,  # type: ignore[arg-type]
+        tenant_id=tenant_id,
+        slug=item.slug,
+        name=item.name,
+        description=item.description,
+        permissions=tuple(item.permissions or []),
+        is_active=item.is_active,
+        required=required,
+    )
+
+
+def validate_template(kind: str, permissions: list[str]) -> list[str]:
+    if kind not in TEMPLATE_KINDS:
+        raise AdminValidationError("Template kind must be group or role.")
+    if kind == "group":
+        if permissions:
+            raise AdminValidationError("Group templates do not contain permissions; assign roles to groups inside each tenant.")
+        return []
+    try:
+        return validate_tenant_permissions(permissions)
+    except ValueError as exc:
+        raise AdminValidationError(str(exc)) from exc
+
+
+def create_template(
+    session: Session,
+    *,
+    kind: str,
+    slug: str,
+    name: str,
+    description: str | None,
+    permissions: list[str],
+    is_active: bool,
+    assignments: list[dict[str, str]],
+) -> GovernanceTemplate:
+    normalized_slug = slugify(slug)
+    permissions = validate_template(kind, permissions)
+    exists = session.query(GovernanceTemplate).filter(
+        GovernanceTemplate.kind == kind,
+        GovernanceTemplate.slug == normalized_slug,
+    ).first()
+    if exists:
+        raise AdminConflictError(f"A {kind} template with slug {normalized_slug!r} already exists.")
+    item = GovernanceTemplate(
+        kind=kind,
+        slug=normalized_slug,
+        name=name.strip(),
+        description=description,
+        permissions=permissions,
+        is_active=is_active,
+    )
+    session.add(item)
+    session.flush()
+    set_template_assignments(session, item, assignments)
+    return item
+
+
+def update_template(
+    session: Session,
+    item: GovernanceTemplate,
+    *,
+    name: str,
+    description: str | None,
+    permissions: list[str],
+    is_active: bool,
+    assignments: list[dict[str, str]],
+) -> GovernanceTemplate:
+    item.name = name.strip()
+    item.description = description
+    item.permissions = validate_template(item.kind, permissions)
+    item.is_active = is_active
+    set_template_assignments(session, item, assignments)
+    sync_template(session, item)
+    return item
+
+
+def set_template_assignments(
+    session: Session,
+    item: GovernanceTemplate,
+    assignments: list[dict[str, str]],
+) -> None:
+    desired: dict[str, str] = {}
+    for assignment in assignments:
+        tenant_id = assignment.get("tenant_id", "")
+        mode = assignment.get("mode", "available")
+        if mode not in ASSIGNMENT_MODES:
+            raise AdminValidationError("Template assignment mode must be available or required.")
+        tenant = session.get(Tenant, tenant_id)
+        if tenant is None:
+            raise AdminValidationError(f"Unknown tenant: {tenant_id}")
+        desired[tenant_id] = mode
+
+    existing = {
+        row.tenant_id: row
+        for row in session.query(GovernanceTemplateAssignment)
+        .filter(GovernanceTemplateAssignment.template_id == item.id)
+        .all()
+    }
+    for tenant_id, row in list(existing.items()):
+        if tenant_id in desired:
+            row.mode = desired[tenant_id]
+            continue
+        _governance_materializer().remove_template(session, _materialization(item, tenant_id=tenant_id))
+        session.delete(row)
+
+    for tenant_id, mode in desired.items():
+        if tenant_id not in existing:
+            session.add(GovernanceTemplateAssignment(template_id=item.id, tenant_id=tenant_id, mode=mode))
+    session.flush()
+    sync_template(session, item)
+
+
+def sync_template(session: Session, item: GovernanceTemplate) -> None:
+    assignments = session.query(GovernanceTemplateAssignment).filter(
+        GovernanceTemplateAssignment.template_id == item.id
+    ).all()
+    for assignment in assignments:
+        required = assignment.mode == "required"
+        _governance_materializer().sync_template(
+            session,
+            _materialization(item, tenant_id=assignment.tenant_id, required=required),
+        )
+    session.flush()
+
+
+def delete_template(session: Session, item: GovernanceTemplate) -> None:
+    assignments = session.query(GovernanceTemplateAssignment).filter(
+        GovernanceTemplateAssignment.template_id == item.id
+    ).all()
+    for assignment in assignments:
+        _governance_materializer().remove_template(session, _materialization(item, tenant_id=assignment.tenant_id))
+    session.delete(item)