From 50a8d459e7fd9d64173cb3d57af66d0ac4a08046 Mon Sep 17 00:00:00 2001 From: Albrecht Degering Date: Fri, 31 Jul 2026 18:07:36 +0200 Subject: [PATCH] Add IDM assignment lifecycle worker contract --- src/govoplan_core/celery_app.py | 53 ++++++++++ src/govoplan_core/core/idm.py | 34 +++++++ tests/test_idm_assignment_lifecycle_worker.py | 97 +++++++++++++++++++ 3 files changed, 184 insertions(+) create mode 100644 tests/test_idm_assignment_lifecycle_worker.py diff --git a/src/govoplan_core/celery_app.py b/src/govoplan_core/celery_app.py index 05b8aa0..767dd15 100644 --- a/src/govoplan_core/celery_app.py +++ b/src/govoplan_core/celery_app.py @@ -19,6 +19,10 @@ from govoplan_core.core.events import ( PlatformEventOutbox, publish_platform_event, ) +from govoplan_core.core.idm import ( + CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE, + IdmAssignmentLifecycle, +) from govoplan_core.core.module_management import load_startup_enabled_modules, startup_candidate_module_ids from govoplan_core.core.mail import CAPABILITY_MAIL_DELIVERY_OUTBOX, MailDeliveryOutboxProvider from govoplan_core.core.modules import ModuleContext @@ -62,6 +66,7 @@ celery.conf.update( "govoplan.postbox.dispatch_routes": {"queue": "postbox"}, "govoplan.events.dispatch_outbox": {"queue": "events"}, "govoplan.events.purge_outbox": {"queue": "events"}, + "govoplan.idm.expire_assignments": {"queue": "idm"}, }, worker_prefetch_multiplier=1, task_acks_late=True, @@ -117,6 +122,11 @@ celery.conf.update( "schedule": 24 * 60 * 60.0, "args": (500,), }, + "idm-assignment-expiry-every-minute": { + "task": "govoplan.idm.expire_assignments", + "schedule": 60.0, + "args": (None, 100), + }, }, ) @@ -238,6 +248,20 @@ def _platform_event_outbox( return capability +def _idm_assignment_lifecycle( + registry: PlatformRegistry | None = None, +) -> IdmAssignmentLifecycle | None: + registry = registry or _platform_registry() + if not registry.has_capability(CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE): + return None + capability = registry.require_capability( + CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE + ) + if not isinstance(capability, IdmAssignmentLifecycle): + raise RuntimeError("IDM assignment lifecycle capability is invalid") + return capability + + @celery.task(name="govoplan.campaigns.send_email", bind=True, max_retries=0) def send_email(self, job_id: str): """Send one explicitly queued campaign job. @@ -485,6 +509,35 @@ def dispatch_postbox_routes( return result +@celery.task( + name="govoplan.idm.expire_assignments", + bind=True, + max_retries=0, +) +def expire_idm_assignments( + self, + tenant_id: str | None = None, + limit: int = 100, +): + """Emit idempotent lifecycle events for elapsed IDM assignments.""" + + from govoplan_core.db.session import get_database + + with get_database().SessionLocal() as session: + provider = _idm_assignment_lifecycle() + if provider is None: + return {"selected": 0, "expired": 0, "assignment_ids": []} + result = dict( + provider.process_expired( + session, + tenant_id=tenant_id, + limit=limit, + ) + ) + session.commit() + return result + + @celery.task( name="govoplan.events.dispatch_outbox", bind=True, diff --git a/src/govoplan_core/core/idm.py b/src/govoplan_core/core/idm.py index 934136a..7cd545b 100644 --- a/src/govoplan_core/core/idm.py +++ b/src/govoplan_core/core/idm.py @@ -9,6 +9,7 @@ from typing import Literal, Protocol, runtime_checkable IDM_MODULE_ID = "idm" CAPABILITY_IDM_DIRECTORY = "idm.directory" CAPABILITY_IDM_FUNCTION_ASSIGNMENTS = "idm.function_assignments" +CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE = "idm.assignmentLifecycle" IdmStatus = Literal["active", "inactive", "suspended"] OrganizationFunctionAssignmentSource = Literal["direct", "delegated", "acting_for", "directory", "governance", "system"] @@ -106,3 +107,36 @@ class IdmFunctionAssignmentDirectory(Protocol): effective_at: datetime | None = None, ) -> Mapping[str, OrganizationFunctionIncumbencyRef]: ... + + +@runtime_checkable +class IdmAssignmentLifecycle(Protocol): + """Worker boundary for time-driven function-assignment transitions.""" + + def process_expired( + self, + session: object, + *, + tenant_id: str | None = None, + effective_at: datetime | None = None, + limit: int = 100, + ) -> Mapping[str, object]: + ... + + +def idm_assignment_lifecycle( + registry: object | None, +) -> IdmAssignmentLifecycle | None: + if ( + registry is None + or not hasattr(registry, "has_capability") + or not hasattr(registry, "capability") + or not registry.has_capability(CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE) + ): + return None + capability = registry.capability(CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE) + return ( + capability + if isinstance(capability, IdmAssignmentLifecycle) + else None + ) diff --git a/tests/test_idm_assignment_lifecycle_worker.py b/tests/test_idm_assignment_lifecycle_worker.py new file mode 100644 index 0000000..9e49f43 --- /dev/null +++ b/tests/test_idm_assignment_lifecycle_worker.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock, patch + +from govoplan_core.celery_app import celery, expire_idm_assignments +from govoplan_core.core.idm import ( + CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE, + IdmAssignmentLifecycle, + idm_assignment_lifecycle, +) +from govoplan_core.core.modules import ModuleContext, ModuleManifest +from govoplan_core.core.registry import PlatformRegistry + + +class _Lifecycle: + def process_expired( + self, + session, + *, + tenant_id=None, + effective_at=None, + limit=100, + ): + del session, tenant_id, effective_at, limit + return {"selected": 0, "expired": 0, "assignment_ids": []} + + +class IdmAssignmentLifecycleWorkerTests(unittest.TestCase): + def test_contract_is_runtime_checkable_and_optional(self) -> None: + provider = _Lifecycle() + self.assertIsInstance(provider, IdmAssignmentLifecycle) + registry = PlatformRegistry() + registry.register( + ModuleManifest( + id="idm_lifecycle_test", + name="IDM lifecycle test", + version="test", + capability_factories={ + CAPABILITY_IDM_ASSIGNMENT_LIFECYCLE: ( + lambda _context: provider + ) + }, + ) + ) + registry.configure_capability_context( + ModuleContext(registry=registry, settings=object()) + ) + + self.assertIs(provider, idm_assignment_lifecycle(registry)) + self.assertIsNone(idm_assignment_lifecycle(PlatformRegistry())) + + def test_worker_commits_provider_outcome(self) -> None: + session = MagicMock() + database = MagicMock() + database.SessionLocal.return_value.__enter__.return_value = session + provider = MagicMock() + provider.process_expired.return_value = { + "selected": 2, + "expired": 1, + "assignment_ids": ["assignment-1"], + } + + with ( + patch( + "govoplan_core.celery_app._idm_assignment_lifecycle", + return_value=provider, + ), + patch( + "govoplan_core.db.session.get_database", + return_value=database, + ), + ): + result = expire_idm_assignments.run("tenant-1", 25) + + provider.process_expired.assert_called_once_with( + session, + tenant_id="tenant-1", + limit=25, + ) + session.commit.assert_called_once_with() + self.assertEqual(1, result["expired"]) + + def test_route_and_periodic_job_are_registered(self) -> None: + self.assertEqual( + {"queue": "idm"}, + celery.conf.task_routes["govoplan.idm.expire_assignments"], + ) + schedule = celery.conf.beat_schedule[ + "idm-assignment-expiry-every-minute" + ] + self.assertEqual("govoplan.idm.expire_assignments", schedule["task"]) + self.assertEqual(60.0, schedule["schedule"]) + + +if __name__ == "__main__": + unittest.main()