537 lines
18 KiB
Python
537 lines
18 KiB
Python
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,
|
|
OrganizationModelUpgrade,
|
|
OrganizationRelation,
|
|
OrganizationRelationType,
|
|
OrganizationStructure,
|
|
OrganizationUnit,
|
|
OrganizationUnitType,
|
|
)
|
|
from govoplan_organizations.backend.templates import (
|
|
OrganizationTemplateError,
|
|
canonical_template_definition,
|
|
instantiate_template_version,
|
|
)
|
|
from govoplan_organizations.backend.upgrades import (
|
|
OrganizationUpgradeError,
|
|
apply_model_upgrade,
|
|
cancel_model_upgrade,
|
|
create_model_upgrade_preview,
|
|
current_model_instantiation,
|
|
)
|
|
|
|
|
|
TABLES = [
|
|
OrganizationModelTemplate.__table__,
|
|
OrganizationModelTemplateVersion.__table__,
|
|
OrganizationUnitType.__table__,
|
|
OrganizationStructure.__table__,
|
|
OrganizationRelationType.__table__,
|
|
OrganizationFunctionType.__table__,
|
|
OrganizationUnit.__table__,
|
|
OrganizationFunction.__table__,
|
|
OrganizationRelation.__table__,
|
|
OrganizationModelInstantiation.__table__,
|
|
OrganizationModelUpgrade.__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 test_three_way_upgrade_previews_additions_and_local_divergence(self) -> None:
|
|
template, source, target = self._instantiated_upgrade_fixture(
|
|
mutate_target=lambda definition: definition.units.append(
|
|
definition.units[1].model_copy(
|
|
update={
|
|
"slug": "service-office",
|
|
"name": "Service office",
|
|
"parent_slug": "municipality",
|
|
}
|
|
)
|
|
)
|
|
)
|
|
office = (
|
|
self.session.query(OrganizationUnit)
|
|
.filter_by(tenant_id="tenant-1", slug="office")
|
|
.one()
|
|
)
|
|
office.name = "Tenant-specific office"
|
|
self.session.commit()
|
|
|
|
preview = create_model_upgrade_preview(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
target_version=target,
|
|
actor_account_id="account-1",
|
|
idempotency_key="preview-additive",
|
|
)
|
|
self.session.commit()
|
|
|
|
self.assertEqual(1, preview.preview["counts"]["compatible_addition"])
|
|
self.assertEqual(1, preview.preview["counts"]["local_divergence"])
|
|
self.assertEqual(0, preview.preview["requires_decisions"])
|
|
applied, instantiation = apply_model_upgrade(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
upgrade_id=preview.id,
|
|
expected_revision=1,
|
|
decisions={},
|
|
actor_account_id="account-2",
|
|
)
|
|
self.session.commit()
|
|
|
|
self.assertEqual("applied", applied.status)
|
|
self.assertEqual(target.id, instantiation.template_version_id)
|
|
self.assertEqual(
|
|
"Tenant-specific office",
|
|
self.session.query(OrganizationUnit)
|
|
.filter_by(tenant_id="tenant-1", slug="office")
|
|
.one()
|
|
.name,
|
|
)
|
|
self.assertIsNotNone(
|
|
self.session.query(OrganizationUnit)
|
|
.filter_by(tenant_id="tenant-1", slug="service-office")
|
|
.one_or_none()
|
|
)
|
|
self.assertEqual("superseded", source.status)
|
|
self.assertEqual(instantiation.id, current_model_instantiation(self.session, tenant_id="tenant-1").id)
|
|
self.assertEqual(template.id, instantiation.template_id)
|
|
|
|
def test_divergent_upgrade_requires_bounded_decision_and_detects_stale_state(self) -> None:
|
|
_template, _source, target = self._instantiated_upgrade_fixture(
|
|
mutate_target=lambda definition: setattr(
|
|
definition.units[1], "name", "Template office"
|
|
)
|
|
)
|
|
office = self.session.query(OrganizationUnit).filter_by(
|
|
tenant_id="tenant-1", slug="office"
|
|
).one()
|
|
office.name = "Local office"
|
|
self.session.commit()
|
|
preview = create_model_upgrade_preview(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
target_version=target,
|
|
actor_account_id="account-1",
|
|
idempotency_key="preview-divergent",
|
|
)
|
|
self.session.commit()
|
|
conflict = next(
|
|
entry
|
|
for entry in preview.preview["entries"]
|
|
if entry["id"] == "units:office"
|
|
)
|
|
self.assertTrue(conflict["requires_decision"])
|
|
with self.assertRaisesRegex(OrganizationUpgradeError, "decision is required"):
|
|
apply_model_upgrade(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
upgrade_id=preview.id,
|
|
expected_revision=1,
|
|
decisions={},
|
|
actor_account_id="account-2",
|
|
)
|
|
self.session.rollback()
|
|
|
|
office = self.session.query(OrganizationUnit).filter_by(
|
|
tenant_id="tenant-1", slug="office"
|
|
).one()
|
|
office.description = "Changed after preview"
|
|
self.session.commit()
|
|
with self.assertRaisesRegex(OrganizationUpgradeError, "changed after this preview"):
|
|
apply_model_upgrade(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
upgrade_id=preview.id,
|
|
expected_revision=1,
|
|
decisions={"units:office": {"action": "use_target"}},
|
|
actor_account_id="account-2",
|
|
)
|
|
|
|
def test_divergent_upgrade_applies_explicit_target_decision(self) -> None:
|
|
_template, _source, target = self._instantiated_upgrade_fixture(
|
|
mutate_target=lambda definition: setattr(
|
|
definition.units[1], "name", "Template office"
|
|
)
|
|
)
|
|
office = self.session.query(OrganizationUnit).filter_by(
|
|
tenant_id="tenant-1", slug="office"
|
|
).one()
|
|
office.name = "Local office"
|
|
self.session.commit()
|
|
preview = create_model_upgrade_preview(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
target_version=target,
|
|
actor_account_id="account-1",
|
|
idempotency_key="preview-explicit-decision",
|
|
)
|
|
self.session.commit()
|
|
applied, _instantiation = apply_model_upgrade(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
upgrade_id=preview.id,
|
|
expected_revision=1,
|
|
decisions={"units:office": {"action": "use_target"}},
|
|
actor_account_id="account-2",
|
|
)
|
|
self.session.commit()
|
|
self.assertEqual("applied", applied.status)
|
|
self.assertEqual(
|
|
"Template office",
|
|
self.session.query(OrganizationUnit)
|
|
.filter_by(tenant_id="tenant-1", slug="office")
|
|
.one()
|
|
.name,
|
|
)
|
|
self.assertEqual(
|
|
{"action": "use_target"},
|
|
applied.decisions["units:office"],
|
|
)
|
|
|
|
def test_invalid_local_references_block_and_preview_can_be_cancelled(self) -> None:
|
|
_template, _source, target = self._instantiated_upgrade_fixture()
|
|
office = self.session.query(OrganizationUnit).filter_by(
|
|
tenant_id="tenant-1", slug="office"
|
|
).one()
|
|
office.unit_type_id = "missing-unit-type"
|
|
self.session.commit()
|
|
preview = create_model_upgrade_preview(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
target_version=target,
|
|
actor_account_id="account-1",
|
|
idempotency_key="preview-invalid",
|
|
)
|
|
self.session.commit()
|
|
self.assertGreater(preview.preview["blocking_invalid_references"], 0)
|
|
with self.assertRaisesRegex(OrganizationUpgradeError, "Invalid tenant references"):
|
|
apply_model_upgrade(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
upgrade_id=preview.id,
|
|
expected_revision=1,
|
|
decisions={},
|
|
actor_account_id="account-2",
|
|
)
|
|
self.session.rollback()
|
|
cancelled = cancel_model_upgrade(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
upgrade_id=preview.id,
|
|
expected_revision=1,
|
|
actor_account_id="account-1",
|
|
)
|
|
self.session.commit()
|
|
self.assertEqual("cancelled", cancelled.status)
|
|
self.assertEqual(2, cancelled.revision)
|
|
self.assertIsNone(cancelled.applied_at)
|
|
|
|
def test_unchanged_upgrade_preview_has_no_changes(self) -> None:
|
|
_template, _source, target = self._instantiated_upgrade_fixture()
|
|
preview = create_model_upgrade_preview(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
target_version=target,
|
|
actor_account_id="account-1",
|
|
idempotency_key="preview-unchanged",
|
|
)
|
|
self.assertEqual([], preview.preview["entries"])
|
|
self.assertEqual(0, preview.preview["requires_decisions"])
|
|
|
|
def _instantiated_upgrade_fixture(self, mutate_target=None):
|
|
source_definition = _definition()
|
|
source_payload, source_hash = canonical_template_definition(source_definition)
|
|
target_definition = source_definition.model_copy(deep=True)
|
|
if mutate_target is not None:
|
|
mutate_target(target_definition)
|
|
target_payload, target_hash = canonical_template_definition(target_definition)
|
|
template = OrganizationModelTemplate(
|
|
id="template-upgrade",
|
|
slug="upgrade-template",
|
|
name="Upgrade template",
|
|
)
|
|
source_version = OrganizationModelTemplateVersion(
|
|
id="template-upgrade-v1",
|
|
template_id=template.id,
|
|
version="1.0.0",
|
|
status="published",
|
|
definition=source_payload,
|
|
definition_sha256=source_hash,
|
|
)
|
|
target_version = OrganizationModelTemplateVersion(
|
|
id="template-upgrade-v2",
|
|
template_id=template.id,
|
|
version="2.0.0",
|
|
status="published",
|
|
definition=target_payload,
|
|
definition_sha256=target_hash,
|
|
)
|
|
self.session.add_all([template, source_version, target_version])
|
|
self.session.flush()
|
|
source = instantiate_template_version(
|
|
self.session,
|
|
tenant_id="tenant-1",
|
|
template=template,
|
|
version=source_version,
|
|
actor_account_id="account-1",
|
|
)
|
|
self.session.commit()
|
|
return template, source, target_version
|
|
|
|
|
|
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()
|