[DOP-20962] Add scheduler integration test

Author: Ilyas Gasanov

syncmaster/worker/config.py

@@ -15,6 +15,5 @@
     task_cls=WorkerTask,
     imports=[
         "syncmaster.worker.transfer",
-        "tests.test_integration.test_scheduler.test_task",
     ],
 )

syncmaster/worker/transfer.py

@@ -1,6 +1,7 @@
 # SPDX-FileCopyrightText: 2023-2024 MTS PJSC
 # SPDX-License-Identifier: Apache-2.0
 import logging
+import time
 from datetime import datetime, timezone
 
 import onetl
@@ -103,3 +104,23 @@ def load_correlation_id(task, *args, **kwargs) -> None:
     # Here we're able to load the correlation ID from the headers
     id_value = task.request.get(CORRELATION_CELERY_HEADER_ID)
     correlation_id.set(id_value)
+
+
+@celery.task(name="tick", bind=True, track_started=True)
+def tick(self: WorkerTask, run_id: int) -> None:
+    """This task can be used for testing."""
+    with Session(self.engine) as session:
+        run = session.get(Run, run_id)
+        if run is None:
+            raise RunNotFoundError
+
+        run.started_at = datetime.now(tz=timezone.utc)
+        run.status = Status.STARTED
+        session.add(run)
+        session.commit()
+
+        time.sleep(2)  # to make sure that previous status is handled in test
+        run.status = Status.FINISHED
+        run.ended_at = datetime.now(tz=timezone.utc)
+        session.add(run)
+        session.commit()

tests/test_integration/test_scheduler/test_scheduler.py

@@ -23,7 +23,7 @@ async def test_scheduler(
     transfer_fetcher = TransferFetcher(settings)
     transfers = await transfer_fetcher.fetch_updated_jobs()
     assert transfers
-    assert group_transfer.transfer.id in [t.id for t in transfers]
+    assert group_transfer.transfer.id in {t.id for t in transfers}
 
     transfer_job_manager.update_jobs(transfers)