[DOP-20962] Add scheduler integration test

Author: Ilyas Gasanov

.github/workflows/scheduler-tests.yml

@@ -0,0 +1,79 @@
+name: Scheduler Tests
+on:
+  workflow_call:
+
+env:
+  DEFAULT_PYTHON: '3.12'
+
+jobs:
+  tests:
+    name: Run Scheduler tests
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v3
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v3
+
+      - name: Cache jars
+        uses: actions/cache@v4
+        with:
+          path: ./cached_jars
+          key: ${{ runner.os }}-python-${{ env.DEFAULT_PYTHON }}-test-scheduler
+          restore-keys: |
+            ${{ runner.os }}-python-${{ env.DEFAULT_PYTHON }}-test-scheduler
+            ${{ runner.os }}-python-
+
+      - name: Build Worker Image
+        uses: docker/build-push-action@v6
+        with:
+          context: .
+          tags: mtsrus/syncmaster-worker:${{ github.sha }}
+          target: test
+          file: docker/Dockerfile.worker
+          load: true
+          cache-from: mtsrus/syncmaster-worker:develop
+
+      - name: Docker compose up
+        run: |
+          docker compose -f docker-compose.test.yml --profile all down -v --remove-orphans
+          docker compose -f docker-compose.test.yml --profile worker up -d --wait --wait-timeout 200
+        env:
+          WORKER_IMAGE_TAG: ${{ github.sha }}
+
+      # This is important, as coverage is exported after receiving SIGTERM
+      - name: Run Scheduler Tests
+        run: |
+          docker compose -f ./docker-compose.test.yml --profile worker exec -T worker coverage run -m pytest -vvv -s -m "scheduler"
+
+      - name: Dump worker logs on failure
+        if: failure()
+        uses: jwalton/gh-docker-logs@v2
+        with:
+          images: mtsrus/syncmaster-worker
+          dest: ./logs
+
+      - name: Shutdown
+        if: always()
+        run: |
+          docker compose -f docker-compose.test.yml --profile all down -v --remove-orphans
+
+      - name: Upload worker logs
+        uses: actions/upload-artifact@v4
+        if: failure()
+        with:
+          name: worker-logs-scheduler
+          path: logs/*
+
+      - name: Upload coverage results
+        uses: actions/upload-artifact@v4
+        with:
+          name: coverage-scheduler
+          path: reports/*
+          # https://github.com/actions/upload-artifact/issues/602
+          include-hidden-files: true

.github/workflows/tests.yml

@@ -32,6 +32,10 @@ jobs:
     name: S3 tests
     uses: ./.github/workflows/s3-tests.yml
 
+  scheduler_tests:
+    name: Scheduler tests
+    uses: ./.github/workflows/scheduler-tests.yml
+
   unit_tests:
     name: Unit tests
     uses: ./.github/workflows/unit-test.yml

syncmaster/scheduler/__init__.py

@@ -1,2 +1,4 @@
 # SPDX-FileCopyrightText: 2023-2024 MTS PJSC
 # SPDX-License-Identifier: Apache-2.0
+from syncmaster.scheduler.transfer_fetcher import TransferFetcher
+from syncmaster.scheduler.transfer_job_manager import TransferJobManager

syncmaster/worker/config.py

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

tests/test_integration/test_scheduler/__init__.py

@@ -0,0 +1,33 @@
+import asyncio
+
+import pytest
+from apscheduler.triggers.cron import CronTrigger
+from pytest_mock import MockerFixture
+
+from syncmaster.scheduler.transfer_job_manager import TransferJobManager
+from syncmaster.worker.config import celery
+
+
+@pytest.fixture
+def mock_send_task_to_tick(mocker: MockerFixture):
+    original_to_thread = asyncio.to_thread
+    return mocker.patch(
+        "asyncio.to_thread",
+        new=lambda func, *args, **kwargs: original_to_thread(celery.send_task, "tick", *args[1:], **kwargs),
+    )
+
+
+@pytest.fixture
+def mock_add_job(mocker: MockerFixture, transfer_job_manager: TransferJobManager):
+    original_add_job = transfer_job_manager.scheduler.add_job
+    return mocker.patch.object(
+        transfer_job_manager.scheduler,
+        "add_job",
+        side_effect=lambda func, id, trigger, misfire_grace_time, args: original_add_job(
+            func=func,
+            id=id,
+            trigger=CronTrigger(second="*"),
+            misfire_grace_time=misfire_grace_time,
+            args=args,
+        ),
+    )

tests/test_integration/test_scheduler/fixtures.py

@@ -0,0 +1,45 @@
+import asyncio
+
+import pytest
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from syncmaster.db.models import Run, Status
+from syncmaster.scheduler import TransferFetcher, TransferJobManager
+from syncmaster.settings import Settings
+from tests.mocks import MockTransfer
+
+pytestmark = [pytest.mark.asyncio, pytest.mark.worker]
+
+
+async def test_scheduler(
+    session: AsyncSession,
+    settings: Settings,
+    group_transfer: MockTransfer,
+    transfer_job_manager: TransferJobManager,
+    mock_send_task_to_tick,
+    mock_add_job,
+):
+    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]
+
+    transfer_job_manager.update_jobs(transfers)
+
+    job = transfer_job_manager.scheduler.get_job(str(group_transfer.id))
+    assert job is not None
+
+    await asyncio.sleep(1)  # make sure that created job with every-second cron worked
+
+    run = await session.scalar(
+        select(Run).filter_by(transfer_id=group_transfer.id).order_by(Run.created_at.desc()),
+    )
+    assert run is not None
+    assert run.status in [Status.CREATED, Status.STARTED]
+
+    await asyncio.sleep(2)
+    await session.refresh(run)
+    completed_run = await session.scalar(select(Run).filter_by(id=run.id))
+    assert completed_run.status == Status.FINISHED
+    assert completed_run.ended_at is not None

tests/test_integration/test_scheduler/test_schedule.py

@@ -0,0 +1,28 @@
+import time
+from datetime import datetime, timezone
+
+from sqlalchemy.orm import Session
+
+from syncmaster.db.models.run import Run, Status
+from syncmaster.exceptions.run import RunNotFoundError
+from syncmaster.worker.base import WorkerTask
+from syncmaster.worker.config import celery
+
+
+@celery.task(name="tick", bind=True, track_started=True)
+def tick(self: WorkerTask, run_id: int) -> None:
+    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_task.py

@@ -1,4 +1,3 @@
-import secrets
 from collections.abc import AsyncGenerator
 
 import pytest_asyncio
@@ -46,7 +45,7 @@ async def group_transfer(
 
     queue = await create_queue(
         session=session,
-        name=f"{secrets.token_hex(5)}_test_queue",
+        name=f"test_queue",
         group_id=group.id,
     )
 

tests/test_unit/test_transfers/transfer_fixtures/transfer_fixture.py

@@ -154,7 +154,7 @@ async def create_transfer(
     source_params: dict | None = None,
     target_params: dict | None = None,
     is_scheduled: bool = True,
-    schedule: str = "0 0 * * *",
+    schedule: str = "* * * * *",
     strategy_params: dict | None = None,
     description: str = "",
 ) -> Transfer: