Gate calendar sync workers by tenant entitlement

tests/test_tasks.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+import unittest
+from contextlib import contextmanager
+from types import SimpleNamespace
+from unittest.mock import patch
+
+from govoplan_calendar.backend.tasks import sync_due_caldav_sources_task
+from govoplan_core.core.module_entitlements import TenantModuleAdmission
+
+
+class CalendarTaskEntitlementTests(unittest.TestCase):
+    def test_disabled_tenant_preserves_due_sync_for_operator(self) -> None:
+        session = SimpleNamespace(commit=lambda: None)
+
+        @contextmanager
+        def session_scope():
+            yield session
+
+        admission = TenantModuleAdmission(
+            tenant_id="tenant-1",
+            module_id="calendar",
+            revision=2,
+            work_state="accepted",
+            allowed=False,
+            disposition="operator_action_required",
+            reason="Calendar is unavailable.",
+        )
+        resolver = SimpleNamespace(admission=lambda *_args, **_kwargs: admission)
+        registry = SimpleNamespace(
+            tenant_entitlement_resolver=lambda: resolver,
+        )
+        database = SimpleNamespace(SessionLocal=session_scope)
+
+        with (
+            patch(
+                "govoplan_core.core.worker_runtime.build_worker_platform_registry",
+                return_value=registry,
+            ),
+            patch(
+                "govoplan_core.db.session.get_database",
+                return_value=database,
+            ),
+            patch(
+                "govoplan_calendar.backend.service.sync_due_sources"
+            ) as sync_due,
+        ):
+            result = sync_due_caldav_sources_task.run("tenant-1", 10)
+
+        sync_due.assert_not_called()
+        self.assertEqual("operator_action_required", result[0]["status"])
+
+
+if __name__ == "__main__":
+    unittest.main()