56 lines
1.7 KiB
Python
56 lines
1.7 KiB
Python
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()
|