[DOP-21799] Refactor celery initialization

Author: Ilyas Gasanov

docs/conf.py

@@ -147,11 +147,6 @@
 # If true, `todo` and `todoList` produce output, else they produce nothing.
 todo_include_todos = False
 
-# set the environment variable before imports
-# TODO: remove after global init of WorkerAppSettings in worker/__init__.py
-os.environ["SYNCMASTER__ENCRYPTION__CRYPTO_KEY"] = "crypto_key"
-os.environ["SYNCMASTER__DATABASE__URL"] = "postgresql+asyncpg://syncmaster:changeme@db:5432/syncmaster"
-os.environ["SYNCMASTER__BROKER__URL"] = "amqp://guest:guest@localhost:5672/"
 
 # -- Options for HTMLHelp output ------------------------------------------
 

syncmaster/scheduler/celery.py

@@ -3,4 +3,4 @@
 from syncmaster.scheduler.settings import SchedulerAppSettings
 from syncmaster.worker import celery_factory
 
-celery = celery_factory(SchedulerAppSettings())
+app = celery_factory(SchedulerAppSettings())

syncmaster/scheduler/transfer_job_manager.py

@@ -9,7 +9,7 @@
 from syncmaster.backend.services.unit_of_work import UnitOfWork
 from syncmaster.db.models import RunType, Status, Transfer
 from syncmaster.exceptions.run import CannotConnectToTaskQueueError
-from syncmaster.scheduler.celery import celery
+from syncmaster.scheduler.celery import app as celery
 from syncmaster.scheduler.settings import SchedulerAppSettings as Settings
 from syncmaster.scheduler.utils import get_async_session
 from syncmaster.schemas.v1.connections.connection import ReadAuthDataSchema

syncmaster/worker/celery.py

@@ -3,4 +3,4 @@
 from syncmaster.worker import celery_factory
 from syncmaster.worker.settings import WorkerAppSettings
 
-celery = celery_factory(WorkerAppSettings())
+app = celery_factory(WorkerAppSettings())

syncmaster/worker/transfer.py

@@ -16,7 +16,7 @@
 from syncmaster.db.models import AuthData, Run, Status, Transfer
 from syncmaster.db.repositories.utils import decrypt_auth_data
 from syncmaster.exceptions.run import RunNotFoundError
-from syncmaster.worker.celery import celery
+from syncmaster.worker.celery import app as celery
 from syncmaster.worker.controller import TransferController
 from syncmaster.worker.settings import WorkerAppSettings
 

tests/test_integration/celery_test.py

@@ -1,3 +1,3 @@
-from syncmaster.scheduler.celery import celery
+from syncmaster.scheduler.celery import app as celery
 
 celery.conf.update(imports=list(celery.conf.imports) + ["tests.test_integration.test_scheduler.test_task"])

tests/test_integration/test_scheduler/test_task.py

@@ -5,7 +5,7 @@
 
 from syncmaster.db.models.run import Run, Status
 from syncmaster.exceptions.run import RunNotFoundError
-from syncmaster.scheduler.celery import celery
+from syncmaster.scheduler.celery import app as celery
 from syncmaster.worker.base import WorkerTask
 
 

tests/test_unit/test_runs/test_create_run.py

@@ -20,7 +20,7 @@ async def test_developer_plus_can_create_run_of_transfer_his_group(
 ) -> None:
     # Arrange
     user = group_transfer.owner_group.get_member_of_role(role_developer_plus)
-    mock_send_task = mocker.patch("syncmaster.worker.celery.send_task")
+    mock_send_task = mocker.patch("syncmaster.worker.celery.app.send_task")
     mock_to_thread = mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     run = (
@@ -73,7 +73,7 @@ async def test_groupless_user_cannot_create_run(
     mocker,
 ) -> None:
     # Arrange
-    mocker.patch("syncmaster.worker.celery.send_task")
+    mocker.patch("syncmaster.worker.celery.app.send_task")
     mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     # Act
@@ -103,7 +103,7 @@ async def test_group_member_cannot_create_run_of_other_group_transfer(
     role_guest_plus: UserTestRoles,
 ):
     # Arrange
-    mocker.patch("syncmaster.worker.celery.send_task")
+    mocker.patch("syncmaster.worker.celery.app.send_task")
     mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
     user = group.get_member_of_role(role_guest_plus)
 
@@ -139,7 +139,7 @@ async def test_superuser_can_create_run(
     mocker,
 ) -> None:
     # Arrange
-    mock_send_task = mocker.patch("syncmaster.worker.celery.send_task")
+    mock_send_task = mocker.patch("syncmaster.worker.celery.app.send_task")
     mock_to_thread = mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     # Act
@@ -183,7 +183,7 @@ async def test_unauthorized_user_cannot_create_run(
     mocker,
 ) -> None:
     # Arrange
-    mocker.patch("syncmaster.worker.celery.send_task")
+    mocker.patch("syncmaster.worker.celery.app.send_task")
     mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     # Act
@@ -212,7 +212,7 @@ async def test_group_member_cannot_create_run_of_unknown_transfer_error(
 ) -> None:
     # Arrange
     user = group_transfer.owner_group.get_member_of_role(role_guest_plus)
-    mocker.patch("syncmaster.worker.celery.send_task")
+    mocker.patch("syncmaster.worker.celery.app.send_task")
     mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     # Act
@@ -240,7 +240,7 @@ async def test_superuser_cannot_create_run_of_unknown_transfer_error(
     mocker,
 ) -> None:
     # Arrange
-    mocker.patch("syncmaster.worker.celery.send_task")
+    mocker.patch("syncmaster.worker.celery.app.send_task")
     mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     # Act

tests/test_unit/test_scheduler/test_transfer_job_manager.py

@@ -80,7 +80,7 @@ async def test_send_job_to_celery_with_success(
     group_transfer: MockTransfer,
 ):
     # Arrange
-    mock_send_task = mocker.patch("syncmaster.worker.celery.send_task")
+    mock_send_task = mocker.patch("syncmaster.worker.celery.app.send_task")
     mock_to_thread = mocker.patch("asyncio.to_thread", new_callable=AsyncMock)
 
     # Act
@@ -107,7 +107,7 @@ async def test_send_job_to_celery_with_failure(
     group_transfer: MockTransfer,
 ):
     # Arrange
-    mocker.patch("syncmaster.worker.celery.send_task")
+    mocker.patch("syncmaster.worker.celery.app.send_task")
     mocker.patch("asyncio.to_thread", new_callable=AsyncMock, side_effect=KombuError)
 
     # Act & Assert