[DOP-20962] Add scheduler integration test (#118)

Author: IlyasDevelopment

.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 "worker and scheduler_integration"
+
+      - 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

docker-compose.test.yml

@@ -78,6 +78,7 @@ services:
       context: .
       target: test
     command: --loglevel=info -Q test_queue
+    entrypoint: [python, -m, celery, -A, tests.test_integration.celery_test, worker, --max-tasks-per-child=1]
     env_file: .env.docker
     volumes:
       - ./syncmaster:/app/syncmaster
@@ -90,7 +91,7 @@ services:
         condition: service_healthy
       rabbitmq:
         condition: service_healthy
-    profiles: [worker, s3, oracle, hdfs, hive, all]
+    profiles: [worker, scheduler, s3, oracle, hdfs, hive, all]
 
   test-postgres:
     image: postgres

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/settings/__init__.py

@@ -53,8 +53,6 @@ class Settings(BaseSettings):
     SCHEDULER_TRANSFER_FETCHING_TIMEOUT: int = 180  # seconds
     SCHEDULER_MISFIRE_GRACE_TIME: int = 300  # seconds
 
-    CORRELATION_CELERY_HEADER_ID: str = "CORRELATION_CELERY_HEADER_ID"
-
     TOKEN_EXPIRED_TIME: int = 60 * 60 * 10  # 10 hours
     CREATE_SPARK_SESSION_FUNCTION: ImportString = "syncmaster.worker.spark.get_worker_spark_session"
 

tests/conftest.py

@@ -32,6 +32,7 @@
     "tests.test_unit.test_runs.run_fixtures",
     "tests.test_unit.test_connections.connection_fixtures",
     "tests.test_unit.test_scheduler.scheduler_fixtures",
+    "tests.test_integration.test_scheduler.scheduler_fixtures",
 ]
 
 

tests/test_integration/celery_test.py

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

tests/test_integration/test_scheduler/__init__.py

@@ -0,0 +1,7 @@
+from tests.test_integration.test_scheduler.scheduler_fixtures.mocker_fixtures import (
+    mock_add_job,
+    mock_send_task_to_tick,
+)
+from tests.test_integration.test_scheduler.scheduler_fixtures.transfer_fixture import (
+    group_transfer_integration_mock,
+)

tests/test_integration/test_scheduler/scheduler_fixtures/__init__.py

@@ -0,0 +1,33 @@
+import asyncio
+
+import pytest
+from apscheduler.triggers.cron import CronTrigger
+from pytest_mock import MockerFixture, MockType
+
+from syncmaster.scheduler.transfer_job_manager import TransferJobManager
+from syncmaster.worker.config import celery
+
+
+@pytest.fixture
+def mock_send_task_to_tick(mocker: MockerFixture) -> MockType:
+    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) -> MockType:
+    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,
+        ),
+    )