49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from govoplan_core.celery_app import celery, dispatch_calendar_outbox
|
|
|
|
|
|
class CalendarOutboxWorkerTests(unittest.TestCase):
|
|
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.return_value = {
|
|
"processed": 1,
|
|
"succeeded": 1,
|
|
"retrying": 0,
|
|
"failed": 0,
|
|
"operations": [{"id": "operation-1", "status": "succeeded"}],
|
|
}
|
|
|
|
with (
|
|
patch("govoplan_core.celery_app._calendar_outbox", return_value=provider),
|
|
patch("govoplan_core.db.session.get_database", return_value=database),
|
|
):
|
|
result = dispatch_calendar_outbox.run("tenant-1", 25)
|
|
|
|
provider.dispatch_due.assert_called_once_with(
|
|
session,
|
|
tenant_id="tenant-1",
|
|
limit=25,
|
|
)
|
|
session.commit.assert_called_once_with()
|
|
self.assertEqual(result["succeeded"], 1)
|
|
|
|
def test_calendar_worker_route_and_periodic_recovery_are_registered(self) -> None:
|
|
self.assertEqual(
|
|
celery.conf.task_routes["govoplan.calendar.dispatch_outbox"],
|
|
{"queue": "calendar"},
|
|
)
|
|
schedule = celery.conf.beat_schedule["calendar-outbox-every-minute"]
|
|
self.assertEqual(schedule["task"], "govoplan.calendar.dispatch_outbox")
|
|
self.assertEqual(schedule["schedule"], 60.0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|