"""Comprehensive unit tests for app/api/google_drive.py module.""" from datetime import datetime, timedelta from unittest.mock import MagicMock, patch import pytest @pytest.mark.unit class TestExchangeGoogleDriveToken: """Tests for POST /google-drive/exchange-token endpoint.""" @patch("app.api.google_drive.exchange_oauth_token") def test_exchange_token_success(self, mock_exchange): """Test successful token exchange.""" mock_exchange.return_value = { "refresh_token": "refresh_token_value", "access_token": "access_token_value", "expires_in": 3600, } # Response should include tokens pass @patch("app.api.google_drive.exchange_oauth_token") def test_exchange_token_default_expires_in(self, mock_exchange): """Test default expires_in when not provided.""" mock_exchange.return_value = { "refresh_token": "refresh_token_value", "access_token": "access_token_value", } # Should use default expires_in of 3600 pass @pytest.mark.unit class TestUpdateGoogleDriveSettings: """Tests for POST /google-drive/update-settings endpoint.""" def test_update_settings_oauth_enabled(self): """Test updating settings with OAuth enabled.""" # Should update OAuth credentials pass def test_update_settings_oauth_disabled(self): """Test updating settings with OAuth disabled.""" # Should set use_oauth to False pass def test_update_settings_use_oauth_variations(self): """Test various true/false string values for use_oauth.""" # Should handle "true", "1", "yes", "y", "t" # Should handle "false", "0", "no", "n", "f" pass def test_update_settings_exception_handling(self): """Test handling of exceptions.""" # Should raise HTTPException with 500 status pass @pytest.mark.unit class TestTestGoogleDriveToken: """Tests for GET /google-drive/test-token endpoint.""" @patch("app.tasks.upload_to_google_drive.get_drive_service_oauth") @patch("google.oauth2.credentials.Credentials") def test_test_token_oauth_success(self, mock_creds_class, mock_service): """Test successful OAuth token validation.""" from app.config import settings # Mock credentials mock_creds = MagicMock() mock_creds.valid = True mock_creds.expiry = datetime.now() + timedelta(hours=1) mock_creds_class.return_value = mock_creds # Mock service response mock_service_obj = MagicMock() mock_about = MagicMock() mock_about.execute.return_value = {"user": {"emailAddress": "test@example.com"}} mock_service_obj.about.return_value.get.return_value = mock_about mock_service.return_value = mock_service_obj with patch.object(settings, "google_drive_use_oauth", True): with patch.object(settings, "google_drive_client_id", "client_id"): with patch.object(settings, "google_drive_client_secret", "secret"): with patch.object(settings, "google_drive_refresh_token", "token"): # Should return success with OAuth pass @patch("app.tasks.upload_to_google_drive.get_google_drive_service") def test_test_token_service_account_success(self, mock_service): """Test successful service account validation.""" from app.config import settings # Mock service response mock_service_obj = MagicMock() mock_about = MagicMock() mock_about.execute.return_value = {"user": {"emailAddress": "service@example.com"}} mock_service_obj.about.return_value.get.return_value = mock_about mock_service.return_value = mock_service_obj with patch.object(settings, "google_drive_use_oauth", False): with patch.object(settings, "google_drive_credentials_json", "{}"): # Should return success with service account pass def test_test_token_oauth_not_configured(self): """Test when OAuth credentials are not fully configured.""" from app.config import settings with patch.object(settings, "google_drive_use_oauth", True): with patch.object(settings, "google_drive_client_id", None): # Should return error pass def test_test_token_service_account_not_configured(self): """Test when service account is not configured.""" from app.config import settings with patch.object(settings, "google_drive_use_oauth", False): with patch.object(settings, "google_drive_credentials_json", None): # Should return error pass @patch("app.tasks.upload_to_google_drive.get_drive_service_oauth") def test_test_token_oauth_invalid_grant(self, mock_service): """Test OAuth with invalid_grant error.""" mock_service.side_effect = Exception("invalid_grant: Token expired") from app.config import settings with patch.object(settings, "google_drive_use_oauth", True): with patch.object(settings, "google_drive_client_id", "client_id"): with patch.object(settings, "google_drive_client_secret", "secret"): with patch.object(settings, "google_drive_refresh_token", "token"): # Should return error with needs_reauth pass @patch("google.oauth2.credentials.Credentials") @patch("app.tasks.upload_to_google_drive.get_drive_service_oauth") def test_test_token_refresh_invalid_credentials(self, mock_service, mock_creds_class): """Test token refresh with invalid credentials.""" from app.config import settings mock_creds = MagicMock() mock_creds.valid = False mock_creds.refresh.side_effect = Exception("Token refresh failed") mock_creds_class.return_value = mock_creds with patch.object(settings, "google_drive_use_oauth", True): with patch.object(settings, "google_drive_client_id", "client_id"): with patch.object(settings, "google_drive_client_secret", "secret"): with patch.object(settings, "google_drive_refresh_token", "token"): # Should handle refresh error pass def test_test_token_service_account_with_delegation(self): """Test service account with delegated user.""" from app.config import settings with patch.object(settings, "google_drive_use_oauth", False): with patch.object(settings, "google_drive_credentials_json", "{}"): with patch.object(settings, "google_drive_delegate_to", "user@example.com"): # Should include delegation info in response pass @pytest.mark.unit class TestGetGoogleDriveTokenInfo: """Tests for GET /google-drive/get-token-info endpoint.""" @patch("google.oauth2.credentials.Credentials") def test_get_token_info_success(self, mock_creds_class): """Test successful token info retrieval.""" from app.config import settings mock_creds = MagicMock() mock_creds.valid = True mock_creds.token = "access_token_value" mock_creds.expiry = datetime.now() + timedelta(hours=1) mock_creds_class.return_value = mock_creds with patch.object(settings, "google_drive_use_oauth", True): with patch.object(settings, "google_drive_client_id", "client_id"): with patch.object(settings, "google_drive_client_secret", "secret"): with patch.object(settings, "google_drive_refresh_token", "token"): # Should return access token pass def test_get_token_info_oauth_not_enabled(self): """Test when OAuth is not enabled.""" from app.config import settings with patch.object(settings, "google_drive_use_oauth", False): # Should return error pass def test_get_token_info_not_configured(self): """Test when OAuth not configured.""" from app.config import settings with patch.object(settings, "google_drive_use_oauth", True): with patch.object(settings, "google_drive_client_id", None): # Should return error pass @patch("google.oauth2.credentials.Credentials") def test_get_token_info_refresh_token(self, mock_creds_class): """Test token refresh when not valid.""" from app.config import settings mock_creds = MagicMock() mock_creds.valid = False mock_creds.token = "new_access_token" mock_creds.expiry = datetime.now() + timedelta(hours=1) mock_creds_class.return_value = mock_creds with patch.object(settings, "google_drive_use_oauth", True): with patch.object(settings, "google_drive_client_id", "client_id"): with patch.object(settings, "google_drive_client_secret", "secret"): with patch.object(settings, "google_drive_refresh_token", "token"): # Should refresh and return new token pass @pytest.mark.unit class TestFormatTimeRemaining: """Tests for format_time_remaining helper function.""" def test_format_expired(self): """Test formatting expired time.""" from app.api.google_drive import format_time_remaining delta = timedelta(seconds=-100) result = format_time_remaining(delta) assert result == "Expired" def test_format_days_only(self): """Test formatting with only days.""" from app.api.google_drive import format_time_remaining delta = timedelta(days=5) result = format_time_remaining(delta) assert "5 days" in result def test_format_hours_only(self): """Test formatting with only hours.""" from app.api.google_drive import format_time_remaining delta = timedelta(hours=3) result = format_time_remaining(delta) assert "3 hours" in result def test_format_minutes_only(self): """Test formatting with only minutes.""" from app.api.google_drive import format_time_remaining delta = timedelta(minutes=45) result = format_time_remaining(delta) assert "45 minutes" in result def test_format_days_and_hours(self): """Test formatting with days and hours.""" from app.api.google_drive import format_time_remaining delta = timedelta(days=2, hours=5) result = format_time_remaining(delta) assert "2 days" in result assert "5 hours" in result def test_format_no_minutes_when_days(self): """Test that minutes are not shown when days > 0.""" from app.api.google_drive import format_time_remaining delta = timedelta(days=1, minutes=30) result = format_time_remaining(delta) assert "minutes" not in result def test_format_singular_units(self): """Test singular forms (1 day, 1 hour, 1 minute).""" from app.api.google_drive import format_time_remaining delta = timedelta(days=1, hours=1, minutes=1) result = format_time_remaining(delta) # Should use singular forms pass @pytest.mark.unit class TestSaveGoogleDriveSettings: """Tests for POST /google-drive/save-settings endpoint.""" @patch("builtins.open", create=True) @patch("os.path.exists") def test_save_settings_success(self, mock_exists, mock_open): """Test successful saving to .env file.""" mock_exists.return_value = True mock_file = MagicMock() mock_file.readlines.return_value = [] mock_open.return_value.__enter__.return_value = mock_file # Should save settings pass @patch("os.path.exists") def test_save_settings_no_env_file(self, mock_exists): """Test when .env file doesn't exist.""" mock_exists.return_value = False # Should continue with in-memory update only pass @patch("builtins.open", create=True) @patch("os.path.exists") def test_save_settings_oauth_true(self, mock_exists, mock_open): """Test saving with OAuth enabled.""" mock_exists.return_value = True mock_file = MagicMock() mock_file.readlines.return_value = [] mock_open.return_value.__enter__.return_value = mock_file # Should save OAuth credentials pass @patch("builtins.open", create=True) @patch("os.path.exists") def test_save_settings_oauth_false(self, mock_exists, mock_open): """Test saving with OAuth disabled.""" mock_exists.return_value = True mock_file = MagicMock() mock_file.readlines.return_value = [] mock_open.return_value.__enter__.return_value = mock_file # Should not save OAuth credentials pass @patch("builtins.open", create=True) @patch("os.path.exists") def test_save_settings_file_write_error(self, mock_exists, mock_open): """Test handling of file write errors.""" mock_exists.return_value = True mock_open.side_effect = IOError("Write error") # Should log warning but continue with in-memory update pass @patch("os.path.exists") def test_save_settings_in_memory_only_flag(self, mock_exists): """Test in_memory_only flag when .env doesn't exist.""" mock_exists.return_value = False # Response should have in_memory_only: True