Provide first system administrator authority

2026-08-04 10:07:20 +02:00
parent e710cf5fb8
commit 668f7cf108
3 changed files with 263 additions and 2 deletions
@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+import unittest
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from govoplan_access.backend.db.base import AccessBase
+from govoplan_access.backend.db.models import (
+    Account,
+    Role,
+    SystemRoleAssignment,
+    User,
+    UserRoleAssignment,
+)
+from govoplan_access.backend.security.passwords import verify_password
+from govoplan_access.backend.tenancy.provisioning import LegacyFirstAdminProvisioner
+from govoplan_core.core.access import FirstAdminProvisioningError
+from govoplan_core.tenancy.scope import Tenant, create_scope_tables, scope_registry
+
+
+class FirstAdminProvisioningTests(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, expire_on_commit=False)
+        self.session = self.Session()
+        self.tenant = Tenant(id="tenant-1", slug="default", name="Default Tenant")
+        self.session.add(self.tenant)
+        self.session.flush()
+        self.provisioner = LegacyFirstAdminProvisioner()
+
+    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 test_creates_one_system_owner_with_a_login_membership(self) -> None:
+        created = self.provisioner.create_first_system_administrator(
+            self.session,
+            tenant=self.tenant,
+            email="Owner@Example.test",
+            display_name="System Owner",
+            password="a-production-password",
+        )
+        self.session.flush()
+
+        account = self.session.get(Account, created.account_id)
+        membership = self.session.get(User, created.membership_id)
+        self.assertIsNotNone(account)
+        self.assertIsNotNone(membership)
+        assert account is not None
+        assert membership is not None
+        self.assertTrue(verify_password("a-production-password", account.password_hash))
+        self.assertEqual(membership.tenant_id, self.tenant.id)
+        self.assertTrue(membership.is_tenant_admin)
+        system_role = (
+            self.session.query(Role)
+            .join(SystemRoleAssignment, SystemRoleAssignment.role_id == Role.id)
+            .filter(SystemRoleAssignment.account_id == account.id)
+            .one()
+        )
+        tenant_role = (
+            self.session.query(Role)
+            .join(UserRoleAssignment, UserRoleAssignment.role_id == Role.id)
+            .filter(UserRoleAssignment.user_id == membership.id)
+            .one()
+        )
+        self.assertEqual(system_role.slug, "system_owner")
+        self.assertEqual(tenant_role.slug, "owner")
+        self.assertTrue(
+            self.provisioner.has_durable_system_administrator(self.session)
+        )
+
+        with self.assertRaisesRegex(FirstAdminProvisioningError, "already exists"):
+            self.provisioner.create_first_system_administrator(
+                self.session,
+                tenant=self.tenant,
+                email="second@example.test",
+                display_name=None,
+                password="another-production-password",
+            )
+
+    def test_refuses_to_promote_or_reset_an_existing_account(self) -> None:
+        self.session.add(
+            Account(
+                email="existing@example.test",
+                normalized_email="existing@example.test",
+                is_active=True,
+            )
+        )
+        self.session.flush()
+
+        with self.assertRaisesRegex(FirstAdminProvisioningError, "already belongs"):
+            self.provisioner.create_first_system_administrator(
+                self.session,
+                tenant=self.tenant,
+                email="existing@example.test",
+                display_name=None,
+                password="a-production-password",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()