test(coverage): add celery_worker coverage and remove dead config_validator.py
- Rewrite tests/test_celery_worker.py to mock check_credentials.apply_async at import time, enabling all 35 statements to be covered without Redis - Add tests for conditional beat schedule entries (IMAP, Uptime Kuma) - Delete app/utils/config_validator.py — dead code shadowed by the config_validator/ package directory (Python gives packages precedence) - Remove both files from coverage omit in pyproject.toml - celery_worker.py now at 100% coverage (was 0%) Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
+134
-62
@@ -1,96 +1,168 @@
|
||||
"""
|
||||
Tests for app/celery_worker.py
|
||||
|
||||
This module tests the Celery worker configuration, task imports, and beat schedule.
|
||||
Targets all statements and branches in the Celery worker configuration module,
|
||||
including module-level task imports, beat schedule construction, the conditional
|
||||
IMAP / Uptime Kuma entries, and the None-entry filter.
|
||||
|
||||
Because celery_worker.py fires ``check_credentials.apply_async()`` at import
|
||||
time (which requires a live Redis broker), every test mocks that call via
|
||||
``importlib.reload`` so the module can be exercised without external services.
|
||||
"""
|
||||
|
||||
import importlib
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from app.config import settings
|
||||
|
||||
|
||||
def _reload_celery_worker():
|
||||
"""Reload celery_worker with ``apply_async`` mocked to avoid Redis."""
|
||||
with patch("app.tasks.check_credentials.check_credentials.apply_async"):
|
||||
import app.celery_worker
|
||||
|
||||
importlib.reload(app.celery_worker)
|
||||
return app.celery_worker
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestCeleryWorkerConfig:
|
||||
"""Test Celery worker configuration."""
|
||||
"""Test Celery worker configuration (imports, task routes, side effects)."""
|
||||
|
||||
def test_test_task_function(self):
|
||||
"""Test the test_task function returns expected value."""
|
||||
from app.celery_worker import test_task
|
||||
|
||||
result = test_task()
|
||||
mod = _reload_celery_worker()
|
||||
result = mod.test_task()
|
||||
assert result == "Celery is working!"
|
||||
|
||||
def test_celery_instance_exists(self):
|
||||
"""Test that celery instance exists in module."""
|
||||
from app import celery_worker
|
||||
mod = _reload_celery_worker()
|
||||
assert hasattr(mod, "celery")
|
||||
assert mod.celery is not None
|
||||
|
||||
assert hasattr(celery_worker, "celery")
|
||||
assert celery_worker.celery is not None
|
||||
|
||||
def test_task_routes_exists(self):
|
||||
"""Test that task routes configuration exists."""
|
||||
from app import celery_worker
|
||||
|
||||
# Task routes should be configured
|
||||
assert hasattr(celery_worker.celery.conf, "task_routes")
|
||||
def test_task_routes_configured(self):
|
||||
"""Test that task routes are set."""
|
||||
mod = _reload_celery_worker()
|
||||
routes = mod.celery.conf.task_routes
|
||||
assert routes is not None
|
||||
assert "app.tasks.*" in routes
|
||||
|
||||
def test_all_task_imports_successful(self):
|
||||
"""Test that all task modules are imported successfully."""
|
||||
# Just import the module to verify no import errors
|
||||
from app import celery_worker
|
||||
"""Test that all task modules are imported without errors."""
|
||||
mod = _reload_celery_worker()
|
||||
assert mod is not None
|
||||
|
||||
# Module imported successfully
|
||||
assert celery_worker is not None
|
||||
def test_register_settings_reload_signal_called(self):
|
||||
"""Test that register_settings_reload_signal is called at module load."""
|
||||
with (
|
||||
patch("app.tasks.check_credentials.check_credentials.apply_async"),
|
||||
patch("app.utils.settings_sync.register_settings_reload_signal") as mock_reg,
|
||||
):
|
||||
import app.celery_worker
|
||||
|
||||
importlib.reload(app.celery_worker)
|
||||
mock_reg.assert_called_once()
|
||||
|
||||
def test_check_credentials_apply_async_called(self):
|
||||
"""Test that check_credentials.apply_async is called at module load."""
|
||||
with patch("app.tasks.check_credentials.check_credentials.apply_async") as mock_apply:
|
||||
import app.celery_worker
|
||||
|
||||
importlib.reload(app.celery_worker)
|
||||
mock_apply.assert_called_once_with(countdown=10)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
class TestBeatScheduleConfiguration:
|
||||
"""Test Celery beat schedule configuration."""
|
||||
"""Test Celery beat schedule configuration (lines 56-97)."""
|
||||
|
||||
def test_beat_schedule_structure(self):
|
||||
"""Test that beat schedule has expected structure."""
|
||||
from app.celery_worker import celery
|
||||
def test_beat_schedule_always_present_entries(self):
|
||||
"""Test that the three unconditional entries are always present."""
|
||||
mod = _reload_celery_worker()
|
||||
schedule = mod.celery.conf.beat_schedule
|
||||
assert isinstance(schedule, dict)
|
||||
assert "check-credentials-regularly" in schedule
|
||||
assert "check-credentials-daily" in schedule
|
||||
assert "monitor-stalled-steps" in schedule
|
||||
|
||||
# Beat schedule should be a dictionary
|
||||
assert isinstance(celery.conf.beat_schedule, dict)
|
||||
def test_credential_check_regular_schedule(self):
|
||||
"""Test 'check-credentials-regularly' entry."""
|
||||
mod = _reload_celery_worker()
|
||||
entry = mod.celery.conf.beat_schedule["check-credentials-regularly"]
|
||||
assert entry["task"] == "app.tasks.check_credentials.check_credentials"
|
||||
assert "schedule" in entry
|
||||
assert entry["options"]["expires"] == 240
|
||||
|
||||
# Should include credential check tasks
|
||||
assert "check-credentials-regularly" in celery.conf.beat_schedule
|
||||
assert "check-credentials-daily" in celery.conf.beat_schedule
|
||||
assert "monitor-stalled-steps" in celery.conf.beat_schedule
|
||||
|
||||
def test_credential_check_schedule(self):
|
||||
"""Test credential check schedule configuration."""
|
||||
from app.celery_worker import celery
|
||||
|
||||
schedule = celery.conf.beat_schedule.get("check-credentials-regularly")
|
||||
assert schedule is not None
|
||||
assert schedule["task"] == "app.tasks.check_credentials.check_credentials"
|
||||
assert "schedule" in schedule
|
||||
assert schedule["options"]["expires"] == 240
|
||||
|
||||
def test_daily_credential_check_schedule(self):
|
||||
"""Test daily credential check schedule."""
|
||||
from app.celery_worker import celery
|
||||
|
||||
schedule = celery.conf.beat_schedule.get("check-credentials-daily")
|
||||
assert schedule is not None
|
||||
assert schedule["task"] == "app.tasks.check_credentials.check_credentials"
|
||||
assert "schedule" in schedule
|
||||
assert schedule["options"]["expires"] == 3600
|
||||
def test_credential_check_daily_schedule(self):
|
||||
"""Test 'check-credentials-daily' entry."""
|
||||
mod = _reload_celery_worker()
|
||||
entry = mod.celery.conf.beat_schedule["check-credentials-daily"]
|
||||
assert entry["task"] == "app.tasks.check_credentials.check_credentials"
|
||||
assert entry["options"]["expires"] == 3600
|
||||
|
||||
def test_monitor_stalled_steps_schedule(self):
|
||||
"""Test monitor stalled steps schedule."""
|
||||
from app.celery_worker import celery
|
||||
|
||||
schedule = celery.conf.beat_schedule.get("monitor-stalled-steps")
|
||||
assert schedule is not None
|
||||
assert schedule["task"] == "app.tasks.monitor_stalled_steps.monitor_stalled_steps"
|
||||
assert "schedule" in schedule
|
||||
assert schedule["options"]["expires"] == 55
|
||||
"""Test 'monitor-stalled-steps' entry."""
|
||||
mod = _reload_celery_worker()
|
||||
entry = mod.celery.conf.beat_schedule["monitor-stalled-steps"]
|
||||
assert entry["task"] == "app.tasks.monitor_stalled_steps.monitor_stalled_steps"
|
||||
assert entry["options"]["expires"] == 55
|
||||
|
||||
def test_no_none_entries_in_beat_schedule(self):
|
||||
"""Test that None entries are filtered from beat schedule."""
|
||||
from app.celery_worker import celery
|
||||
|
||||
# No None values in beat schedule
|
||||
for key, value in celery.conf.beat_schedule.items():
|
||||
mod = _reload_celery_worker()
|
||||
for key, value in mod.celery.conf.beat_schedule.items():
|
||||
assert value is not None, f"Beat schedule entry '{key}' should not be None"
|
||||
|
||||
# -- Conditional schedule entries ----------------------------------------
|
||||
|
||||
def test_imap_schedule_absent_when_not_configured(self):
|
||||
"""Test IMAP polling absent when neither imap host is set."""
|
||||
mod = _reload_celery_worker()
|
||||
assert "poll-inboxes-every-minute" not in mod.celery.conf.beat_schedule
|
||||
|
||||
def test_uptime_kuma_schedule_absent_when_not_configured(self):
|
||||
"""Test Uptime Kuma ping absent when url is not set."""
|
||||
mod = _reload_celery_worker()
|
||||
assert "ping-uptime-kuma" not in mod.celery.conf.beat_schedule
|
||||
|
||||
def test_imap_schedule_present_when_imap1_configured(self):
|
||||
"""Test IMAP polling present when imap1_host is set."""
|
||||
original = settings.imap1_host
|
||||
try:
|
||||
settings.imap1_host = "imap.example.com"
|
||||
mod = _reload_celery_worker()
|
||||
assert "poll-inboxes-every-minute" in mod.celery.conf.beat_schedule
|
||||
entry = mod.celery.conf.beat_schedule["poll-inboxes-every-minute"]
|
||||
assert entry["task"] == "app.tasks.imap_tasks.pull_all_inboxes"
|
||||
assert entry["options"]["expires"] == 55
|
||||
finally:
|
||||
settings.imap1_host = original
|
||||
|
||||
def test_imap_schedule_present_when_imap2_configured(self):
|
||||
"""Test IMAP polling present when imap2_host is set."""
|
||||
original = settings.imap2_host
|
||||
try:
|
||||
settings.imap2_host = "imap2.example.com"
|
||||
mod = _reload_celery_worker()
|
||||
assert "poll-inboxes-every-minute" in mod.celery.conf.beat_schedule
|
||||
finally:
|
||||
settings.imap2_host = original
|
||||
|
||||
def test_uptime_kuma_schedule_present_when_configured(self):
|
||||
"""Test Uptime Kuma ping present when uptime_kuma_url is set."""
|
||||
orig_url = settings.uptime_kuma_url
|
||||
orig_interval = settings.uptime_kuma_ping_interval
|
||||
try:
|
||||
settings.uptime_kuma_url = "https://uptime.example.com/api/push/abc"
|
||||
settings.uptime_kuma_ping_interval = 3
|
||||
mod = _reload_celery_worker()
|
||||
assert "ping-uptime-kuma" in mod.celery.conf.beat_schedule
|
||||
entry = mod.celery.conf.beat_schedule["ping-uptime-kuma"]
|
||||
assert entry["task"] == "app.tasks.uptime_kuma_tasks.ping_uptime_kuma"
|
||||
assert entry["options"]["expires"] == 55
|
||||
finally:
|
||||
settings.uptime_kuma_url = orig_url
|
||||
settings.uptime_kuma_ping_interval = orig_interval
|
||||
|
||||
Reference in New Issue
Block a user