test: cover organization administration workflows

2026-07-31 04:21:34 +02:00
parent 06bd1c9003
commit 17ab9c9c8d
2 changed files with 202 additions and 0 deletions
@@ -0,0 +1,166 @@
+from __future__ import annotations
+
+import unittest
+from types import SimpleNamespace
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from govoplan_core.core.change_sequence import ChangeSequenceEntry
+from govoplan_core.core.events import EventBus, event_bus_context
+from govoplan_core.db.base import Base
+from govoplan_organizations.backend.api.v1.routes import (
+    create_function,
+    create_unit,
+    get_organization_model,
+    update_function,
+    update_unit,
+)
+from govoplan_organizations.backend.api.v1.schemas import (
+    FunctionCreateRequest,
+    FunctionUpdateRequest,
+    UnitCreateRequest,
+    UnitUpdateRequest,
+)
+from govoplan_organizations.backend.db.models import (
+    OrganizationFunction,
+    OrganizationFunctionType,
+    OrganizationRelation,
+    OrganizationRelationType,
+    OrganizationStructure,
+    OrganizationTenantSettings,
+    OrganizationUnit,
+    OrganizationUnitType,
+)
+
+
+class OrganizationAdministrationRouteTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        self.tables = [
+            ChangeSequenceEntry.__table__,
+            OrganizationUnitType.__table__,
+            OrganizationStructure.__table__,
+            OrganizationRelationType.__table__,
+            OrganizationUnit.__table__,
+            OrganizationRelation.__table__,
+            OrganizationFunctionType.__table__,
+            OrganizationFunction.__table__,
+            OrganizationTenantSettings.__table__,
+        ]
+        Base.metadata.create_all(self.engine, tables=self.tables)
+        self.Session = sessionmaker(bind=self.engine, expire_on_commit=False)
+        self.session: Session = self.Session()
+        self.principal = SimpleNamespace(
+            tenant_id="tenant-1",
+            account_id="account-1",
+            membership_id="membership-1",
+            scopes=frozenset(),
+        )
+
+    def tearDown(self) -> None:
+        self.session.close()
+        Base.metadata.drop_all(self.engine, tables=list(reversed(self.tables)))
+        self.engine.dispose()
+
+    def test_unit_tree_and_function_create_update_list_deactivate(self) -> None:
+        events: list[object] = []
+        bus = EventBus()
+        bus.subscribe("*", events.append)
+
+        with event_bus_context(bus):
+            root = create_unit(
+                UnitCreateRequest(name="Central administration", slug="central"),
+                session=self.session,
+                principal=self.principal,
+            )
+            child = create_unit(
+                UnitCreateRequest(
+                    name="Procurement",
+                    slug="procurement",
+                    parent_id=root.id,
+                ),
+                session=self.session,
+                principal=self.principal,
+            )
+            function = create_function(
+                FunctionCreateRequest(
+                    name="Procurement lead",
+                    slug="lead",
+                    organization_unit_id=child.id,
+                    delegable=True,
+                ),
+                session=self.session,
+                principal=self.principal,
+            )
+
+            updated_child = update_unit(
+                child.id,
+                UnitUpdateRequest(
+                    name="Strategic procurement",
+                    is_active=False,
+                ),
+                session=self.session,
+                principal=self.principal,
+            )
+            updated_function = update_function(
+                function.id,
+                FunctionUpdateRequest(is_active=False),
+                session=self.session,
+                principal=self.principal,
+            )
+
+        model = get_organization_model(
+            session=self.session,
+            principal=self.principal,
+        )
+
+        self.assertEqual(root.id, next(item for item in model.units if item.id == child.id).parent_id)
+        self.assertEqual("Strategic procurement", updated_child.name)
+        self.assertFalse(updated_child.is_active)
+        self.assertFalse(updated_function.is_active)
+        self.assertEqual(child.id, updated_function.organization_unit_id)
+        self.assertEqual(2, len(model.units))
+        self.assertEqual(1, len(model.functions))
+        self.assertEqual(
+            [
+                "organizations.unit.created.v1",
+                "organizations.unit.created.v1",
+                "organizations.function.created.v1",
+                "organizations.unit.deactivated.v1",
+                "organizations.function.deactivated.v1",
+            ],
+            [
+                event.type
+                for event in events
+                if event.module_id == "organizations"
+            ],
+        )
+
+    def test_model_listing_is_tenant_scoped(self) -> None:
+        create_unit(
+            UnitCreateRequest(name="Visible unit", slug="visible"),
+            session=self.session,
+            principal=self.principal,
+        )
+        create_unit(
+            UnitCreateRequest(name="Other unit", slug="other"),
+            session=self.session,
+            principal=SimpleNamespace(
+                tenant_id="tenant-2",
+                account_id="account-2",
+                membership_id="membership-2",
+                scopes=frozenset(),
+            ),
+        )
+
+        model = get_organization_model(
+            session=self.session,
+            principal=self.principal,
+        )
+
+        self.assertEqual(["Visible unit"], [item.name for item in model.units])
+
+
+if __name__ == "__main__":
+    unittest.main()