feat: expand platform APIs, sources, and test coverage
Some checks failed
CI/CD Pipeline / Run Tests (pull_request) Successful in 1m53s
CI/CD Pipeline / Telegram Notify Success (push) Has been cancelled
CI/CD Pipeline / Run Tests (push) Has been cancelled
CI/CD Pipeline / Code Quality Checks (push) Has been cancelled
CI/CD Pipeline / Code Quality Checks (pull_request) Failing after 2m54s
CI/CD Pipeline / Telegram Notify Success (pull_request) Has been skipped
Some checks failed
CI/CD Pipeline / Run Tests (pull_request) Successful in 1m53s
CI/CD Pipeline / Telegram Notify Success (push) Has been cancelled
CI/CD Pipeline / Run Tests (push) Has been cancelled
CI/CD Pipeline / Code Quality Checks (push) Has been cancelled
CI/CD Pipeline / Code Quality Checks (pull_request) Failing after 2m54s
CI/CD Pipeline / Telegram Notify Success (pull_request) Has been skipped
This commit is contained in:
61
tests/apps/core/test_celery_module.py
Normal file
61
tests/apps/core/test_celery_module.py
Normal file
@@ -0,0 +1,61 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from django.test import SimpleTestCase
|
||||
|
||||
|
||||
CELERY_MODULE_PATH = (
|
||||
Path(__file__).resolve().parents[3] / "src" / "core" / "celery.py"
|
||||
)
|
||||
|
||||
|
||||
def _load_module(module_name: str):
|
||||
spec = importlib.util.spec_from_file_location(module_name, CELERY_MODULE_PATH)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
assert spec.loader is not None
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
class CeleryModuleTest(SimpleTestCase):
|
||||
def test_import_requires_django_settings_module(self):
|
||||
with patch.dict(os.environ, {}, clear=True):
|
||||
with self.assertRaisesMessage(
|
||||
RuntimeError,
|
||||
"DJANGO_SETTINGS_MODULE is not set.",
|
||||
):
|
||||
_load_module("isolated_core_celery_missing")
|
||||
|
||||
def test_import_runs_startup_checks_for_worker_runtime(self):
|
||||
app_mock = MagicMock()
|
||||
app_mock.conf = SimpleNamespace()
|
||||
|
||||
with patch.dict(os.environ, {"DJANGO_SETTINGS_MODULE": "settings.test"}, clear=True):
|
||||
with patch.object(sys, "argv", ["celery", "-A", "project", "worker"]):
|
||||
with patch("apps.core.startup_checks.run_startup_checks") as checks_mock:
|
||||
with patch("celery.Celery", return_value=app_mock):
|
||||
module = _load_module("isolated_core_celery_worker")
|
||||
|
||||
checks_mock.assert_called_once_with(component="celery")
|
||||
app_mock.config_from_object.assert_called_once_with(
|
||||
"django.conf:settings",
|
||||
namespace="CELERY",
|
||||
)
|
||||
app_mock.autodiscover_tasks.assert_called_once_with()
|
||||
self.assertEqual(module.app, app_mock)
|
||||
|
||||
def test_debug_task_prints_request(self):
|
||||
with patch.dict(os.environ, {"DJANGO_SETTINGS_MODULE": "settings.test"}, clear=True):
|
||||
with patch.object(sys, "argv", ["python", "manage.py", "shell"]):
|
||||
module = _load_module("isolated_core_celery_debug")
|
||||
|
||||
with patch("builtins.print") as print_mock:
|
||||
module.debug_task.run()
|
||||
|
||||
print_mock.assert_called_once()
|
||||
Reference in New Issue
Block a user