feat: add versioned organization model templates

This commit is contained in:
2026-07-30 17:42:07 +02:00
parent 5ebdffc0ec
commit 326bf3f56e
12 changed files with 1629 additions and 9 deletions

View File

@@ -0,0 +1,285 @@
from __future__ import annotations
import unittest
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker
from govoplan_core.db.base import Base
from govoplan_organizations.backend.api.v1.schemas import (
OrganizationModelTemplateDefinition,
)
from govoplan_organizations.backend.db.models import (
OrganizationFunction,
OrganizationFunctionType,
OrganizationModelInstantiation,
OrganizationModelTemplate,
OrganizationModelTemplateVersion,
OrganizationRelation,
OrganizationRelationType,
OrganizationStructure,
OrganizationUnit,
OrganizationUnitType,
)
from govoplan_organizations.backend.templates import (
OrganizationTemplateError,
canonical_template_definition,
instantiate_template_version,
)
TABLES = [
OrganizationModelTemplate.__table__,
OrganizationModelTemplateVersion.__table__,
OrganizationUnitType.__table__,
OrganizationStructure.__table__,
OrganizationRelationType.__table__,
OrganizationFunctionType.__table__,
OrganizationUnit.__table__,
OrganizationFunction.__table__,
OrganizationRelation.__table__,
OrganizationModelInstantiation.__table__,
]
class OrganizationModelTemplateTests(unittest.TestCase):
def setUp(self) -> None:
self.engine = create_engine("sqlite:///:memory:")
Base.metadata.create_all(self.engine, tables=TABLES)
self.session: Session = sessionmaker(
bind=self.engine,
expire_on_commit=False,
)()
def tearDown(self) -> None:
self.session.close()
Base.metadata.drop_all(self.engine, tables=reversed(TABLES))
self.engine.dispose()
def test_published_template_is_copied_into_tenant_owned_records(self) -> None:
definition, fingerprint = canonical_template_definition(_definition())
template = OrganizationModelTemplate(
id="template-1",
slug="municipality",
name="Municipality",
)
version = OrganizationModelTemplateVersion(
id="template-version-1",
template_id=template.id,
version="1.0.0",
status="published",
definition=definition,
definition_sha256=fingerprint,
)
self.session.add_all([template, version])
self.session.commit()
result = instantiate_template_version(
self.session,
tenant_id="tenant-1",
template=template,
version=version,
actor_account_id="account-1",
)
self.session.commit()
self.assertEqual(
{
"unit_types": 1,
"structures": 1,
"relation_types": 1,
"units": 2,
"relations": 1,
"function_types": 1,
"functions": 1,
},
result.object_counts,
)
office = (
self.session.query(OrganizationUnit)
.filter(
OrganizationUnit.tenant_id == "tenant-1",
OrganizationUnit.slug == "office",
)
.one()
)
root = (
self.session.query(OrganizationUnit)
.filter(
OrganizationUnit.tenant_id == "tenant-1",
OrganizationUnit.slug == "municipality",
)
.one()
)
self.assertEqual(root.id, office.parent_id)
self.assertEqual(
"tenant_owned_no_live_inheritance",
result.provenance["copy_semantics"],
)
self.assertEqual(
"1.0.0",
office.settings["template_provenance"]["template_version"],
)
version.definition = {}
self.session.commit()
self.assertEqual("Office", office.name)
def test_instantiation_refuses_implicit_merge_into_existing_model(self) -> None:
definition, fingerprint = canonical_template_definition(_definition())
template = OrganizationModelTemplate(
id="template-2",
slug="municipality-2",
name="Municipality",
)
version = OrganizationModelTemplateVersion(
id="template-version-2",
template_id=template.id,
version="1",
status="published",
definition=definition,
definition_sha256=fingerprint,
)
self.session.add_all(
[
template,
version,
OrganizationUnitType(
tenant_id="tenant-1",
slug="existing",
name="Existing",
),
]
)
self.session.commit()
with self.assertRaisesRegex(
OrganizationTemplateError,
"only into an empty tenant model",
):
instantiate_template_version(
self.session,
tenant_id="tenant-1",
template=template,
version=version,
actor_account_id="account-1",
)
def test_unknown_template_reference_is_rejected_before_storage(self) -> None:
definition = _definition().model_copy(deep=True)
definition.units[1].unit_type_slug = "unknown"
with self.assertRaisesRegex(
OrganizationTemplateError,
"unknown unit type",
):
canonical_template_definition(definition)
def test_parent_cycles_are_rejected_before_storage(self) -> None:
definition = _definition().model_copy(deep=True)
definition.units[0].parent_slug = "office"
with self.assertRaisesRegex(
OrganizationTemplateError,
"cycle in unit parent hierarchy",
):
canonical_template_definition(definition)
def test_relation_unit_type_mismatch_is_rejected(self) -> None:
definition = _definition().model_copy(deep=True)
definition.unit_types.append(
definition.unit_types[0].model_copy(
update={"slug": "other", "name": "Other"}
)
)
definition.units[1].unit_type_slug = "other"
with self.assertRaisesRegex(
OrganizationTemplateError,
"does not allow source unit",
):
canonical_template_definition(definition)
def test_function_unit_type_mismatch_is_rejected(self) -> None:
definition = _definition().model_copy(deep=True)
definition.unit_types.append(
definition.unit_types[0].model_copy(
update={"slug": "other", "name": "Other"}
)
)
definition.units[1].unit_type_slug = "other"
definition.relations = []
with self.assertRaisesRegex(
OrganizationTemplateError,
"does not apply to unit",
):
canonical_template_definition(definition)
def _definition() -> OrganizationModelTemplateDefinition:
return OrganizationModelTemplateDefinition.model_validate(
{
"unit_types": [
{
"slug": "administrative-unit",
"name": "Administrative unit",
}
],
"structures": [
{
"slug": "administrative",
"name": "Administrative hierarchy",
}
],
"relation_types": [
{
"slug": "reports-to",
"name": "Reports to",
"structure_slug": "administrative",
"source_unit_type_slug": "administrative-unit",
"target_unit_type_slug": "administrative-unit",
}
],
"units": [
{
"slug": "municipality",
"name": "Municipality",
"unit_type_slug": "administrative-unit",
},
{
"slug": "office",
"name": "Office",
"unit_type_slug": "administrative-unit",
"parent_slug": "municipality",
},
],
"relations": [
{
"structure_slug": "administrative",
"relation_type_slug": "reports-to",
"source_unit_slug": "office",
"target_unit_slug": "municipality",
}
],
"function_types": [
{
"slug": "head",
"name": "Head",
"organization_unit_type_slug": "administrative-unit",
}
],
"functions": [
{
"slug": "head",
"name": "Office head",
"function_type_slug": "head",
"organization_unit_slug": "office",
}
],
}
)
if __name__ == "__main__":
unittest.main()