108 lines
3.9 KiB
Python
108 lines
3.9 KiB
Python
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()
|