From a7f47b1bf0e395fef79ccc8d8f0fa4854cc0648f Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 13 Feb 2026 23:07:54 +0000 Subject: [PATCH] test: increase coverage for celery_app.py and google_drive.py to 100% Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- tests/test_celery_app.py | 226 +++++++++++++++++++++++++++++++ tests/test_views_google_drive.py | 104 ++++++++++++++ 2 files changed, 330 insertions(+) create mode 100644 tests/test_celery_app.py diff --git a/tests/test_celery_app.py b/tests/test_celery_app.py new file mode 100644 index 00000000..a6602624 --- /dev/null +++ b/tests/test_celery_app.py @@ -0,0 +1,226 @@ +""" +Tests for app/celery_app.py + +This module tests the Celery app configuration and task failure handler. +""" + +import logging +from unittest.mock import MagicMock, patch + +import pytest + + +@pytest.mark.unit +class TestCeleryAppConfig: + """Test Celery app configuration.""" + + def test_celery_instance_exists(self): + """Test that celery instance exists and is properly configured.""" + from app.celery_app import celery + + assert celery is not None + assert celery.main == "document_processor" + + def test_celery_broker_configured(self): + """Test that celery broker is configured.""" + from app.celery_app import celery + + assert celery.conf.broker_url is not None + assert celery.conf.result_backend is not None + + def test_celery_default_queue(self): + """Test that default queue is set to document_processor.""" + from app.celery_app import celery + + assert celery.conf.task_default_queue == "document_processor" + + def test_celery_task_routes(self): + """Test that task routes are configured.""" + from app.celery_app import celery + + assert celery.conf.task_routes is not None + assert "app.tasks.*" in celery.conf.task_routes + assert celery.conf.task_routes["app.tasks.*"]["queue"] == "document_processor" + + def test_broker_connection_retry_on_startup(self): + """Test that broker connection retry on startup is enabled.""" + from app.celery_app import celery + + assert celery.conf.broker_connection_retry_on_startup is True + + +@pytest.mark.unit +class TestTaskFailureHandler: + """Test task failure handler signal.""" + + @patch("app.celery_app.settings") + @patch("app.utils.notification.notify_celery_failure") + def test_task_failure_handler_sends_notification(self, mock_notify, mock_settings): + """Test that task failure handler sends notification when enabled.""" + # Configure settings to enable notifications + mock_settings.notify_on_task_failure = True + + # Import the handler + from app.celery_app import task_failure_handler + + # Create mock sender with task name + mock_sender = MagicMock() + mock_sender.name = "test.task" + + # Create exception instance + test_exception = ValueError("Test error") + + # Call the handler + task_failure_handler( + sender=mock_sender, + task_id="test-task-id", + exception=test_exception, + args=[1, 2, 3], + kwargs={"key": "value"}, + ) + + # Verify notification was sent with correct parameters + mock_notify.assert_called_once() + call_kwargs = mock_notify.call_args[1] + assert call_kwargs["task_name"] == "test.task" + assert call_kwargs["task_id"] == "test-task-id" + assert isinstance(call_kwargs["exc"], ValueError) + assert str(call_kwargs["exc"]) == "Test error" + assert call_kwargs["args"] == [1, 2, 3] + assert call_kwargs["kwargs"] == {"key": "value"} + + @patch("app.celery_app.settings") + def test_task_failure_handler_disabled_notification(self, mock_settings): + """Test that task failure handler does not send notification when disabled.""" + # Configure settings to disable notifications + mock_settings.notify_on_task_failure = False + + # Import the handler + from app.celery_app import task_failure_handler + + with patch("app.utils.notification.notify_celery_failure") as mock_notify: + # Create mock sender + mock_sender = MagicMock() + mock_sender.name = "test.task" + + # Call the handler + task_failure_handler( + sender=mock_sender, + task_id="test-task-id", + exception=ValueError("Test error"), + ) + + # Verify notification was NOT sent + mock_notify.assert_not_called() + + @patch("app.celery_app.settings") + @patch("app.utils.notification.notify_celery_failure") + def test_task_failure_handler_with_no_sender(self, mock_notify, mock_settings): + """Test task failure handler when sender is None.""" + mock_settings.notify_on_task_failure = True + + from app.celery_app import task_failure_handler + + # Call with no sender + task_failure_handler( + sender=None, + task_id="test-task-id", + exception=ValueError("Test error"), + ) + + # Should use "Unknown" as task name + mock_notify.assert_called_once() + call_args = mock_notify.call_args[1] + assert call_args["task_name"] == "Unknown" + + @patch("app.celery_app.settings") + @patch("app.utils.notification.notify_celery_failure") + def test_task_failure_handler_with_no_task_id(self, mock_notify, mock_settings): + """Test task failure handler when task_id is None.""" + mock_settings.notify_on_task_failure = True + + from app.celery_app import task_failure_handler + + mock_sender = MagicMock() + mock_sender.name = "test.task" + + # Call with no task_id + task_failure_handler( + sender=mock_sender, + task_id=None, + exception=ValueError("Test error"), + ) + + # Should use "N/A" as task_id + mock_notify.assert_called_once() + call_args = mock_notify.call_args[1] + assert call_args["task_id"] == "N/A" + + @patch("app.celery_app.settings") + @patch("app.utils.notification.notify_celery_failure") + def test_task_failure_handler_with_empty_args_kwargs(self, mock_notify, mock_settings): + """Test task failure handler with no args or kwargs.""" + mock_settings.notify_on_task_failure = True + + from app.celery_app import task_failure_handler + + mock_sender = MagicMock() + mock_sender.name = "test.task" + + # Call with None args/kwargs + task_failure_handler( + sender=mock_sender, + task_id="test-task-id", + exception=ValueError("Test error"), + args=None, + kwargs=None, + ) + + # Should use empty list/dict as defaults + mock_notify.assert_called_once() + call_args = mock_notify.call_args[1] + assert call_args["args"] == [] + assert call_args["kwargs"] == {} + + @patch("app.celery_app.settings") + @patch("app.utils.notification.notify_celery_failure", side_effect=Exception("Notification failed")) + def test_task_failure_handler_exception_handling(self, mock_notify, mock_settings, caplog): + """Test that exceptions in notification are caught and logged.""" + mock_settings.notify_on_task_failure = True + + from app.celery_app import task_failure_handler + + mock_sender = MagicMock() + mock_sender.name = "test.task" + + # Call the handler - should not raise exception + with caplog.at_level(logging.ERROR): + task_failure_handler( + sender=mock_sender, + task_id="test-task-id", + exception=ValueError("Test error"), + ) + + # Verify the exception was logged + assert any("Failed to send task failure notification" in record.message for record in caplog.records) + + @patch("app.celery_app.settings") + @patch("app.utils.notification.notify_celery_failure") + def test_task_failure_handler_called_by_signal(self, mock_notify, mock_settings): + """Test that the handler is properly connected to the task_failure signal.""" + mock_settings.notify_on_task_failure = True + + # Import to ensure signal is connected + from app.celery_app import task_failure_handler + + # Import the signal + from celery.signals import task_failure + + # The handler should be connected to the signal + # We can test this by verifying the signal has receivers + receivers = task_failure.receivers + assert len(receivers) > 0 + + # Simply verify that importing the handler doesn't cause errors + # The actual signal connection is tested implicitly by the other tests + assert callable(task_failure_handler) diff --git a/tests/test_views_google_drive.py b/tests/test_views_google_drive.py index af047299..5c419814 100644 --- a/tests/test_views_google_drive.py +++ b/tests/test_views_google_drive.py @@ -1,6 +1,8 @@ """Tests for app/views/google_drive.py module.""" import pytest +from unittest.mock import patch +import urllib.parse @pytest.mark.integration @@ -26,3 +28,105 @@ class TestGoogleDriveViews: """Test the Google Drive OAuth callback with auth code.""" response = client.get("/google-drive-callback?code=test_code") assert response.status_code == 200 + + def test_google_drive_callback_with_code_and_state(self, client): + """Test the Google Drive OAuth callback with code and state.""" + response = client.get("/google-drive-callback?code=test_code&state=test_state") + assert response.status_code == 200 + + def test_google_drive_auth_start_with_redirect_uri(self, client): + """Test starting Google Drive OAuth flow with explicit redirect_uri.""" + client_id = "test_client_id_123" + redirect_uri = "https://example.com/callback" + response = client.get( + f"/google-drive-auth-start?client_id={client_id}&redirect_uri={redirect_uri}", + follow_redirects=False + ) + + assert response.status_code in [302, 307] # Redirect status codes + + # Verify redirect location + location = response.headers.get("location") + assert location is not None + assert "accounts.google.com/o/oauth2/auth" in location + assert f"client_id={client_id}" in location + assert urllib.parse.quote(redirect_uri) in location + assert "response_type=code" in location + assert "access_type=offline" in location + assert "prompt=consent" in location + # Verify scope includes drive.file + assert "scope=" in location + + def test_google_drive_auth_start_without_redirect_uri(self, client): + """Test starting Google Drive OAuth flow without explicit redirect_uri.""" + client_id = "test_client_id_456" + response = client.get( + f"/google-drive-auth-start?client_id={client_id}", + follow_redirects=False + ) + + assert response.status_code in [302, 307] # Redirect status codes + + # Verify redirect location + location = response.headers.get("location") + assert location is not None + assert "accounts.google.com/o/oauth2/auth" in location + assert f"client_id={client_id}" in location + # Should use default redirect_uri based on request host + assert "redirect_uri=" in location + + def test_google_drive_auth_start_scope_configuration(self, client): + """Test that Google Drive auth start uses correct OAuth scope.""" + client_id = "test_client_id_789" + response = client.get( + f"/google-drive-auth-start?client_id={client_id}", + follow_redirects=False + ) + + location = response.headers.get("location") + assert location is not None + + # The scope should be URL encoded, so check for the encoded version + # drive.file scope: https://www.googleapis.com/auth/drive.file + expected_scope = urllib.parse.quote("https://www.googleapis.com/auth/drive.file") + assert expected_scope in location + + @patch("app.views.google_drive.settings") + def test_google_drive_setup_page_with_folder_id_empty(self, mock_settings, client): + """Test setup page when folder_id is None/empty - should show not configured.""" + mock_settings.google_drive_use_oauth = False + mock_settings.google_drive_client_id = "test_client_id" + mock_settings.google_drive_client_secret = "test_secret" + mock_settings.google_drive_refresh_token = "test_token" + mock_settings.google_drive_credentials_json = '{"test": "creds"}' + mock_settings.google_drive_folder_id = None # Empty folder ID + + response = client.get("/google-drive-setup") + assert response.status_code == 200 + # The page should indicate not fully configured due to missing folder_id + + @patch("app.views.google_drive.settings") + def test_google_drive_setup_page_oauth_mode(self, mock_settings, client): + """Test setup page in OAuth mode.""" + mock_settings.google_drive_use_oauth = True + mock_settings.google_drive_client_id = "oauth_client_id" + mock_settings.google_drive_client_secret = "oauth_secret" + mock_settings.google_drive_refresh_token = "oauth_token" + mock_settings.google_drive_folder_id = "test_folder_id" + mock_settings.google_drive_credentials_json = None + + response = client.get("/google-drive-setup") + assert response.status_code == 200 + + @patch("app.views.google_drive.settings") + def test_google_drive_setup_page_service_account_mode(self, mock_settings, client): + """Test setup page in service account mode.""" + mock_settings.google_drive_use_oauth = False + mock_settings.google_drive_credentials_json = '{"type": "service_account"}' + mock_settings.google_drive_folder_id = "test_folder_id" + mock_settings.google_drive_client_id = None + mock_settings.google_drive_client_secret = None + mock_settings.google_drive_refresh_token = None + + response = client.get("/google-drive-setup") + assert response.status_code == 200