gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #420 from christianlouis/copilot/improve-code-coverage-tests

Browse Source 
fix(tests): correct mock patch targets in google_drive coverage tests
This commit is contained in:
Christian Krakau-Louis
2026-02-26 11:46:03 +01:00
committed by GitHub
parent 06f125d7e6 c6ac6b50e9
commit a64de0a6cb
5 changed files with 144 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/utils/config_validator.py
-30
View File
@@ -1,30 +0,0 @@
#!/usr/bin/env python3
"""
Configuration validation for the application.
This file serves as a backward-compatible interface to the config_validator package.
"""
from app.utils.config_validator.masking import mask_sensitive_value
from app.utils.config_validator.providers import get_provider_status
from app.utils.config_validator.settings_display import dump_all_settings, get_settings_for_display
# Import and re-export all functions from the new package
from app.utils.config_validator.validators import (
check_all_configs,
validate_auth_config,
validate_email_config,
validate_notification_config,
validate_storage_configs,
)
__all__ = [
"validate_email_config",
"validate_storage_configs",
"validate_notification_config",
"validate_auth_config",
"mask_sensitive_value",
"get_provider_status",
"get_settings_for_display",
"dump_all_settings",
"check_all_configs",
]
pyproject.toml
-8
View File
@@ -210,14 +210,6 @@ omit = [
"*/__pycache__/*",
"*/venv/*",
"*/env/*",
# This file is shadowed by the config_validator/ package directory and
# can never be imported via the normal Python import system. It is kept
# for historical reference only.
"app/utils/config_validator.py",
# celery_worker.py is an entry-point script for the Celery worker process;
# it initialises Celery beat schedules and cannot be meaningfully unit-tested
# without a live Redis + Celery environment.
"app/celery_worker.py",
]
[tool.coverage.report]
tests/test_api_google_drive_coverage2.py
+8 -10
View File
@@ -31,7 +31,7 @@ class TestUpdateSettingsExceptionHandler:
class TestTestTokenServiceAccount:
"""Cover lines 214-255: service account test-token paths."""
@patch("app.api.google_drive.get_google_drive_service")
@patch("app.tasks.upload_to_google_drive.get_google_drive_service")
def test_test_token_service_account_success_no_delegation(self, mock_get_service, client: TestClient):
"""Test successful service account connection without delegation."""
from app.config import settings
@@ -55,7 +55,7 @@ class TestTestTokenServiceAccount:
assert data["auth_type"] == "service_account"
assert "sa@project.iam.gserviceaccount.com" in data["message"]
@patch("app.api.google_drive.get_google_drive_service")
@patch("app.tasks.upload_to_google_drive.get_google_drive_service")
def test_test_token_service_account_with_delegation(self, mock_get_service, client: TestClient):
"""Test service account with delegation shows delegated user info."""
from app.config import settings
@@ -93,7 +93,7 @@ class TestTestTokenServiceAccount:
assert data["status"] == "error"
assert "not configured" in data["message"]
@patch("app.api.google_drive.get_google_drive_service")
@patch("app.tasks.upload_to_google_drive.get_google_drive_service")
def test_test_token_service_account_connection_error(self, mock_get_service, client: TestClient):
"""Test service account connection error (lines 245-251)."""
from app.config import settings
@@ -118,13 +118,11 @@ class TestGetTokenInfoOuterException:
def test_get_token_info_outer_exception(self, client: TestClient):
"""Trigger the outer exception handler in get_google_drive_token_info."""
from app.config import settings
with patch.object(
type(settings),
"google_drive_use_oauth",
property(fget=lambda self: (_ for _ in ()).throw(Exception("boom"))),
):
mock_settings = MagicMock()
# Property on the mock type so getattr() propagates a non-AttributeError,
# bypassing the default value and reaching the outer except block.
type(mock_settings).google_drive_use_oauth = property(lambda self: (_ for _ in ()).throw(Exception("boom")))
with patch("app.api.google_drive.settings", mock_settings):
response = client.get("/api/google-drive/get-token-info")
assert response.status_code == 200
tests/test_celery_worker.py
+134 -62
View File
@@ -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
tests/test_config_validator_reexport.py
+2 -2
View File
@@ -1,7 +1,7 @@
"""
Tests for app/utils/config_validator.py
Tests for app/utils/config_validator/ package
This module tests the config_validator re-export module.
This module tests the config_validator package re-exports from __init__.py.
"""
import pytest