167 lines
5.5 KiB
Python
167 lines
5.5 KiB
Python
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()
|