avm/mostovik-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(celery): schedule weekly updates and run startup refresh
Some checks failed
CI/CD Pipeline / Run Tests (push) Successful in 1m41s
Details
CI/CD Pipeline / Telegram Notify Success (push) Has been cancelled
Details
CI/CD Pipeline / Code Quality Checks (push) Has been cancelled
Details
CI/CD Pipeline / Code Quality Checks (pull_request) Successful in 1m31s
Details
CI/CD Pipeline / Telegram Notify Success (pull_request) Has been cancelled
Details
CI/CD Pipeline / Run Tests (pull_request) Has been cancelled
Details

Browse Source
This commit is contained in:
Aleksandr Meshchriakov
2026-03-20 11:14:03 +01:00
parent c8bdc282de
commit b8e3f0a5d3
4 changed files with 134 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
tests/apps/core/test_celery_module.py
View File

@@ -4,7 +4,7 @@ import importlib.util
import os
import sys
from pathlib import Path
from types import SimpleNamespace
from types import ModuleType, SimpleNamespace
from unittest.mock import MagicMock, patch
from django.test import SimpleTestCase
@@ -46,6 +46,59 @@ class CeleryModuleTest(SimpleTestCase):
)
app_mock.autodiscover_tasks.assert_called_once_with()
self.assertEqual(module.app, app_mock)
self.assertIn("parse-industrial-production-daily", module.app.conf.beat_schedule)
self.assertIn("parse-manufactures-daily", module.app.conf.beat_schedule)
self.assertIn("parse-industrial-products-daily", module.app.conf.beat_schedule)
self.assertIn("parse-inspections-weekly", module.app.conf.beat_schedule)
def test_startup_refresh_queues_when_lock_acquired(self):
with patch.dict(
os.environ, {"DJANGO_SETTINGS_MODULE": "settings.test"}, clear=True
), patch.object(sys, "argv", ["python", "manage.py", "shell"]):
module = _load_module("isolated_core_celery_startup_refresh")
fake_tasks_module = ModuleType("apps.parsers.tasks")
apply_async_mock = MagicMock(return_value=SimpleNamespace(id="task-123"))
fake_tasks_module.parse_all_sources = SimpleNamespace(
apply_async=apply_async_mock
)
with patch.object(module, "settings", SimpleNamespace(
CELERY_STARTUP_REFRESH_ENABLED=True,
CELERY_STARTUP_REFRESH_LOCK_KEY="startup-lock",
CELERY_STARTUP_REFRESH_LOCK_TTL_SECONDS=120,
CELERY_STARTUP_REFRESH_DELAY_SECONDS=45,
)), patch.object(module.cache, "add", return_value=True) as add_mock, patch.dict(
sys.modules, {"apps.parsers.tasks": fake_tasks_module}
):
module._queue_startup_sources_refresh()
add_mock.assert_called_once_with("startup-lock", "1", timeout=120)
apply_async_mock.assert_called_once_with(countdown=45)
def test_startup_refresh_skips_when_lock_exists(self):
with patch.dict(
os.environ, {"DJANGO_SETTINGS_MODULE": "settings.test"}, clear=True
), patch.object(sys, "argv", ["python", "manage.py", "shell"]):
module = _load_module("isolated_core_celery_startup_skip")
fake_tasks_module = ModuleType("apps.parsers.tasks")
apply_async_mock = MagicMock()
fake_tasks_module.parse_all_sources = SimpleNamespace(
apply_async=apply_async_mock
)
with patch.object(module, "settings", SimpleNamespace(
CELERY_STARTUP_REFRESH_ENABLED=True,
CELERY_STARTUP_REFRESH_LOCK_KEY="startup-lock",
CELERY_STARTUP_REFRESH_LOCK_TTL_SECONDS=120,
CELERY_STARTUP_REFRESH_DELAY_SECONDS=45,
)), patch.object(module.cache, "add", return_value=False), patch.dict(
sys.modules, {"apps.parsers.tasks": fake_tasks_module}
):
module._queue_startup_sources_refresh()
apply_async_mock.assert_not_called()
def test_debug_task_prints_request(self):
with patch.dict(