94 lines
2.9 KiB
Python
94 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from govoplan_core.celery_app import celery, dispatch_postbox_routes
|
|
from govoplan_core.core.modules import ModuleContext, ModuleManifest
|
|
from govoplan_core.core.postbox import (
|
|
CAPABILITY_POSTBOX_ROUTING,
|
|
PostboxRoutingProvider,
|
|
postbox_routing_provider,
|
|
)
|
|
from govoplan_core.core.registry import PlatformRegistry
|
|
|
|
|
|
class _RoutingProvider:
|
|
def dispatch_due_routes(
|
|
self,
|
|
session,
|
|
*,
|
|
tenant_id=None,
|
|
limit=50,
|
|
):
|
|
del session, tenant_id, limit
|
|
return {"selected": 0}
|
|
|
|
|
|
class PostboxRoutingWorkerTests(unittest.TestCase):
|
|
def test_contract_is_runtime_checkable_and_resolved(self) -> None:
|
|
provider = _RoutingProvider()
|
|
self.assertIsInstance(provider, PostboxRoutingProvider)
|
|
registry = PlatformRegistry()
|
|
registry.register(
|
|
ModuleManifest(
|
|
id="postbox_routing_test",
|
|
name="Postbox routing test",
|
|
version="test",
|
|
capability_factories={
|
|
CAPABILITY_POSTBOX_ROUTING: (
|
|
lambda _context: provider
|
|
)
|
|
},
|
|
)
|
|
)
|
|
registry.configure_capability_context(
|
|
ModuleContext(registry=registry, settings=object())
|
|
)
|
|
|
|
self.assertIs(provider, postbox_routing_provider(registry))
|
|
self.assertIsNone(postbox_routing_provider(PlatformRegistry()))
|
|
|
|
def test_worker_commits_provider_outcome(self) -> None:
|
|
session = MagicMock()
|
|
database = MagicMock()
|
|
database.SessionLocal.return_value.__enter__.return_value = session
|
|
provider = MagicMock()
|
|
provider.dispatch_due_routes.return_value = {
|
|
"selected": 1,
|
|
"delivered": 1,
|
|
}
|
|
|
|
with (
|
|
patch(
|
|
"govoplan_core.celery_app._postbox_routing_provider",
|
|
return_value=provider,
|
|
),
|
|
patch(
|
|
"govoplan_core.db.session.get_database",
|
|
return_value=database,
|
|
),
|
|
):
|
|
result = dispatch_postbox_routes.run("tenant-1", 25)
|
|
|
|
provider.dispatch_due_routes.assert_called_once_with(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
limit=25,
|
|
)
|
|
session.commit.assert_called_once_with()
|
|
self.assertEqual(1, result["delivered"])
|
|
|
|
def test_route_and_periodic_recovery_are_registered(self) -> None:
|
|
self.assertEqual(
|
|
{"queue": "postbox"},
|
|
celery.conf.task_routes["govoplan.postbox.dispatch_routes"],
|
|
)
|
|
schedule = celery.conf.beat_schedule["postbox-routes-every-minute"]
|
|
self.assertEqual("govoplan.postbox.dispatch_routes", schedule["task"])
|
|
self.assertEqual(60.0, schedule["schedule"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|