Resolve current principals for automation

2026-07-28 15:03:50 +02:00
parent f1d64d247e
commit fdfcfbb440
4 changed files with 338 additions and 1 deletions
--- a/tests/test_automation_principal.py
+++ b/tests/test_automation_principal.py
@@ -0,0 +1,162 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_access.backend.auth.dependencies import (
+    AccessAutomationPrincipalProvider,
+)
+from govoplan_access.backend.db.base import AccessBase
+from govoplan_access.backend.db.models import Account, User
+from govoplan_access.backend.manifest import manifest
+from govoplan_access.backend.security.sessions import (
+    UserAuthorizationContext,
+)
+from govoplan_core.auth import ApiPrincipal
+from govoplan_core.core.access import (
+    CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
+)
+from govoplan_core.core.automation import AutomationPrincipalRequest
+from govoplan_core.tenancy.scope import (
+    Tenant,
+    create_scope_tables,
+    scope_registry,
+)
+
+
+class AutomationPrincipalTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        create_scope_tables(self.engine)
+        AccessBase.metadata.create_all(bind=self.engine)
+        self.Session = sessionmaker(bind=self.engine)
+        self.session = self.Session()
+        self.account = Account(
+            id="account-1",
+            email="owner@example.test",
+            normalized_email="owner@example.test",
+        )
+        self.tenant = Tenant(
+            id="tenant-1",
+            slug="tenant-1",
+            name="Tenant 1",
+        )
+        self.user = User(
+            id="user-1",
+            tenant_id=self.tenant.id,
+            account_id=self.account.id,
+            email=self.account.email,
+        )
+        self.session.add_all([self.tenant, self.account, self.user])
+        self.session.commit()
+        self.provider = AccessAutomationPrincipalProvider()
+
+    def tearDown(self) -> None:
+        self.session.close()
+        AccessBase.metadata.drop_all(bind=self.engine)
+        scope_registry.metadata.drop_all(bind=self.engine)
+        self.engine.dispose()
+
+    def _request(self) -> AutomationPrincipalRequest:
+        return AutomationPrincipalRequest(
+            tenant_id=self.tenant.id,
+            account_id=self.account.id,
+            membership_id=self.user.id,
+            authorization_ref="dataflow-trigger:1",
+            grant_scopes=(
+                "dataflow:pipeline:run",
+                "datasources:catalogue:read",
+            ),
+        )
+
+    def test_resolution_intersects_trigger_grant_with_current_scopes(self) -> None:
+        context = UserAuthorizationContext(
+            tenant_roles=[],
+            system_roles=[],
+            groups=[],
+            function_assignment_ids=(),
+            function_delegation_ids=(),
+            scopes=[
+                "dataflow:pipeline:run",
+                "datasources:catalogue:read",
+                "system:settings:write",
+            ],
+        )
+        with patch(
+            "govoplan_access.backend.auth.dependencies."
+            "collect_user_authorization_context",
+            return_value=context,
+        ):
+            result = self.provider.resolve_automation_principal(
+                self.session,
+                request=self._request(),
+            )
+
+        self.assertTrue(result.allowed)
+        self.assertIsInstance(result.principal, ApiPrincipal)
+        self.assertEqual(
+            frozenset(
+                {
+                    "dataflow:pipeline:run",
+                    "datasources:catalogue:read",
+                }
+            ),
+            result.principal.scopes,
+        )
+        self.assertEqual(
+            "dataflow-trigger:1",
+            result.principal.principal.service_account_id,
+        )
+        self.assertNotIn(
+            "system:settings:write",
+            result.principal.scopes,
+        )
+
+    def test_revoked_scope_and_suspended_owner_fail_closed(self) -> None:
+        context = UserAuthorizationContext(
+            tenant_roles=[],
+            system_roles=[],
+            groups=[],
+            function_assignment_ids=(),
+            function_delegation_ids=(),
+            scopes=["dataflow:pipeline:run"],
+        )
+        with patch(
+            "govoplan_access.backend.auth.dependencies."
+            "collect_user_authorization_context",
+            return_value=context,
+        ):
+            result = self.provider.resolve_automation_principal(
+                self.session,
+                request=self._request(),
+            )
+        self.assertFalse(result.allowed)
+        self.assertEqual(
+            ("datasources:catalogue:read",),
+            result.missing_scopes,
+        )
+
+        self.account.is_active = False
+        self.session.flush()
+        suspended = self.provider.resolve_automation_principal(
+            self.session,
+            request=self._request(),
+        )
+        self.assertFalse(suspended.allowed)
+        self.assertEqual(
+            "inactive_or_inconsistent",
+            suspended.provenance["status"],
+        )
+
+    def test_manifest_registers_automation_resolution(self) -> None:
+        self.assertIn(
+            CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
+            manifest.capability_factories,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()