diff --git a/app/api/url_upload.py b/app/api/url_upload.py index c6fb81e2..7367138b 100644 --- a/app/api/url_upload.py +++ b/app/api/url_upload.py @@ -12,7 +12,7 @@ from typing import Optional import requests from fastapi import APIRouter, HTTPException, Request -from pydantic import BaseModel, HttpUrl, validator +from pydantic import BaseModel, HttpUrl, field_validator from app.auth import require_login from app.config import settings @@ -31,7 +31,8 @@ class URLUploadRequest(BaseModel): url: HttpUrl filename: Optional[str] = None - @validator("url") + @field_validator("url") + @classmethod def validate_url_scheme(cls, v): """Ensure only HTTP/HTTPS schemes are allowed""" parsed = urllib.parse.urlparse(str(v)) diff --git a/app/config.py b/app/config.py index c19b79fa..391e2b22 100644 --- a/app/config.py +++ b/app/config.py @@ -3,11 +3,13 @@ import os from typing import Any, List, Optional, Union -from pydantic import Field, validator -from pydantic_settings import BaseSettings +from pydantic import Field, field_validator +from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): + model_config = SettingsConfigDict(env_file=".env") + database_url: str redis_url: str openai_api_key: str @@ -265,7 +267,8 @@ class Settings(BaseSettings): description="Stricter rate limit for authentication endpoints to prevent brute force attacks.", ) - @validator("notification_urls", pre=True) + @field_validator("notification_urls", mode="before") + @classmethod def parse_notification_urls(cls, v): """Parse notification URLs from string or list""" if isinstance(v, str): @@ -276,12 +279,13 @@ class Settings(BaseSettings): return [] return v - @validator("session_secret") - def validate_session_secret(cls, v, values): + @field_validator("session_secret") + @classmethod + def validate_session_secret(cls, v, info): """Validate that session_secret is set and has sufficient length when auth is enabled""" - if values.get("auth_enabled") and not v: + if info.data.get("auth_enabled") and not v: raise ValueError("SESSION_SECRET must be set when AUTH_ENABLED=True") - if values.get("auth_enabled") and v and len(v) < 32: + if info.data.get("auth_enabled") and v and len(v) < 32: raise ValueError("SESSION_SECRET must be at least 32 characters long") return v @@ -347,28 +351,5 @@ class Settings(BaseSettings): # Return basic info if file not found return f"Version: {self.version}\nBuild Date: {self.build_date}\nGit SHA: {self.git_sha}" - class Config: - env_file = ".env" - - # Convert string representations of booleans to actual booleans - # and strip quotes from string values - @classmethod - def parse_env_var(cls, field_name: str, raw_val: str) -> Any: - # First, strip quotes from the value if it's a string - if isinstance(raw_val, str): - if (raw_val.startswith('"') and raw_val.endswith('"')) or ( - raw_val.startswith("'") and raw_val.endswith("'") - ): - raw_val = raw_val[1:-1] - raw_val = raw_val.strip() - - # Convert string representations of booleans to actual booleans - if field_name.endswith("_enabled") or field_name == "debug": - if raw_val.lower() in ("false", "0", "no", "n", "f"): - return False - if raw_val.lower() in ("true", "1", "yes", "y", "t"): - return True - return raw_val - settings = Settings() diff --git a/app/database.py b/app/database.py index 5df280a8..6451ae21 100644 --- a/app/database.py +++ b/app/database.py @@ -5,7 +5,7 @@ import os from sqlalchemy import create_engine, exc from sqlalchemy.engine.url import make_url -from sqlalchemy.ext.declarative import declarative_base +from sqlalchemy.orm import declarative_base from sqlalchemy.orm import sessionmaker from app.config import settings diff --git a/app/utils/config_validator/settings_display.py b/app/utils/config_validator/settings_display.py index 75399c99..e7771ea1 100644 --- a/app/utils/config_validator/settings_display.py +++ b/app/utils/config_validator/settings_display.py @@ -9,12 +9,15 @@ from app.utils.config_validator.masking import mask_sensitive_value logger = logging.getLogger(__name__) +# Pydantic model attributes that should not be iterated as user settings +_PYDANTIC_INTERNALS = {"model_computed_fields", "model_config", "model_extra", "model_fields", "model_fields_set"} + def dump_all_settings(): """Log all settings values for diagnostic purposes""" logger.info("--- DUMPING ALL SETTINGS FOR DIAGNOSTIC PURPOSES ---") for key in dir(settings): - if not key.startswith("_") and not callable(getattr(settings, key)): + if not key.startswith("_") and key not in _PYDANTIC_INTERNALS and not callable(getattr(settings, key)): value = getattr(settings, key) # Mask sensitive values in logs if ( @@ -190,8 +193,8 @@ def get_settings_for_display(show_values=False): key for key in dir(settings) if not key.startswith("_") + and key not in _PYDANTIC_INTERNALS and not callable(getattr(settings, key)) - and key not in ["model_computed_fields", "model_config", "model_extra", "model_fields", "model_fields_set"] ] ) diff --git a/tests/test_api_dropbox.py b/tests/test_api_dropbox.py new file mode 100644 index 00000000..67bb01f3 --- /dev/null +++ b/tests/test_api_dropbox.py @@ -0,0 +1,309 @@ +""" +Tests for app/api/dropbox.py module. + +Covers Dropbox OAuth endpoints, settings management, and token testing. +""" + +import os +from unittest.mock import MagicMock, Mock, patch + +import pytest +import requests + + +@pytest.mark.unit +class TestExchangeDropboxToken: + """Tests for exchange_dropbox_token endpoint.""" + + @patch("app.api.dropbox.exchange_oauth_token") + def test_successful_exchange(self, mock_exchange, client): + """Test successful OAuth token exchange.""" + mock_exchange.return_value = { + "refresh_token": "test-refresh", + "access_token": "test-access", + "expires_in": 14400, + } + + response = client.post( + "/api/dropbox/exchange-token", + data={ + "client_id": "test-client-id", + "client_secret": "test-client-secret", + "redirect_uri": "http://localhost/callback", + "code": "test-auth-code", + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["refresh_token"] == "test-refresh" + assert data["access_token"] == "test-access" + assert data["expires_in"] == 14400 + + @patch("app.api.dropbox.exchange_oauth_token") + def test_exchange_default_expiry(self, mock_exchange, client): + """Test token exchange returns default expires_in when not provided.""" + mock_exchange.return_value = { + "refresh_token": "test-refresh", + "access_token": "test-access", + } + + response = client.post( + "/api/dropbox/exchange-token", + data={ + "client_id": "test-client-id", + "client_secret": "test-client-secret", + "redirect_uri": "http://localhost/callback", + "code": "test-auth-code", + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["expires_in"] == 14400 + + +@pytest.mark.unit +class TestUpdateDropboxSettings: + """Tests for update_dropbox_settings endpoint.""" + + @patch("app.api.dropbox.settings") + def test_update_all_settings(self, mock_settings, client): + """Test updating all Dropbox settings in memory.""" + mock_settings.dropbox_refresh_token = "" + mock_settings.dropbox_app_key = "" + mock_settings.dropbox_app_secret = "" + mock_settings.dropbox_folder = "" + + response = client.post( + "/api/dropbox/update-settings", + data={ + "refresh_token": "new-refresh-token", + "app_key": "new-app-key", + "app_secret": "new-app-secret", + "folder_path": "/Documents", + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "success" + + @patch("app.api.dropbox.settings") + def test_update_refresh_token_only(self, mock_settings, client): + """Test updating only refresh token.""" + mock_settings.dropbox_refresh_token = "" + + response = client.post( + "/api/dropbox/update-settings", + data={ + "refresh_token": "new-refresh-token", + }, + ) + + assert response.status_code == 200 + + @patch("app.api.dropbox.settings") + def test_update_settings_exception(self, mock_settings, client): + """Test that exceptions return 500 error.""" + # Make setting the attribute raise an exception + type(mock_settings).dropbox_refresh_token = property( + lambda self: "", lambda self, v: (_ for _ in ()).throw(RuntimeError("forced")) + ) + + response = client.post( + "/api/dropbox/update-settings", + data={"refresh_token": "token"}, + ) + + assert response.status_code == 500 + + +@pytest.mark.unit +class TestTestDropboxToken: + """Tests for test_dropbox_token endpoint.""" + + @patch("app.api.dropbox.settings") + def test_not_configured(self, mock_settings, client): + """Test response when Dropbox is not configured.""" + mock_settings.dropbox_refresh_token = "" + mock_settings.dropbox_app_key = "" + mock_settings.dropbox_app_secret = "" + + response = client.get("/api/dropbox/test-token") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "error" + assert "not fully configured" in data["message"] + + @patch("app.api.dropbox.requests.post") + @patch("app.api.dropbox.settings") + def test_valid_token(self, mock_settings, mock_post, client): + """Test successful token validation.""" + mock_settings.dropbox_refresh_token = "valid-refresh-token" + mock_settings.dropbox_app_key = "app-key" + mock_settings.dropbox_app_secret = "app-secret" + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "email": "user@example.com", + "name": {"display_name": "Test User"}, + } + mock_post.return_value = mock_response + + response = client.get("/api/dropbox/test-token") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "success" + assert data["account"] == "user@example.com" + assert data["account_name"] == "Test User" + + @patch("app.api.dropbox.requests.post") + @patch("app.api.dropbox.settings") + def test_expired_token_refreshed(self, mock_settings, mock_post, client): + """Test that expired token triggers refresh and retry.""" + mock_settings.dropbox_refresh_token = "refresh-token" + mock_settings.dropbox_app_key = "app-key" + mock_settings.dropbox_app_secret = "app-secret" + mock_settings.http_request_timeout = 30 + + # First call returns 401, second call (refresh) returns 200, third call returns 200 + mock_401 = Mock() + mock_401.status_code = 401 + + mock_refresh = Mock() + mock_refresh.status_code = 200 + mock_refresh.json.return_value = {"access_token": "new-access-token"} + + mock_success = Mock() + mock_success.status_code = 200 + mock_success.json.return_value = { + "email": "user@example.com", + "name": {"display_name": "Test User"}, + } + + mock_post.side_effect = [mock_401, mock_refresh, mock_success] + + response = client.get("/api/dropbox/test-token") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "success" + + @patch("app.api.dropbox.requests.post") + @patch("app.api.dropbox.settings") + def test_refresh_token_expired(self, mock_settings, mock_post, client): + """Test handling when refresh token itself is expired.""" + mock_settings.dropbox_refresh_token = "expired-refresh" + mock_settings.dropbox_app_key = "app-key" + mock_settings.dropbox_app_secret = "app-secret" + mock_settings.http_request_timeout = 30 + + # First call returns 401, refresh also fails + mock_401 = Mock() + mock_401.status_code = 401 + + mock_refresh_fail = Mock() + mock_refresh_fail.status_code = 400 + mock_refresh_fail.text = "invalid_grant" + + mock_post.side_effect = [mock_401, mock_refresh_fail] + + response = client.get("/api/dropbox/test-token") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "error" + assert data["needs_reauth"] is True + + @patch("app.api.dropbox.requests.post") + @patch("app.api.dropbox.settings") + def test_token_validation_failure(self, mock_settings, mock_post, client): + """Test handling non-401, non-200 response.""" + mock_settings.dropbox_refresh_token = "token" + mock_settings.dropbox_app_key = "app-key" + mock_settings.dropbox_app_secret = "app-secret" + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.status_code = 500 + mock_response.text = "Internal Server Error" + mock_post.return_value = mock_response + + response = client.get("/api/dropbox/test-token") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "error" + + @patch("app.api.dropbox.requests.post") + @patch("app.api.dropbox.settings") + def test_connection_error(self, mock_settings, mock_post, client): + """Test handling of connection exceptions.""" + mock_settings.dropbox_refresh_token = "token" + mock_settings.dropbox_app_key = "app-key" + mock_settings.dropbox_app_secret = "app-secret" + mock_settings.http_request_timeout = 30 + + mock_post.side_effect = requests.exceptions.ConnectionError("Connection refused") + + response = client.get("/api/dropbox/test-token") + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "error" + assert "Connection error" in data["message"] + + +@pytest.mark.unit +class TestSaveDropboxSettings: + """Tests for save_dropbox_settings endpoint.""" + + @patch("app.api.dropbox.settings") + def test_save_settings_env_not_found(self, mock_settings, client): + """Test error when .env file is not found.""" + # The endpoint constructs the env path using __file__ + with patch("os.path.exists", return_value=False): + response = client.post( + "/api/dropbox/save-settings", + data={"refresh_token": "test-token"}, + ) + + assert response.status_code == 500 + + @patch("app.api.dropbox.settings") + def test_save_settings_success(self, mock_settings, client, tmp_path): + """Test successful save of Dropbox settings to .env file.""" + mock_settings.dropbox_refresh_token = "" + mock_settings.dropbox_app_key = "" + mock_settings.dropbox_app_secret = "" + mock_settings.dropbox_folder = "" + + # Create a temporary .env file + env_file = tmp_path / ".env" + env_file.write_text("DROPBOX_REFRESH_TOKEN=old_token\nOTHER_VAR=value\n") + + with ( + patch("app.api.dropbox.os.path.join", return_value=str(env_file)), + patch("app.api.dropbox.os.path.exists", return_value=True), + patch("app.api.dropbox.os.path.dirname", return_value=str(tmp_path)), + ): + response = client.post( + "/api/dropbox/save-settings", + data={ + "refresh_token": "new-token", + "app_key": "new-key", + }, + ) + + assert response.status_code == 200 + data = response.json() + assert data["status"] == "success" + + # Verify the .env file was updated + content = env_file.read_text() + assert "new-token" in content diff --git a/tests/test_config_validator_coverage.py b/tests/test_config_validator_coverage.py new file mode 100644 index 00000000..c870d806 --- /dev/null +++ b/tests/test_config_validator_coverage.py @@ -0,0 +1,99 @@ +""" +Tests for app/utils/config_validator.py module coverage. + +Ensures all imports and __all__ exports in the re-export module are exercised +so coverage tools track the module as covered. +""" + +import pytest + + +@pytest.mark.unit +class TestConfigValidatorModuleCoverage: + """Ensure every line in config_validator.py is exercised by coverage.""" + + def test_all_imports_and_exports_exercised(self): + """Import every symbol from config_validator to ensure line coverage.""" + # These imports exercise lines 7-17 (import statements) + from app.utils.config_validator import ( + check_all_configs, + dump_all_settings, + get_provider_status, + get_settings_for_display, + mask_sensitive_value, + validate_email_config, + validate_notification_config, + validate_storage_configs, + ) + + # Verify all functions are callable (exercises the __all__ list, lines 19-28) + for fn in [ + validate_email_config, + validate_storage_configs, + validate_notification_config, + mask_sensitive_value, + get_provider_status, + get_settings_for_display, + dump_all_settings, + check_all_configs, + ]: + assert callable(fn) + + def test_all_list_contents(self): + """Verify __all__ is correctly defined and complete.""" + import app.utils.config_validator as mod + + expected = { + "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", + } + assert set(mod.__all__) == expected + + def test_mask_sensitive_value_returns_masked(self): + """Test that mask_sensitive_value masks a real value.""" + from app.utils.config_validator import mask_sensitive_value + + result = mask_sensitive_value("my_secret_value_12345") + assert "my_secret_value_12345" != result + + def test_validate_storage_configs_returns_dict(self): + """Test validate_storage_configs returns a dict.""" + from app.utils.config_validator import validate_storage_configs + + result = validate_storage_configs() + assert isinstance(result, dict) + + def test_validate_email_config_returns_list(self): + """Test validate_email_config returns a list.""" + from app.utils.config_validator import validate_email_config + + result = validate_email_config() + assert isinstance(result, list) + + def test_validate_notification_config_returns_list(self): + """Test validate_notification_config returns a list.""" + from app.utils.config_validator import validate_notification_config + + result = validate_notification_config() + assert isinstance(result, list) + + def test_get_provider_status_returns_dict(self): + """Test get_provider_status returns a dict.""" + from app.utils.config_validator import get_provider_status + + result = get_provider_status() + assert isinstance(result, dict) + + def test_check_all_configs_returns_dict(self): + """Test check_all_configs returns a dict.""" + from app.utils.config_validator import check_all_configs + + result = check_all_configs() + assert isinstance(result, dict) diff --git a/tests/test_migrate_logs_to_steps.py b/tests/test_migrate_logs_to_steps.py new file mode 100644 index 00000000..1754edb5 --- /dev/null +++ b/tests/test_migrate_logs_to_steps.py @@ -0,0 +1,469 @@ +""" +Tests for app/utils/migrate_logs_to_steps.py module. + +Covers migrate_logs_to_steps, _parse_logs_to_step_states, migrate_all_files, +and verify_migration with comprehensive unit tests. +""" + +from datetime import datetime, timezone +from unittest.mock import MagicMock, patch + +import pytest + +from app.models import FileProcessingStep, ProcessingLog +from app.utils.migrate_logs_to_steps import ( + _parse_logs_to_step_states, + migrate_all_files, + migrate_logs_to_steps, + verify_migration, +) + + +def _make_log(file_id, step_name, status, message=None, timestamp=None): + """Helper to create a mock ProcessingLog.""" + log = MagicMock(spec=ProcessingLog) + log.file_id = file_id + log.step_name = step_name + log.status = status + log.message = message + log.timestamp = timestamp or datetime.now(timezone.utc) + return log + + +def _make_step(file_id, step_name, status, started_at=None, completed_at=None, error_message=None): + """Helper to create a mock FileProcessingStep.""" + step = MagicMock(spec=FileProcessingStep) + step.file_id = file_id + step.step_name = step_name + step.status = status + step.started_at = started_at + step.completed_at = completed_at + step.error_message = error_message + return step + + +@pytest.mark.unit +class TestParseLogsToStepStates: + """Tests for _parse_logs_to_step_states function.""" + + def test_empty_logs(self): + """Test parsing empty log list returns empty dict.""" + result = _parse_logs_to_step_states([]) + assert result == {} + + def test_single_in_progress_log(self): + """Test parsing a single in_progress log entry.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + logs = [_make_log(1, "hash_file", "in_progress", timestamp=ts)] + + result = _parse_logs_to_step_states(logs) + + assert "hash_file" in result + assert result["hash_file"]["status"] == "in_progress" + assert result["hash_file"]["started_at"] == ts + assert result["hash_file"]["completed_at"] is None + + def test_in_progress_then_success(self): + """Test step that starts and completes successfully.""" + ts1 = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + ts2 = datetime(2024, 1, 1, 12, 0, 5, tzinfo=timezone.utc) + logs = [ + _make_log(1, "hash_file", "in_progress", timestamp=ts1), + _make_log(1, "hash_file", "success", timestamp=ts2), + ] + + result = _parse_logs_to_step_states(logs) + + assert result["hash_file"]["status"] == "success" + assert result["hash_file"]["started_at"] == ts1 + assert result["hash_file"]["completed_at"] == ts2 + assert result["hash_file"]["error_message"] is None + + def test_in_progress_then_failure(self): + """Test step that starts and fails.""" + ts1 = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + ts2 = datetime(2024, 1, 1, 12, 0, 5, tzinfo=timezone.utc) + logs = [ + _make_log(1, "ocr", "in_progress", timestamp=ts1), + _make_log(1, "ocr", "failure", message="OCR failed", timestamp=ts2), + ] + + result = _parse_logs_to_step_states(logs) + + assert result["ocr"]["status"] == "failure" + assert result["ocr"]["started_at"] == ts1 + assert result["ocr"]["completed_at"] == ts2 + assert result["ocr"]["error_message"] == "OCR failed" + + def test_success_without_in_progress(self): + """Test step that goes directly to success without in_progress.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + logs = [_make_log(1, "hash_file", "success", timestamp=ts)] + + result = _parse_logs_to_step_states(logs) + + assert result["hash_file"]["status"] == "success" + assert result["hash_file"]["started_at"] == ts + assert result["hash_file"]["completed_at"] == ts + + def test_failure_without_in_progress(self): + """Test step that goes directly to failure without in_progress.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + logs = [_make_log(1, "upload", "failure", message="Connection error", timestamp=ts)] + + result = _parse_logs_to_step_states(logs) + + assert result["upload"]["status"] == "failure" + assert result["upload"]["started_at"] == ts + assert result["upload"]["error_message"] == "Connection error" + + def test_pending_status(self): + """Test step with pending status.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + logs = [_make_log(1, "upload", "pending", timestamp=ts)] + + result = _parse_logs_to_step_states(logs) + + assert result["upload"]["status"] == "pending" + assert result["upload"]["started_at"] is None + + def test_queued_status(self): + """Test step with queued status.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + logs = [_make_log(1, "upload", "queued", timestamp=ts)] + + result = _parse_logs_to_step_states(logs) + + assert result["upload"]["status"] == "pending" + + def test_multiple_steps(self): + """Test parsing logs for multiple steps.""" + ts1 = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + ts2 = datetime(2024, 1, 1, 12, 0, 5, tzinfo=timezone.utc) + ts3 = datetime(2024, 1, 1, 12, 0, 10, tzinfo=timezone.utc) + ts4 = datetime(2024, 1, 1, 12, 0, 15, tzinfo=timezone.utc) + logs = [ + _make_log(1, "hash_file", "in_progress", timestamp=ts1), + _make_log(1, "hash_file", "success", timestamp=ts2), + _make_log(1, "ocr", "in_progress", timestamp=ts3), + _make_log(1, "ocr", "success", timestamp=ts4), + ] + + result = _parse_logs_to_step_states(logs) + + assert len(result) == 2 + assert result["hash_file"]["status"] == "success" + assert result["ocr"]["status"] == "success" + + def test_retry_overwrites_previous_state(self): + """Test that later success after failure wins (retry scenario).""" + ts1 = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + ts2 = datetime(2024, 1, 1, 12, 0, 5, tzinfo=timezone.utc) + ts3 = datetime(2024, 1, 1, 12, 0, 10, tzinfo=timezone.utc) + ts4 = datetime(2024, 1, 1, 12, 0, 15, tzinfo=timezone.utc) + logs = [ + _make_log(1, "upload", "in_progress", timestamp=ts1), + _make_log(1, "upload", "failure", message="timeout", timestamp=ts2), + _make_log(1, "upload", "in_progress", timestamp=ts3), + _make_log(1, "upload", "success", timestamp=ts4), + ] + + result = _parse_logs_to_step_states(logs) + + assert result["upload"]["status"] == "success" + assert result["upload"]["error_message"] is None + assert result["upload"]["completed_at"] == ts4 + + def test_pending_does_not_override_in_progress(self): + """Test that pending does not override in_progress status.""" + ts1 = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + ts2 = datetime(2024, 1, 1, 12, 0, 5, tzinfo=timezone.utc) + logs = [ + _make_log(1, "upload", "in_progress", timestamp=ts1), + _make_log(1, "upload", "pending", timestamp=ts2), + ] + + result = _parse_logs_to_step_states(logs) + + assert result["upload"]["status"] == "in_progress" + + +@pytest.mark.unit +class TestMigrateLogsToSteps: + """Tests for migrate_logs_to_steps function.""" + + def test_no_logs_found(self, db_session): + """Test migration when no logs exist for the file.""" + result = migrate_logs_to_steps(db_session, file_id=999) + + assert result["file_id"] == 999 + assert result["steps_created"] == 0 + assert result["steps_updated"] == 0 + assert result["steps_skipped"] == 0 + assert result["errors"] == [] + + def test_creates_new_steps(self, db_session): + """Test migration creates FileProcessingStep entries from logs.""" + ts1 = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + ts2 = datetime(2024, 1, 1, 12, 0, 5, tzinfo=timezone.utc) + + # Add processing logs + log1 = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="in_progress", timestamp=ts1) + log2 = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="success", timestamp=ts2) + db_session.add_all([log1, log2]) + db_session.commit() + + result = migrate_logs_to_steps(db_session, file_id=1) + + assert result["steps_created"] == 1 + assert result["steps_updated"] == 0 + assert result["errors"] == [] + + # Verify step was actually created + steps = db_session.query(FileProcessingStep).filter(FileProcessingStep.file_id == 1).all() + assert len(steps) == 1 + assert steps[0].step_name == "hash_file" + assert steps[0].status == "success" + + def test_updates_existing_step(self, db_session): + """Test migration updates existing step when status differs.""" + ts1 = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + ts2 = datetime(2024, 1, 1, 12, 0, 5, tzinfo=timezone.utc) + + # Add a log with success + log = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="success", timestamp=ts2) + db_session.add(log) + + # Add existing step with different status + existing = FileProcessingStep(file_id=1, step_name="hash_file", status="in_progress", started_at=ts1) + db_session.add(existing) + db_session.commit() + + result = migrate_logs_to_steps(db_session, file_id=1) + + assert result["steps_updated"] == 1 + assert result["steps_created"] == 0 + + def test_skips_unchanged_step(self, db_session): + """Test migration skips step that is already up to date.""" + ts1 = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + ts2 = datetime(2024, 1, 1, 12, 0, 5, tzinfo=timezone.utc) + + # Add logs + log1 = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="in_progress", timestamp=ts1) + log2 = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="success", timestamp=ts2) + db_session.add_all([log1, log2]) + + # Add existing step with the same values + existing = FileProcessingStep( + file_id=1, step_name="hash_file", status="success", started_at=ts1, completed_at=ts2 + ) + db_session.add(existing) + db_session.commit() + + result = migrate_logs_to_steps(db_session, file_id=1) + + assert result["steps_skipped"] == 1 + assert result["steps_created"] == 0 + assert result["steps_updated"] == 0 + + def test_dry_run_does_not_commit(self, db_session): + """Test that dry_run mode does not persist changes.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + log = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="success", timestamp=ts) + db_session.add(log) + db_session.commit() + + result = migrate_logs_to_steps(db_session, file_id=1, dry_run=True) + + assert result["steps_created"] == 1 + + # After dry run + rollback, no steps should be persisted + steps = db_session.query(FileProcessingStep).filter(FileProcessingStep.file_id == 1).all() + assert len(steps) == 0 + + def test_handles_exception_gracefully(self, db_session): + """Test that exceptions are caught and reported.""" + with patch.object(db_session, "query", side_effect=Exception("DB error")): + result = migrate_logs_to_steps(db_session, file_id=1) + + assert "DB error" in result["errors"][0] + assert result["steps_created"] == 0 + + +@pytest.mark.unit +class TestMigrateAllFiles: + """Tests for migrate_all_files function.""" + + def test_no_files_to_migrate(self, db_session): + """Test with no files needing migration.""" + result = migrate_all_files(db_session) + + assert result["total_files"] == 0 + assert result["files_migrated"] == 0 + assert result["files_failed"] == 0 + + def test_migrates_files_with_logs_but_no_steps(self, db_session): + """Test migrating files that have logs but no steps.""" + ts1 = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + ts2 = datetime(2024, 1, 1, 12, 0, 5, tzinfo=timezone.utc) + + # Add logs for file 1 + log1 = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="in_progress", timestamp=ts1) + log2 = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="success", timestamp=ts2) + db_session.add_all([log1, log2]) + db_session.commit() + + result = migrate_all_files(db_session) + + assert result["total_files"] == 1 + assert result["files_migrated"] == 1 + assert result["total_steps_created"] == 1 + + def test_skips_files_already_with_steps(self, db_session): + """Test that files with existing steps are skipped.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + + # Add log and step for file 1 + log = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="success", timestamp=ts) + step = FileProcessingStep(file_id=1, step_name="hash_file", status="success") + db_session.add_all([log, step]) + db_session.commit() + + result = migrate_all_files(db_session) + + assert result["total_files"] == 0 + + def test_handles_file_with_none_file_id(self, db_session): + """Test that logs with None file_id are excluded.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + log = ProcessingLog(file_id=None, task_id="t1", step_name="hash_file", status="success", timestamp=ts) + db_session.add(log) + db_session.commit() + + result = migrate_all_files(db_session) + + assert result["total_files"] == 0 + + def test_batch_processing(self, db_session): + """Test batch processing with small batch size.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + + for i in range(3): + log = ProcessingLog(file_id=i + 10, task_id=f"t{i}", step_name="hash_file", status="success", timestamp=ts) + db_session.add(log) + db_session.commit() + + result = migrate_all_files(db_session, batch_size=2) + + assert result["total_files"] == 3 + assert result["files_migrated"] == 3 + + def test_tracks_failed_files(self, db_session): + """Test that migration errors for individual files are tracked.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + log = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="success", timestamp=ts) + db_session.add(log) + db_session.commit() + + with patch( + "app.utils.migrate_logs_to_steps.migrate_logs_to_steps", + return_value={"errors": ["forced error"], "steps_created": 0, "steps_updated": 0, "steps_skipped": 0}, + ): + result = migrate_all_files(db_session) + + assert result["files_failed"] == 1 + assert "forced error" in result["errors"] + + def test_dry_run(self, db_session): + """Test that dry_run is propagated.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + log = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="success", timestamp=ts) + db_session.add(log) + db_session.commit() + + result = migrate_all_files(db_session, dry_run=True) + + # File still processed, but changes not committed + assert result["total_files"] == 1 + + +@pytest.mark.unit +class TestVerifyMigration: + """Tests for verify_migration function.""" + + def test_no_logs_found(self, db_session): + """Test verification when no logs exist.""" + result = verify_migration(db_session, file_id=999) + + assert result["is_valid"] is False + assert "No logs found" in result["discrepancies"][0] + + def test_valid_migration(self, db_session): + """Test verification passes when migration is correct.""" + ts1 = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + ts2 = datetime(2024, 1, 1, 12, 0, 5, tzinfo=timezone.utc) + + # Add logs + log1 = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="in_progress", timestamp=ts1) + log2 = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="success", timestamp=ts2) + db_session.add_all([log1, log2]) + + # Add matching step + step = FileProcessingStep(file_id=1, step_name="hash_file", status="success", started_at=ts1, completed_at=ts2) + db_session.add(step) + db_session.commit() + + result = verify_migration(db_session, file_id=1) + + assert result["is_valid"] is True + assert result["discrepancies"] == [] + assert "hash_file" in result["log_steps"] + assert "hash_file" in result["table_steps"] + + def test_missing_step_in_table(self, db_session): + """Test verification detects missing steps.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + log = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="success", timestamp=ts) + db_session.add(log) + db_session.commit() + + result = verify_migration(db_session, file_id=1) + + assert result["is_valid"] is False + assert any("missing from table" in d for d in result["discrepancies"]) + + def test_status_mismatch(self, db_session): + """Test verification detects status mismatches.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + + log = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="success", timestamp=ts) + db_session.add(log) + + step = FileProcessingStep(file_id=1, step_name="hash_file", status="failure") + db_session.add(step) + db_session.commit() + + result = verify_migration(db_session, file_id=1) + + assert result["is_valid"] is False + assert any("status mismatch" in d for d in result["discrepancies"]) + + def test_extra_step_in_table(self, db_session): + """Test that extra steps in table are noted but don't invalidate.""" + ts = datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc) + + log = ProcessingLog(file_id=1, task_id="t1", step_name="hash_file", status="success", timestamp=ts) + db_session.add(log) + + # Add matching step + an extra one + step1 = FileProcessingStep(file_id=1, step_name="hash_file", status="success") + step2 = FileProcessingStep(file_id=1, step_name="extra_step", status="success") + db_session.add_all([step1, step2]) + db_session.commit() + + result = verify_migration(db_session, file_id=1) + + assert any("Extra step" in d for d in result["discrepancies"]) + # Extra steps don't mark as invalid + assert result["is_valid"] is True diff --git a/tests/test_upload_to_dropbox.py b/tests/test_upload_to_dropbox.py new file mode 100644 index 00000000..da873c9b --- /dev/null +++ b/tests/test_upload_to_dropbox.py @@ -0,0 +1,333 @@ +""" +Tests for app/tasks/upload_to_dropbox.py module. + +Covers _validate_dropbox_settings, get_dropbox_access_token, get_dropbox_client, +and upload_to_dropbox Celery task. +""" + +import os +from unittest.mock import MagicMock, Mock, patch + +import pytest +from dropbox.exceptions import ApiError, AuthError + + +@pytest.mark.unit +class TestValidateDropboxSettings: + """Tests for _validate_dropbox_settings function.""" + + def test_all_settings_present(self): + """Test validation passes when all settings are present.""" + from app.tasks.upload_to_dropbox import _validate_dropbox_settings + + with patch("app.tasks.upload_to_dropbox.settings") as mock_settings: + mock_settings.dropbox_refresh_token = "token" + mock_settings.dropbox_app_key = "key" + mock_settings.dropbox_app_secret = "secret" + + assert _validate_dropbox_settings() is True + + def test_missing_refresh_token(self): + """Test validation fails when refresh token is missing.""" + from app.tasks.upload_to_dropbox import _validate_dropbox_settings + + with patch("app.tasks.upload_to_dropbox.settings") as mock_settings: + mock_settings.dropbox_refresh_token = "" + mock_settings.dropbox_app_key = "key" + mock_settings.dropbox_app_secret = "secret" + + assert _validate_dropbox_settings() is False + + def test_missing_app_key(self): + """Test validation fails when app key is missing.""" + from app.tasks.upload_to_dropbox import _validate_dropbox_settings + + with patch("app.tasks.upload_to_dropbox.settings") as mock_settings: + mock_settings.dropbox_refresh_token = "token" + mock_settings.dropbox_app_key = "" + mock_settings.dropbox_app_secret = "secret" + + assert _validate_dropbox_settings() is False + + def test_missing_app_secret(self): + """Test validation fails when app secret is missing.""" + from app.tasks.upload_to_dropbox import _validate_dropbox_settings + + with patch("app.tasks.upload_to_dropbox.settings") as mock_settings: + mock_settings.dropbox_refresh_token = "token" + mock_settings.dropbox_app_key = "key" + mock_settings.dropbox_app_secret = "" + + assert _validate_dropbox_settings() is False + + def test_missing_all_settings(self): + """Test validation fails when all settings are missing.""" + from app.tasks.upload_to_dropbox import _validate_dropbox_settings + + with patch("app.tasks.upload_to_dropbox.settings") as mock_settings: + mock_settings.dropbox_refresh_token = None + mock_settings.dropbox_app_key = None + mock_settings.dropbox_app_secret = None + + assert _validate_dropbox_settings() is False + + +@pytest.mark.unit +class TestGetDropboxAccessToken: + """Tests for get_dropbox_access_token function.""" + + @patch("app.tasks.upload_to_dropbox.requests.post") + @patch("app.tasks.upload_to_dropbox.settings") + def test_successful_refresh(self, mock_settings, mock_post): + """Test successful token refresh.""" + from app.tasks.upload_to_dropbox import get_dropbox_access_token + + mock_settings.dropbox_refresh_token = "refresh-token" + mock_settings.dropbox_app_key = "app-key" + mock_settings.dropbox_app_secret = "app-secret" + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = {"access_token": "new-access-token"} + mock_post.return_value = mock_response + + token = get_dropbox_access_token() + assert token == "new-access-token" + + @patch("app.tasks.upload_to_dropbox.requests.post") + @patch("app.tasks.upload_to_dropbox.settings") + def test_refresh_failure_raises(self, mock_settings, mock_post): + """Test that failed token refresh raises exception.""" + from app.tasks.upload_to_dropbox import get_dropbox_access_token + + mock_settings.dropbox_refresh_token = "refresh-token" + mock_settings.dropbox_app_key = "app-key" + mock_settings.dropbox_app_secret = "app-secret" + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.status_code = 400 + mock_response.text = "invalid_grant" + mock_post.return_value = mock_response + + with pytest.raises(Exception, match="Failed to refresh"): + get_dropbox_access_token() + + def test_returns_none_when_settings_missing(self): + """Test returns None when settings are not configured.""" + from app.tasks.upload_to_dropbox import get_dropbox_access_token + + with patch("app.tasks.upload_to_dropbox.settings") as mock_settings: + mock_settings.dropbox_refresh_token = None + mock_settings.dropbox_app_key = None + mock_settings.dropbox_app_secret = None + + result = get_dropbox_access_token() + assert result is None + + +@pytest.mark.unit +class TestGetDropboxClient: + """Tests for get_dropbox_client function.""" + + @patch("app.tasks.upload_to_dropbox.dropbox.Dropbox") + @patch("app.tasks.upload_to_dropbox.settings") + def test_successful_client_creation(self, mock_settings, mock_dropbox): + """Test successful Dropbox client creation.""" + from app.tasks.upload_to_dropbox import get_dropbox_client + + mock_settings.dropbox_app_key = "app-key" + mock_settings.dropbox_app_secret = "app-secret" + mock_settings.dropbox_refresh_token = "refresh-token" + + mock_instance = Mock() + mock_dropbox.return_value = mock_instance + + client = get_dropbox_client() + assert client == mock_instance + mock_instance.users_get_current_account.assert_called_once() + + @patch("app.tasks.upload_to_dropbox.settings") + def test_missing_app_key_raises(self, mock_settings): + """Test that missing app key raises ValueError.""" + from app.tasks.upload_to_dropbox import get_dropbox_client + + mock_settings.dropbox_app_key = "" + mock_settings.dropbox_app_secret = "secret" + mock_settings.dropbox_refresh_token = "token" + + with pytest.raises(ValueError, match="app key or app secret"): + get_dropbox_client() + + @patch("app.tasks.upload_to_dropbox.settings") + def test_missing_refresh_token_raises(self, mock_settings): + """Test that missing refresh token raises ValueError.""" + from app.tasks.upload_to_dropbox import get_dropbox_client + + mock_settings.dropbox_app_key = "key" + mock_settings.dropbox_app_secret = "secret" + mock_settings.dropbox_refresh_token = "" + + with pytest.raises(ValueError, match="refresh token"): + get_dropbox_client() + + @patch("app.tasks.upload_to_dropbox.dropbox.Dropbox") + @patch("app.tasks.upload_to_dropbox.settings") + def test_auth_error_propagated(self, mock_settings, mock_dropbox): + """Test that AuthError is propagated.""" + from app.tasks.upload_to_dropbox import get_dropbox_client + + mock_settings.dropbox_app_key = "key" + mock_settings.dropbox_app_secret = "secret" + mock_settings.dropbox_refresh_token = "token" + + mock_instance = Mock() + mock_instance.users_get_current_account.side_effect = AuthError("req-id", "Invalid token") + mock_dropbox.return_value = mock_instance + + with pytest.raises(AuthError): + get_dropbox_client() + + +@pytest.mark.unit +class TestUploadToDropbox: + """Tests for upload_to_dropbox Celery task.""" + + @patch("app.tasks.upload_to_dropbox.log_task_progress") + def test_file_not_found(self, mock_log): + """Test that missing file raises FileNotFoundError.""" + from app.tasks.upload_to_dropbox import upload_to_dropbox + + mock_self = MagicMock() + mock_self.request.id = "test-task" + + with pytest.raises(FileNotFoundError): + upload_to_dropbox.__wrapped__(mock_self, "/nonexistent/file.pdf", file_id=1) + + @patch("app.tasks.upload_to_dropbox.log_task_progress") + @patch("app.tasks.upload_to_dropbox.settings") + def test_skipped_when_not_configured(self, mock_settings, mock_log, tmp_path): + """Test upload skipped when Dropbox not configured.""" + from app.tasks.upload_to_dropbox import upload_to_dropbox + + mock_settings.dropbox_app_key = "" + mock_settings.dropbox_app_secret = "" + mock_settings.dropbox_refresh_token = "" + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + result = upload_to_dropbox.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Skipped" + + @patch("app.tasks.upload_to_dropbox.get_unique_filename") + @patch("app.tasks.upload_to_dropbox.extract_remote_path") + @patch("app.tasks.upload_to_dropbox.get_dropbox_client") + @patch("app.tasks.upload_to_dropbox.log_task_progress") + @patch("app.tasks.upload_to_dropbox.settings") + def test_successful_small_file_upload( + self, mock_settings, mock_log, mock_client, mock_extract, mock_unique, tmp_path + ): + """Test successful upload of a small file.""" + from app.tasks.upload_to_dropbox import upload_to_dropbox + + mock_settings.dropbox_app_key = "key" + mock_settings.dropbox_app_secret = "secret" + mock_settings.dropbox_refresh_token = "token" + mock_settings.dropbox_folder = "/uploads" + mock_settings.workdir = str(tmp_path) + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"small file content") + + mock_dbx = Mock() + mock_client.return_value = mock_dbx + mock_extract.return_value = "uploads/test.pdf" + mock_unique.return_value = "/uploads/test.pdf" + + result = upload_to_dropbox.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Completed" + assert result["file_path"] == str(test_file) + mock_dbx.files_upload.assert_called_once() + + @patch("app.tasks.upload_to_dropbox.get_unique_filename") + @patch("app.tasks.upload_to_dropbox.extract_remote_path") + @patch("app.tasks.upload_to_dropbox.get_dropbox_client") + @patch("app.tasks.upload_to_dropbox.log_task_progress") + @patch("app.tasks.upload_to_dropbox.settings") + def test_large_file_chunked_upload( + self, mock_settings, mock_log, mock_client, mock_extract, mock_unique, tmp_path + ): + """Test chunked upload for large files (>10MB).""" + from app.tasks.upload_to_dropbox import upload_to_dropbox + + mock_settings.dropbox_app_key = "key" + mock_settings.dropbox_app_secret = "secret" + mock_settings.dropbox_refresh_token = "token" + mock_settings.dropbox_folder = "/uploads" + mock_settings.workdir = str(tmp_path) + + # Create a file larger than 10MB threshold + test_file = tmp_path / "large.pdf" + test_file.write_bytes(b"x" * (11 * 1024 * 1024)) + + mock_dbx = Mock() + mock_client.return_value = mock_dbx + mock_extract.return_value = "uploads/large.pdf" + mock_unique.return_value = "/uploads/large.pdf" + + mock_session = Mock() + mock_session.session_id = "session-123" + mock_dbx.files_upload_session_start.return_value = mock_session + + result = upload_to_dropbox.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Completed" + mock_dbx.files_upload_session_start.assert_called_once() + + @patch("app.tasks.upload_to_dropbox.get_dropbox_client") + @patch("app.tasks.upload_to_dropbox.log_task_progress") + @patch("app.tasks.upload_to_dropbox.settings") + def test_auth_error_handling(self, mock_settings, mock_log, mock_client, tmp_path): + """Test AuthError during upload is caught and re-raised.""" + from app.tasks.upload_to_dropbox import upload_to_dropbox + + mock_settings.dropbox_app_key = "key" + mock_settings.dropbox_app_secret = "secret" + mock_settings.dropbox_refresh_token = "token" + mock_settings.dropbox_folder = "/uploads" + mock_settings.workdir = str(tmp_path) + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + mock_client.side_effect = AuthError("req-id", "Invalid token") + + with pytest.raises(Exception, match="Authentication failed"): + upload_to_dropbox.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + @patch("app.tasks.upload_to_dropbox.get_dropbox_client") + @patch("app.tasks.upload_to_dropbox.log_task_progress") + @patch("app.tasks.upload_to_dropbox.settings") + def test_api_error_handling(self, mock_settings, mock_log, mock_client, tmp_path): + """Test ApiError during upload is caught and re-raised.""" + from app.tasks.upload_to_dropbox import upload_to_dropbox + + mock_settings.dropbox_app_key = "key" + mock_settings.dropbox_app_secret = "secret" + mock_settings.dropbox_refresh_token = "token" + mock_settings.dropbox_folder = "/uploads" + mock_settings.workdir = str(tmp_path) + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + mock_error = Mock() + mock_error.is_path.return_value = False + mock_client.side_effect = ApiError("req-id", mock_error, "user msg", "header") + + with pytest.raises(Exception, match="Failed to upload"): + upload_to_dropbox.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() diff --git a/tests/test_upload_to_nextcloud.py b/tests/test_upload_to_nextcloud.py new file mode 100644 index 00000000..050e94c4 --- /dev/null +++ b/tests/test_upload_to_nextcloud.py @@ -0,0 +1,272 @@ +""" +Tests for app/tasks/upload_to_nextcloud.py module. + +Covers the upload_to_nextcloud Celery task including configuration validation, +WebDAV upload, directory creation, and error handling. +""" + +import os +from unittest.mock import MagicMock, Mock, call, patch + +import pytest + + +@pytest.mark.unit +class TestUploadToNextcloud: + """Tests for upload_to_nextcloud Celery task.""" + + @patch("app.tasks.upload_to_nextcloud.log_task_progress") + def test_file_not_found(self, mock_log): + """Test that missing file raises FileNotFoundError.""" + from app.tasks.upload_to_nextcloud import upload_to_nextcloud + + mock_self = MagicMock() + mock_self.request.id = "test-task" + + with pytest.raises(FileNotFoundError): + upload_to_nextcloud.__wrapped__(mock_self, "/nonexistent/file.pdf", file_id=1) + + @patch("app.tasks.upload_to_nextcloud.log_task_progress") + @patch("app.tasks.upload_to_nextcloud.settings") + def test_skipped_when_not_configured(self, mock_settings, mock_log, tmp_path): + """Test upload skipped when Nextcloud URL not configured.""" + from app.tasks.upload_to_nextcloud import upload_to_nextcloud + + mock_settings.nextcloud_upload_url = None + mock_settings.nextcloud_username = None + mock_settings.nextcloud_password = None + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + result = upload_to_nextcloud.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Skipped" + + @patch("app.tasks.upload_to_nextcloud.log_task_progress") + @patch("app.tasks.upload_to_nextcloud.settings") + def test_skipped_when_username_missing(self, mock_settings, mock_log, tmp_path): + """Test upload skipped when username is missing.""" + from app.tasks.upload_to_nextcloud import upload_to_nextcloud + + mock_settings.nextcloud_upload_url = "https://nextcloud.example.com/remote.php/dav" + mock_settings.nextcloud_username = None + mock_settings.nextcloud_password = "password" # noqa: S105 + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + result = upload_to_nextcloud.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Skipped" + + @patch("app.tasks.upload_to_nextcloud.get_unique_filename") + @patch("app.tasks.upload_to_nextcloud.extract_remote_path") + @patch("app.tasks.upload_to_nextcloud.requests") + @patch("app.tasks.upload_to_nextcloud.log_task_progress") + @patch("app.tasks.upload_to_nextcloud.settings") + def test_successful_upload_201(self, mock_settings, mock_log, mock_requests, mock_extract, mock_unique, tmp_path): + """Test successful file upload with 201 response.""" + from app.tasks.upload_to_nextcloud import upload_to_nextcloud + + mock_settings.nextcloud_upload_url = "https://nextcloud.example.com/remote.php/dav/" + mock_settings.nextcloud_username = "user" + mock_settings.nextcloud_password = "pass" # noqa: S105 + mock_settings.nextcloud_folder = "" + mock_settings.workdir = str(tmp_path) + mock_settings.http_request_timeout = 30 + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + mock_extract.return_value = "test.pdf" + mock_unique.return_value = "test.pdf" + + # Mock the PUT response for file upload + mock_put_response = Mock() + mock_put_response.status_code = 201 + mock_requests.put.return_value = mock_put_response + + # Mock PROPFIND for existence check + mock_propfind_response = Mock() + mock_propfind_response.text = "" + mock_requests.request.return_value = mock_propfind_response + + result = upload_to_nextcloud.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Completed" + assert result["response_code"] == 201 + + @patch("app.tasks.upload_to_nextcloud.get_unique_filename") + @patch("app.tasks.upload_to_nextcloud.extract_remote_path") + @patch("app.tasks.upload_to_nextcloud.requests") + @patch("app.tasks.upload_to_nextcloud.log_task_progress") + @patch("app.tasks.upload_to_nextcloud.settings") + def test_successful_upload_204(self, mock_settings, mock_log, mock_requests, mock_extract, mock_unique, tmp_path): + """Test successful file upload with 204 (No Content / overwrite) response.""" + from app.tasks.upload_to_nextcloud import upload_to_nextcloud + + mock_settings.nextcloud_upload_url = "https://nextcloud.example.com/remote.php/dav/" + mock_settings.nextcloud_username = "user" + mock_settings.nextcloud_password = "pass" # noqa: S105 + mock_settings.nextcloud_folder = "" + mock_settings.workdir = str(tmp_path) + mock_settings.http_request_timeout = 30 + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + mock_extract.return_value = "test.pdf" + mock_unique.return_value = "test.pdf" + + mock_put_response = Mock() + mock_put_response.status_code = 204 + mock_requests.put.return_value = mock_put_response + + mock_propfind_response = Mock() + mock_propfind_response.text = "" + mock_requests.request.return_value = mock_propfind_response + + result = upload_to_nextcloud.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Completed" + assert result["response_code"] == 204 + + @patch("app.tasks.upload_to_nextcloud.get_unique_filename") + @patch("app.tasks.upload_to_nextcloud.extract_remote_path") + @patch("app.tasks.upload_to_nextcloud.requests") + @patch("app.tasks.upload_to_nextcloud.log_task_progress") + @patch("app.tasks.upload_to_nextcloud.settings") + def test_upload_failure_status_code( + self, mock_settings, mock_log, mock_requests, mock_extract, mock_unique, tmp_path + ): + """Test upload failure with non-success status code.""" + from app.tasks.upload_to_nextcloud import upload_to_nextcloud + + mock_settings.nextcloud_upload_url = "https://nextcloud.example.com/remote.php/dav/" + mock_settings.nextcloud_username = "user" + mock_settings.nextcloud_password = "pass" # noqa: S105 + mock_settings.nextcloud_folder = "" + mock_settings.workdir = str(tmp_path) + mock_settings.http_request_timeout = 30 + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + mock_extract.return_value = "test.pdf" + mock_unique.return_value = "test.pdf" + + mock_put_response = Mock() + mock_put_response.status_code = 500 + mock_put_response.text = "Internal Server Error" + mock_requests.put.return_value = mock_put_response + + mock_propfind_response = Mock() + mock_propfind_response.text = "" + mock_requests.request.return_value = mock_propfind_response + + with pytest.raises(Exception, match="Failed to upload"): + upload_to_nextcloud.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + @patch("app.tasks.upload_to_nextcloud.get_unique_filename") + @patch("app.tasks.upload_to_nextcloud.extract_remote_path") + @patch("app.tasks.upload_to_nextcloud.requests") + @patch("app.tasks.upload_to_nextcloud.log_task_progress") + @patch("app.tasks.upload_to_nextcloud.settings") + def test_creates_parent_directories( + self, mock_settings, mock_log, mock_requests, mock_extract, mock_unique, tmp_path + ): + """Test that parent directories are created via MKCOL.""" + from app.tasks.upload_to_nextcloud import upload_to_nextcloud + + mock_settings.nextcloud_upload_url = "https://nextcloud.example.com/remote.php/dav/" + mock_settings.nextcloud_username = "user" + mock_settings.nextcloud_password = "pass" # noqa: S105 + mock_settings.nextcloud_folder = "documents" + mock_settings.workdir = str(tmp_path) + mock_settings.http_request_timeout = 30 + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + mock_extract.return_value = "documents/subfolder/test.pdf" + mock_unique.return_value = "documents/subfolder/test.pdf" + + mock_put_response = Mock() + mock_put_response.status_code = 201 + mock_requests.put.return_value = mock_put_response + + mock_request_response = Mock() + mock_request_response.text = "" + mock_requests.request.return_value = mock_request_response + + result = upload_to_nextcloud.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Completed" + # Verify MKCOL calls were made for parent directories + mkcol_calls = [c for c in mock_requests.request.call_args_list if c[0][0] == "MKCOL"] + assert len(mkcol_calls) >= 1 + + @patch("app.tasks.upload_to_nextcloud.get_unique_filename") + @patch("app.tasks.upload_to_nextcloud.extract_remote_path") + @patch("app.tasks.upload_to_nextcloud.requests") + @patch("app.tasks.upload_to_nextcloud.log_task_progress") + @patch("app.tasks.upload_to_nextcloud.settings") + def test_connection_error(self, mock_settings, mock_log, mock_requests, mock_extract, mock_unique, tmp_path): + """Test handling of connection errors during upload.""" + from app.tasks.upload_to_nextcloud import upload_to_nextcloud + + mock_settings.nextcloud_upload_url = "https://nextcloud.example.com/remote.php/dav/" + mock_settings.nextcloud_username = "user" + mock_settings.nextcloud_password = "pass" # noqa: S105 + mock_settings.nextcloud_folder = "" + mock_settings.workdir = str(tmp_path) + mock_settings.http_request_timeout = 30 + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + mock_extract.return_value = "test.pdf" + mock_unique.return_value = "test.pdf" + + mock_requests.put.side_effect = Exception("Connection refused") + mock_requests.request.return_value = Mock(text="") + + with pytest.raises(Exception, match="Failed to upload"): + upload_to_nextcloud.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + @patch("app.tasks.upload_to_nextcloud.get_unique_filename") + @patch("app.tasks.upload_to_nextcloud.extract_remote_path") + @patch("app.tasks.upload_to_nextcloud.requests") + @patch("app.tasks.upload_to_nextcloud.log_task_progress") + @patch("app.tasks.upload_to_nextcloud.settings") + def test_url_trailing_slash_normalization( + self, mock_settings, mock_log, mock_requests, mock_extract, mock_unique, tmp_path + ): + """Test that URLs without trailing slashes are handled.""" + from app.tasks.upload_to_nextcloud import upload_to_nextcloud + + mock_settings.nextcloud_upload_url = "https://nextcloud.example.com/remote.php/dav" + mock_settings.nextcloud_username = "user" + mock_settings.nextcloud_password = "pass" # noqa: S105 + mock_settings.nextcloud_folder = "" + mock_settings.workdir = str(tmp_path) + mock_settings.http_request_timeout = 30 + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + mock_extract.return_value = "test.pdf" + mock_unique.return_value = "test.pdf" + + mock_put_response = Mock() + mock_put_response.status_code = 201 + mock_requests.put.return_value = mock_put_response + + mock_request_response = Mock() + mock_request_response.text = "" + mock_requests.request.return_value = mock_request_response + + result = upload_to_nextcloud.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Completed" diff --git a/tests/test_upload_to_onedrive.py b/tests/test_upload_to_onedrive.py new file mode 100644 index 00000000..b67995d6 --- /dev/null +++ b/tests/test_upload_to_onedrive.py @@ -0,0 +1,404 @@ +""" +Tests for app/tasks/upload_to_onedrive.py module. + +Covers get_onedrive_token, create_upload_session, upload_large_file, +and upload_to_onedrive Celery task. +""" + +import os +from unittest.mock import MagicMock, Mock, patch + +import pytest + + +@pytest.mark.unit +class TestGetOnedriveToken: + """Tests for get_onedrive_token function.""" + + @patch("app.tasks.upload_to_onedrive.msal.ConfidentialClientApplication") + @patch("app.tasks.upload_to_onedrive.settings") + def test_refresh_token_flow(self, mock_settings, mock_msal): + """Test token acquisition using refresh token.""" + from app.tasks.upload_to_onedrive import get_onedrive_token + + mock_settings.onedrive_client_id = "client-id" + mock_settings.onedrive_client_secret = "client-secret" + mock_settings.onedrive_refresh_token = "refresh-token" + mock_settings.onedrive_tenant_id = "common" + + mock_app = Mock() + mock_app.acquire_token_by_refresh_token.return_value = { + "access_token": "new-access-token", + } + mock_msal.return_value = mock_app + + token = get_onedrive_token() + assert token == "new-access-token" + + @patch("app.tasks.upload_to_onedrive.msal.ConfidentialClientApplication") + @patch("app.tasks.upload_to_onedrive.settings") + def test_refresh_token_updates_new_token(self, mock_settings, mock_msal): + """Test that a new refresh token updates settings.""" + from app.tasks.upload_to_onedrive import get_onedrive_token + + mock_settings.onedrive_client_id = "client-id" + mock_settings.onedrive_client_secret = "client-secret" + mock_settings.onedrive_refresh_token = "old-refresh-token" + mock_settings.onedrive_tenant_id = "common" + + mock_app = Mock() + mock_app.acquire_token_by_refresh_token.return_value = { + "access_token": "access-token", + "refresh_token": "new-refresh-token", + } + mock_msal.return_value = mock_app + + get_onedrive_token() + assert mock_settings.onedrive_refresh_token == "new-refresh-token" + + @patch("app.tasks.upload_to_onedrive.msal.ConfidentialClientApplication") + @patch("app.tasks.upload_to_onedrive.settings") + def test_refresh_token_failure(self, mock_settings, mock_msal): + """Test error handling when refresh token fails.""" + from app.tasks.upload_to_onedrive import get_onedrive_token + + mock_settings.onedrive_client_id = "client-id" + mock_settings.onedrive_client_secret = "client-secret" + mock_settings.onedrive_refresh_token = "expired-token" + mock_settings.onedrive_tenant_id = "common" + + mock_app = Mock() + mock_app.acquire_token_by_refresh_token.return_value = { + "error": "invalid_grant", + "error_description": "Token expired", + } + mock_msal.return_value = mock_app + + with pytest.raises(ValueError, match="Failed to get access token"): + get_onedrive_token() + + @patch("app.tasks.upload_to_onedrive.msal.ConfidentialClientApplication") + @patch("app.tasks.upload_to_onedrive.settings") + def test_client_credentials_flow(self, mock_settings, mock_msal): + """Test token acquisition using client credentials (org accounts).""" + from app.tasks.upload_to_onedrive import get_onedrive_token + + mock_settings.onedrive_client_id = "client-id" + mock_settings.onedrive_client_secret = "client-secret" + mock_settings.onedrive_refresh_token = "" + mock_settings.onedrive_tenant_id = "org-tenant-id" + + mock_app = Mock() + mock_app.acquire_token_for_client.return_value = { + "access_token": "client-cred-token", + } + mock_msal.return_value = mock_app + + token = get_onedrive_token() + assert token == "client-cred-token" + + @patch("app.tasks.upload_to_onedrive.msal.ConfidentialClientApplication") + @patch("app.tasks.upload_to_onedrive.settings") + def test_client_credentials_failure(self, mock_settings, mock_msal): + """Test error handling when client credentials flow fails.""" + from app.tasks.upload_to_onedrive import get_onedrive_token + + mock_settings.onedrive_client_id = "client-id" + mock_settings.onedrive_client_secret = "client-secret" + mock_settings.onedrive_refresh_token = "" + mock_settings.onedrive_tenant_id = "org-tenant-id" + + mock_app = Mock() + mock_app.acquire_token_for_client.return_value = { + "error": "unauthorized_client", + "error_description": "Not authorized", + } + mock_msal.return_value = mock_app + + with pytest.raises(ValueError, match="Failed to get access token"): + get_onedrive_token() + + @patch("app.tasks.upload_to_onedrive.settings") + def test_missing_client_id(self, mock_settings): + """Test error when client ID is missing.""" + from app.tasks.upload_to_onedrive import get_onedrive_token + + mock_settings.onedrive_client_id = "" + mock_settings.onedrive_client_secret = "secret" + + with pytest.raises(ValueError, match="client ID and client secret"): + get_onedrive_token() + + @patch("app.tasks.upload_to_onedrive.settings") + def test_no_refresh_token_personal_account(self, mock_settings): + """Test error for personal account without refresh token.""" + from app.tasks.upload_to_onedrive import get_onedrive_token + + mock_settings.onedrive_client_id = "client-id" + mock_settings.onedrive_client_secret = "secret" + mock_settings.onedrive_refresh_token = "" + mock_settings.onedrive_tenant_id = "common" + + with pytest.raises(ValueError, match="ONEDRIVE_REFRESH_TOKEN must be configured"): + get_onedrive_token() + + @patch("app.tasks.upload_to_onedrive.settings") + def test_no_refresh_token_no_tenant(self, mock_settings): + """Test error when no refresh token and no specific tenant.""" + from app.tasks.upload_to_onedrive import get_onedrive_token + + mock_settings.onedrive_client_id = "client-id" + mock_settings.onedrive_client_secret = "secret" + mock_settings.onedrive_refresh_token = "" + mock_settings.onedrive_tenant_id = "" + + with pytest.raises(ValueError, match="ONEDRIVE_REFRESH_TOKEN must be configured"): + get_onedrive_token() + + +@pytest.mark.unit +class TestCreateUploadSession: + """Tests for create_upload_session function.""" + + @patch("app.tasks.upload_to_onedrive.requests.post") + @patch("app.tasks.upload_to_onedrive.settings") + def test_successful_session_creation(self, mock_settings, mock_post): + """Test successful upload session creation.""" + from app.tasks.upload_to_onedrive import create_upload_session + + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = {"uploadUrl": "https://upload.url/session123"} + mock_post.return_value = mock_response + + url = create_upload_session("test.pdf", "Documents/Uploads", "access-token") + + assert url == "https://upload.url/session123" + + @patch("app.tasks.upload_to_onedrive.requests.post") + @patch("app.tasks.upload_to_onedrive.settings") + def test_session_creation_without_folder(self, mock_settings, mock_post): + """Test upload session creation without folder path.""" + from app.tasks.upload_to_onedrive import create_upload_session + + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = {"uploadUrl": "https://upload.url/session456"} + mock_post.return_value = mock_response + + url = create_upload_session("test.pdf", None, "access-token") + + assert url == "https://upload.url/session456" + + @patch("app.tasks.upload_to_onedrive.requests.post") + @patch("app.tasks.upload_to_onedrive.settings") + def test_session_creation_failure(self, mock_settings, mock_post): + """Test error handling when session creation fails.""" + from app.tasks.upload_to_onedrive import create_upload_session + + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.status_code = 403 + mock_response.text = "Access denied" + mock_post.return_value = mock_response + + with pytest.raises(Exception, match="Failed to create upload session"): + create_upload_session("test.pdf", "Documents", "access-token") + + @patch("app.tasks.upload_to_onedrive.requests.post") + @patch("app.tasks.upload_to_onedrive.settings") + def test_url_encoding_of_special_characters(self, mock_settings, mock_post): + """Test that special characters in folder path are URL-encoded.""" + from app.tasks.upload_to_onedrive import create_upload_session + + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = {"uploadUrl": "https://upload.url/session"} + mock_post.return_value = mock_response + + create_upload_session("file with spaces.pdf", "My Documents/Uploads", "access-token") + + # Verify the URL was constructed with encoded components + call_url = mock_post.call_args[0][0] + assert "My%20Documents" in call_url + assert "file%20with%20spaces.pdf" in call_url + + +@pytest.mark.unit +class TestUploadLargeFile: + """Tests for upload_large_file function.""" + + @patch("app.tasks.upload_to_onedrive.requests.put") + @patch("app.tasks.upload_to_onedrive.settings") + def test_small_single_chunk_upload(self, mock_settings, mock_put, tmp_path): + """Test uploading a file that fits in a single chunk.""" + from app.tasks.upload_to_onedrive import upload_large_file + + mock_settings.http_request_timeout = 30 + + test_file = tmp_path / "small.pdf" + test_file.write_bytes(b"small content") + + mock_response = Mock() + mock_response.status_code = 201 + mock_response.json.return_value = {"id": "file123", "name": "small.pdf"} + mock_put.return_value = mock_response + + result = upload_large_file(str(test_file), "https://upload.url/session") + + assert result["id"] == "file123" + + @patch("app.tasks.upload_to_onedrive.time.sleep") + @patch("app.tasks.upload_to_onedrive.requests.put") + @patch("app.tasks.upload_to_onedrive.settings") + def test_chunk_upload_retry_on_failure(self, mock_settings, mock_put, mock_sleep, tmp_path): + """Test retry logic when a chunk upload fails.""" + from app.tasks.upload_to_onedrive import upload_large_file + + mock_settings.http_request_timeout = 30 + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + # First attempt fails, second succeeds + mock_fail = Mock() + mock_fail.status_code = 500 + + mock_success = Mock() + mock_success.status_code = 201 + mock_success.json.return_value = {"id": "file123"} + + mock_put.side_effect = [mock_fail, mock_success] + + result = upload_large_file(str(test_file), "https://upload.url/session") + + assert result["id"] == "file123" + + @patch("app.tasks.upload_to_onedrive.time.sleep") + @patch("app.tasks.upload_to_onedrive.requests.put") + @patch("app.tasks.upload_to_onedrive.settings") + def test_chunk_upload_retry_on_exception(self, mock_settings, mock_put, mock_sleep, tmp_path): + """Test retry logic when an exception occurs during upload.""" + from app.tasks.upload_to_onedrive import upload_large_file + + mock_settings.http_request_timeout = 30 + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + mock_success = Mock() + mock_success.status_code = 201 + mock_success.json.return_value = {"id": "file123"} + + mock_put.side_effect = [Exception("Network error"), mock_success] + + result = upload_large_file(str(test_file), "https://upload.url/session") + + assert result["id"] == "file123" + + @patch("app.tasks.upload_to_onedrive.time.sleep") + @patch("app.tasks.upload_to_onedrive.requests.put") + @patch("app.tasks.upload_to_onedrive.settings") + def test_all_retries_exhausted(self, mock_settings, mock_put, mock_sleep, tmp_path): + """Test that exhausting all retries raises an exception.""" + from app.tasks.upload_to_onedrive import upload_large_file + + mock_settings.http_request_timeout = 30 + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + mock_fail = Mock() + mock_fail.status_code = 500 + mock_fail.text = "Server Error" + mock_put.return_value = mock_fail + + with pytest.raises(Exception, match="Failed to upload chunk"): + upload_large_file(str(test_file), "https://upload.url/session") + + +@pytest.mark.unit +class TestUploadToOnedrive: + """Tests for upload_to_onedrive Celery task.""" + + @patch("app.tasks.upload_to_onedrive.log_task_progress") + def test_file_not_found(self, mock_log): + """Test that missing file raises FileNotFoundError.""" + from app.tasks.upload_to_onedrive import upload_to_onedrive + + mock_self = MagicMock() + mock_self.request.id = "test-task" + + with pytest.raises(FileNotFoundError): + upload_to_onedrive.__wrapped__(mock_self, "/nonexistent/file.pdf", file_id=1) + + @patch("app.tasks.upload_to_onedrive.log_task_progress") + @patch("app.tasks.upload_to_onedrive.settings") + def test_missing_client_id(self, mock_settings, mock_log, tmp_path): + """Test error when OneDrive client ID is not configured.""" + from app.tasks.upload_to_onedrive import upload_to_onedrive + + mock_settings.onedrive_client_id = "" + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + mock_self = MagicMock() + mock_self.request.id = "test-task" + + with pytest.raises(ValueError, match="client ID is not configured"): + upload_to_onedrive.__wrapped__(mock_self, str(test_file), file_id=1) + + @patch("app.tasks.upload_to_onedrive.upload_large_file") + @patch("app.tasks.upload_to_onedrive.create_upload_session") + @patch("app.tasks.upload_to_onedrive.get_onedrive_token") + @patch("app.tasks.upload_to_onedrive.log_task_progress") + @patch("app.tasks.upload_to_onedrive.settings") + def test_successful_upload(self, mock_settings, mock_log, mock_token, mock_session, mock_upload, tmp_path): + """Test successful OneDrive upload.""" + from app.tasks.upload_to_onedrive import upload_to_onedrive + + mock_settings.onedrive_client_id = "client-id" + mock_settings.onedrive_client_secret = "secret" + mock_settings.onedrive_refresh_token = "token" + mock_settings.onedrive_folder_path = "Documents" + mock_settings.onedrive_tenant_id = "common" + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + mock_token.return_value = "access-token" + mock_session.return_value = "https://upload.url/session" + mock_upload.return_value = {"webUrl": "https://onedrive.live.com/test.pdf"} + + result = upload_to_onedrive.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Completed" + assert "Documents" in result["onedrive_path"] + assert result["web_url"] == "https://onedrive.live.com/test.pdf" + + @patch("app.tasks.upload_to_onedrive.get_onedrive_token") + @patch("app.tasks.upload_to_onedrive.log_task_progress") + @patch("app.tasks.upload_to_onedrive.settings") + def test_upload_exception_handling(self, mock_settings, mock_log, mock_token, tmp_path): + """Test that upload errors are properly handled.""" + from app.tasks.upload_to_onedrive import upload_to_onedrive + + mock_settings.onedrive_client_id = "client-id" + mock_settings.onedrive_client_secret = "secret" + mock_settings.onedrive_folder_path = "Documents" + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + mock_token.side_effect = ValueError("Token error") + + with pytest.raises(Exception, match="Failed to upload"): + upload_to_onedrive.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() diff --git a/tests/test_upload_to_paperless.py b/tests/test_upload_to_paperless.py new file mode 100644 index 00000000..aa0c532c --- /dev/null +++ b/tests/test_upload_to_paperless.py @@ -0,0 +1,626 @@ +""" +Tests for app/tasks/upload_to_paperless.py module. + +Covers helper functions (normalize_metadata_value, _is_duplicate_error, poll_task_for_document_id, +get_custom_field_id, set_document_custom_fields) and the upload_to_paperless Celery task. +""" + +import json +import os +from unittest.mock import MagicMock, Mock, mock_open, patch + +import pytest +import requests + +from app.tasks.upload_to_paperless import ( + _get_headers, + _is_duplicate_error, + _paperless_api_url, + get_custom_field_id, + normalize_metadata_value, + poll_task_for_document_id, + set_document_custom_fields, + upload_to_paperless, +) + + +@pytest.mark.unit +class TestNormalizeMetadataValue: + """Tests for normalize_metadata_value function.""" + + def test_none_returns_empty(self): + """Test that None value returns empty string.""" + assert normalize_metadata_value(None) == "" + + def test_empty_string_returns_empty(self): + """Test that empty string returns empty string.""" + assert normalize_metadata_value("") == "" + + def test_unknown_placeholder_returns_empty(self): + """Test that 'Unknown' placeholder returns empty string.""" + assert normalize_metadata_value("Unknown") == "" + + def test_normal_string_passes_through(self): + """Test normal string values pass through.""" + assert normalize_metadata_value("John Doe") == "John Doe" + + def test_integer_converted_to_string(self): + """Test integer values are converted to string.""" + assert normalize_metadata_value(42) == "42" + + def test_float_converted_to_string(self): + """Test float values are converted to string.""" + assert normalize_metadata_value(3.14) == "3.14" + + def test_boolean_converted_to_string(self): + """Test boolean values are converted to string.""" + assert normalize_metadata_value(True) == "True" + + +@pytest.mark.unit +class TestIsDuplicateError: + """Tests for _is_duplicate_error function.""" + + def test_empty_message(self): + """Test that empty/None message returns False.""" + assert _is_duplicate_error("") is False + assert _is_duplicate_error(None) is False + + def test_duplicate_message(self): + """Test detection of duplicate document message.""" + assert _is_duplicate_error("Not consuming duplicate document") is True + + def test_duplicate_case_insensitive(self): + """Test case insensitive duplicate detection.""" + assert _is_duplicate_error("DUPLICATE document not consuming") is True + + def test_non_duplicate_message(self): + """Test that non-duplicate messages return False.""" + assert _is_duplicate_error("Processing completed") is False + + def test_partial_match_not_duplicate(self): + """Test that message with only 'duplicate' but not 'not consuming' returns False.""" + assert _is_duplicate_error("Found a duplicate") is False + + +@pytest.mark.unit +class TestGetHeaders: + """Tests for _get_headers function.""" + + def test_returns_auth_header(self): + """Test that headers include authorization token.""" + with patch("app.tasks.upload_to_paperless.settings") as mock_settings: + mock_settings.paperless_ngx_api_token = "test-token" + headers = _get_headers() + assert headers["Authorization"] == "Token test-token" + + +@pytest.mark.unit +class TestPaperlessApiUrl: + """Tests for _paperless_api_url function.""" + + def test_constructs_url(self): + """Test URL construction.""" + with patch("app.tasks.upload_to_paperless.settings") as mock_settings: + mock_settings.paperless_host = "http://paperless:8000" + url = _paperless_api_url("/api/documents/") + assert url == "http://paperless:8000/api/documents/" + + def test_strips_trailing_slash_from_host(self): + """Test trailing slash is removed from host.""" + with patch("app.tasks.upload_to_paperless.settings") as mock_settings: + mock_settings.paperless_host = "http://paperless:8000/" + url = _paperless_api_url("/api/documents/") + assert url == "http://paperless:8000/api/documents/" + + def test_adds_leading_slash_to_path(self): + """Test leading slash is added to path if missing.""" + with patch("app.tasks.upload_to_paperless.settings") as mock_settings: + mock_settings.paperless_host = "http://paperless:8000" + url = _paperless_api_url("api/documents/") + assert url == "http://paperless:8000/api/documents/" + + +@pytest.mark.unit +class TestPollTaskForDocumentId: + """Tests for poll_task_for_document_id function.""" + + @patch("app.tasks.upload_to_paperless.time.sleep") + @patch("app.tasks.upload_to_paperless.requests.get") + @patch("app.tasks.upload_to_paperless.settings") + def test_success_returns_document_id(self, mock_settings, mock_get, mock_sleep): + """Test successful polling returns document ID.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.json.return_value = [{"status": "SUCCESS", "related_document": "42"}] + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + result = poll_task_for_document_id("test-task-id") + assert result == 42 + + @patch("app.tasks.upload_to_paperless.time.sleep") + @patch("app.tasks.upload_to_paperless.requests.get") + @patch("app.tasks.upload_to_paperless.settings") + def test_success_with_paginated_response(self, mock_settings, mock_get, mock_sleep): + """Test polling handles paginated API responses.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.json.return_value = { + "results": [{"status": "SUCCESS", "related_document": "99"}] + } + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + result = poll_task_for_document_id("test-task-id") + assert result == 99 + + @patch("app.tasks.upload_to_paperless.time.sleep") + @patch("app.tasks.upload_to_paperless.requests.get") + @patch("app.tasks.upload_to_paperless.settings") + def test_failure_raises_runtime_error(self, mock_settings, mock_get, mock_sleep): + """Test that task failure raises RuntimeError.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.json.return_value = [{"status": "FAILURE", "result": "Processing error"}] + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + with pytest.raises(RuntimeError, match="failed"): + poll_task_for_document_id("test-task-id") + + @patch("app.tasks.upload_to_paperless.time.sleep") + @patch("app.tasks.upload_to_paperless.requests.get") + @patch("app.tasks.upload_to_paperless.settings") + def test_duplicate_returns_none(self, mock_settings, mock_get, mock_sleep): + """Test that duplicate document failure returns None.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.json.return_value = [ + {"status": "FAILURE", "result": "Not consuming duplicate document"} + ] + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + result = poll_task_for_document_id("test-task-id") + assert result is None + + @patch("app.tasks.upload_to_paperless.POLL_MAX_ATTEMPTS", 2) + @patch("app.tasks.upload_to_paperless.time.sleep") + @patch("app.tasks.upload_to_paperless.requests.get") + @patch("app.tasks.upload_to_paperless.settings") + def test_timeout_raises_error(self, mock_settings, mock_get, mock_sleep): + """Test that timeout raises TimeoutError.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + + # Return empty results each time + mock_response = Mock() + mock_response.json.return_value = [] + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + with pytest.raises(TimeoutError): + poll_task_for_document_id("test-task-id") + + @patch("app.tasks.upload_to_paperless.POLL_MAX_ATTEMPTS", 2) + @patch("app.tasks.upload_to_paperless.time.sleep") + @patch("app.tasks.upload_to_paperless.requests.get") + @patch("app.tasks.upload_to_paperless.settings") + def test_handles_request_exception(self, mock_settings, mock_get, mock_sleep): + """Test that request exceptions are handled with retries.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + + mock_get.side_effect = requests.exceptions.ConnectionError("Connection refused") + + with pytest.raises(TimeoutError): + poll_task_for_document_id("test-task-id") + + @patch("app.tasks.upload_to_paperless.time.sleep") + @patch("app.tasks.upload_to_paperless.requests.get") + @patch("app.tasks.upload_to_paperless.settings") + def test_success_without_document_id_raises(self, mock_settings, mock_get, mock_sleep): + """Test success status without related_document raises RuntimeError.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.json.return_value = [{"status": "SUCCESS", "related_document": None}] + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + with pytest.raises(RuntimeError, match="no doc ID found"): + poll_task_for_document_id("test-task-id") + + +@pytest.mark.unit +class TestGetCustomFieldId: + """Tests for get_custom_field_id function.""" + + @patch("app.tasks.upload_to_paperless.requests.get") + @patch("app.tasks.upload_to_paperless.settings") + def test_finds_field_by_name(self, mock_settings, mock_get): + """Test finding a custom field by name.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.json.return_value = { + "results": [{"name": "sender", "id": 5}, {"name": "date", "id": 6}] + } + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + assert get_custom_field_id("sender") == 5 + + @patch("app.tasks.upload_to_paperless.requests.get") + @patch("app.tasks.upload_to_paperless.settings") + def test_field_not_found_raises_value_error(self, mock_settings, mock_get): + """Test that missing field raises ValueError.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.json.return_value = {"results": []} + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + with pytest.raises(ValueError, match="not found"): + get_custom_field_id("nonexistent") + + @patch("app.tasks.upload_to_paperless.requests.get") + @patch("app.tasks.upload_to_paperless.settings") + def test_handles_non_paginated_response(self, mock_settings, mock_get): + """Test handling of non-paginated API response (list).""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + + mock_response = Mock() + mock_response.json.return_value = [{"name": "sender", "id": 5}] + mock_response.raise_for_status = Mock() + mock_get.return_value = mock_response + + assert get_custom_field_id("sender") == 5 + + @patch("app.tasks.upload_to_paperless.requests.get") + @patch("app.tasks.upload_to_paperless.settings") + def test_request_exception_is_raised(self, mock_settings, mock_get): + """Test that request exceptions are propagated.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + + mock_get.side_effect = requests.exceptions.ConnectionError("Connection refused") + + with pytest.raises(requests.exceptions.ConnectionError): + get_custom_field_id("sender") + + +@pytest.mark.unit +class TestSetDocumentCustomFields: + """Tests for set_document_custom_fields function.""" + + @patch("app.tasks.upload_to_paperless.get_custom_field_id") + @patch("app.tasks.upload_to_paperless.requests.patch") + @patch("app.tasks.upload_to_paperless.settings") + def test_sets_custom_fields(self, mock_settings, mock_patch, mock_field_id): + """Test setting custom fields on a document.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + mock_field_id.return_value = 5 + mock_response = Mock() + mock_response.raise_for_status = Mock() + mock_patch.return_value = mock_response + + set_document_custom_fields(42, {"sender": "John Doe"}, "task-123") + + mock_patch.assert_called_once() + + def test_empty_fields_returns_immediately(self): + """Test that empty custom fields dict returns without API calls.""" + with patch("app.tasks.upload_to_paperless.requests.patch") as mock_patch: + set_document_custom_fields(42, {}, "task-123") + mock_patch.assert_not_called() + + @patch("app.tasks.upload_to_paperless.get_custom_field_id") + @patch("app.tasks.upload_to_paperless.settings") + def test_skips_unknown_values(self, mock_settings, mock_field_id): + """Test that 'Unknown' values are skipped.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + + with patch("app.tasks.upload_to_paperless.requests.patch") as mock_patch: + set_document_custom_fields(42, {"sender": "Unknown"}, "task-123") + mock_patch.assert_not_called() + + @patch("app.tasks.upload_to_paperless.get_custom_field_id") + @patch("app.tasks.upload_to_paperless.requests.patch") + @patch("app.tasks.upload_to_paperless.settings") + def test_skips_field_not_found(self, mock_settings, mock_patch, mock_field_id): + """Test that fields not found in Paperless are skipped.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + mock_field_id.side_effect = ValueError("Custom field 'foo' not found") + + set_document_custom_fields(42, {"foo": "bar"}, "task-123") + + mock_patch.assert_not_called() + + @patch("app.tasks.upload_to_paperless.get_custom_field_id") + @patch("app.tasks.upload_to_paperless.requests.patch") + @patch("app.tasks.upload_to_paperless.settings") + def test_handles_patch_failure(self, mock_settings, mock_patch, mock_field_id): + """Test that PATCH failure is logged but does not raise.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + mock_field_id.return_value = 5 + + mock_exc = requests.exceptions.HTTPError("500 Server Error") + mock_exc.response = Mock() + mock_exc.response.text = "Internal server error" + mock_patch.side_effect = mock_exc + + # Should not raise + set_document_custom_fields(42, {"sender": "John"}, "task-123") + + +@pytest.mark.unit +class TestUploadToPaperless: + """Tests for upload_to_paperless Celery task.""" + + @patch("app.tasks.upload_to_paperless.log_task_progress") + def test_file_not_found(self, mock_log): + """Test that missing file raises FileNotFoundError.""" + mock_self = MagicMock() + mock_self.request.id = "test-task" + + with pytest.raises(FileNotFoundError): + upload_to_paperless.__wrapped__(mock_self, "/nonexistent/file.pdf", file_id=1) + + @patch("app.tasks.upload_to_paperless.set_document_custom_fields") + @patch("app.tasks.upload_to_paperless.poll_task_for_document_id") + @patch("app.tasks.upload_to_paperless.requests.post") + @patch("app.tasks.upload_to_paperless.settings") + @patch("app.tasks.upload_to_paperless.log_task_progress") + def test_successful_upload(self, mock_log, mock_settings, mock_post, mock_poll, mock_set_fields, tmp_path): + """Test successful upload to Paperless.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + mock_settings.paperless_custom_fields_mapping = None + mock_settings.paperless_custom_field_absender = None + + # Create test file + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"%PDF-1.4 test content") + + mock_response = Mock() + mock_response.text = '"task-uuid-123"' + mock_response.raise_for_status = Mock() + mock_post.return_value = mock_response + + mock_poll.return_value = 42 + + result = upload_to_paperless.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Completed" + assert result["paperless_document_id"] == 42 + + @patch("app.tasks.upload_to_paperless.poll_task_for_document_id") + @patch("app.tasks.upload_to_paperless.requests.post") + @patch("app.tasks.upload_to_paperless.settings") + @patch("app.tasks.upload_to_paperless.log_task_progress") + def test_duplicate_document(self, mock_log, mock_settings, mock_post, mock_poll, tmp_path): + """Test handling of duplicate document detection.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + mock_settings.paperless_custom_fields_mapping = None + mock_settings.paperless_custom_field_absender = None + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"%PDF-1.4 test content") + + mock_response = Mock() + mock_response.text = '"task-uuid-123"' + mock_response.raise_for_status = Mock() + mock_post.return_value = mock_response + + mock_poll.return_value = None # Duplicate detected + + result = upload_to_paperless.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Duplicate" + assert result["paperless_document_id"] is None + + @patch("app.tasks.upload_to_paperless.log_task_progress") + @patch("app.tasks.upload_to_paperless.settings") + def test_missing_config_raises_value_error(self, mock_settings, mock_log, tmp_path): + """Test that missing Paperless config raises ValueError.""" + mock_settings.paperless_host = "" + mock_settings.paperless_ngx_api_token = "" + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"%PDF-1.4 test content") + + mock_self = MagicMock() + mock_self.request.id = "test-task" + + with pytest.raises(ValueError, match="not fully configured"): + upload_to_paperless.__wrapped__(mock_self, str(test_file), file_id=1) + + @patch("app.tasks.upload_to_paperless.requests.post") + @patch("app.tasks.upload_to_paperless.settings") + @patch("app.tasks.upload_to_paperless.log_task_progress") + def test_upload_request_failure(self, mock_log, mock_settings, mock_post, tmp_path): + """Test that failed HTTP request raises.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"%PDF-1.4 test content") + + mock_exc = requests.exceptions.ConnectionError("Connection refused") + mock_exc.response = None + mock_post.side_effect = mock_exc + + mock_self = MagicMock() + mock_self.request.id = "test-task" + + with pytest.raises(requests.exceptions.ConnectionError): + upload_to_paperless.__wrapped__(mock_self, str(test_file), file_id=1) + + @patch("app.tasks.upload_to_paperless.set_document_custom_fields") + @patch("app.tasks.upload_to_paperless.poll_task_for_document_id") + @patch("app.tasks.upload_to_paperless.requests.post") + @patch("app.tasks.upload_to_paperless.settings") + @patch("app.tasks.upload_to_paperless.log_task_progress") + def test_loads_metadata_from_json(self, mock_log, mock_settings, mock_post, mock_poll, mock_set_fields, tmp_path): + """Test that metadata is loaded from accompanying JSON file.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + mock_settings.paperless_custom_fields_mapping = json.dumps({"absender": "Sender"}) + mock_settings.paperless_custom_field_absender = None + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"%PDF-1.4 test content") + + # Create metadata JSON + json_file = tmp_path / "test.json" + json_file.write_text(json.dumps({"absender": "Test Sender", "date": "2024-01-01"})) + + mock_response = Mock() + mock_response.text = '"task-uuid-123"' + mock_response.raise_for_status = Mock() + mock_post.return_value = mock_response + + mock_poll.return_value = 42 + + result = upload_to_paperless.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Completed" + mock_set_fields.assert_called_once() + # Verify the custom fields include the mapped metadata + call_args = mock_set_fields.call_args + assert "Sender" in call_args[0][1] + + @patch("app.tasks.upload_to_paperless.set_document_custom_fields") + @patch("app.tasks.upload_to_paperless.poll_task_for_document_id") + @patch("app.tasks.upload_to_paperless.requests.post") + @patch("app.tasks.upload_to_paperless.settings") + @patch("app.tasks.upload_to_paperless.log_task_progress") + def test_legacy_absender_field(self, mock_log, mock_settings, mock_post, mock_poll, mock_set_fields, tmp_path): + """Test legacy absender field configuration fallback.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + mock_settings.paperless_custom_fields_mapping = None + mock_settings.paperless_custom_field_absender = "Absender" + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"%PDF-1.4 test content") + + json_file = tmp_path / "test.json" + json_file.write_text(json.dumps({"absender": "Legacy Sender"})) + + mock_response = Mock() + mock_response.text = '"task-uuid-123"' + mock_response.raise_for_status = Mock() + mock_post.return_value = mock_response + + mock_poll.return_value = 42 + + result = upload_to_paperless.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Completed" + mock_set_fields.assert_called_once() + call_args = mock_set_fields.call_args + assert "Absender" in call_args[0][1] + + @patch("app.tasks.upload_to_paperless.set_document_custom_fields") + @patch("app.tasks.upload_to_paperless.poll_task_for_document_id") + @patch("app.tasks.upload_to_paperless.requests.post") + @patch("app.tasks.upload_to_paperless.settings") + @patch("app.tasks.upload_to_paperless.log_task_progress") + def test_invalid_json_mapping_handled( + self, mock_log, mock_settings, mock_post, mock_poll, mock_set_fields, tmp_path + ): + """Test that invalid JSON mapping does not crash the task.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + mock_settings.paperless_custom_fields_mapping = "not-valid-json" + mock_settings.paperless_custom_field_absender = None + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"%PDF-1.4 test content") + + mock_response = Mock() + mock_response.text = '"task-uuid-123"' + mock_response.raise_for_status = Mock() + mock_post.return_value = mock_response + + mock_poll.return_value = 42 + + result = upload_to_paperless.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + assert result["status"] == "Completed" + + @patch("app.tasks.upload_to_paperless.set_document_custom_fields") + @patch("app.tasks.upload_to_paperless.poll_task_for_document_id") + @patch("app.tasks.upload_to_paperless.requests.post") + @patch("app.tasks.upload_to_paperless.settings") + @patch("app.tasks.upload_to_paperless.log_task_progress") + def test_custom_fields_failure_does_not_fail_upload( + self, mock_log, mock_settings, mock_post, mock_poll, mock_set_fields, tmp_path + ): + """Test that custom field errors don't fail the entire upload.""" + mock_settings.paperless_host = "http://paperless:8000" + mock_settings.paperless_ngx_api_token = "test-token" + mock_settings.http_request_timeout = 30 + mock_settings.paperless_custom_fields_mapping = json.dumps({"absender": "Sender"}) + mock_settings.paperless_custom_field_absender = None + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"%PDF-1.4 test content") + + json_file = tmp_path / "test.json" + json_file.write_text(json.dumps({"absender": "Test"})) + + mock_response = Mock() + mock_response.text = '"task-uuid-123"' + mock_response.raise_for_status = Mock() + mock_post.return_value = mock_response + + mock_poll.return_value = 42 + mock_set_fields.side_effect = Exception("Custom fields error") + + result = upload_to_paperless.apply(args=[str(test_file)], kwargs={"file_id": 1}).get() + + # Upload should still succeed even though custom fields failed + assert result["status"] == "Completed" diff --git a/tests/test_upload_with_rclone.py b/tests/test_upload_with_rclone.py new file mode 100644 index 00000000..565545fd --- /dev/null +++ b/tests/test_upload_with_rclone.py @@ -0,0 +1,308 @@ +""" +Tests for app/tasks/upload_with_rclone.py module. + +Extends existing tests with comprehensive coverage for upload_with_rclone +and send_to_all_rclone_destinations Celery tasks. +""" + +import os +import subprocess +from unittest.mock import MagicMock, Mock, call, patch + +import pytest + +from app.tasks.upload_with_rclone import send_to_all_rclone_destinations, upload_with_rclone + + +@pytest.mark.unit +class TestUploadWithRcloneExtended: + """Extended tests for upload_with_rclone task.""" + + @patch("app.tasks.upload_with_rclone.subprocess.run") + @patch("app.tasks.upload_with_rclone.log_task_progress") + @patch("app.tasks.upload_with_rclone.settings") + def test_successful_upload(self, mock_settings, mock_log, mock_run, tmp_path): + """Test successful rclone upload.""" + mock_settings.workdir = str(tmp_path) + + # Create test file and rclone config + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + rclone_config = tmp_path / "rclone.conf" + rclone_config.write_text("[gdrive]\ntype = drive\n") + + # Mock successful subprocess calls + mock_mkdir = Mock() + mock_mkdir.returncode = 0 + mock_upload = Mock() + mock_upload.returncode = 0 + mock_upload.stdout = "" + mock_upload.stderr = "" + mock_link = Mock() + mock_link.returncode = 0 + mock_link.stdout = "https://drive.google.com/file/abc123\n" + + mock_run.side_effect = [mock_mkdir, mock_upload, mock_link] + + result = upload_with_rclone(str(test_file), "gdrive:uploads") + + assert result["status"] == "Completed" + assert result["destination"] == "gdrive:uploads" + assert result["public_url"] == "https://drive.google.com/file/abc123" + + @patch("app.tasks.upload_with_rclone.subprocess.run") + @patch("app.tasks.upload_with_rclone.log_task_progress") + @patch("app.tasks.upload_with_rclone.settings") + def test_successful_upload_no_public_url(self, mock_settings, mock_log, mock_run, tmp_path): + """Test successful upload when public link is not available.""" + mock_settings.workdir = str(tmp_path) + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + rclone_config = tmp_path / "rclone.conf" + rclone_config.write_text("[gdrive]\ntype = drive\n") + + mock_mkdir = Mock() + mock_mkdir.returncode = 0 + mock_upload = Mock() + mock_upload.returncode = 0 + mock_link = Mock() + mock_link.returncode = 1 # Public link not available + + mock_run.side_effect = [mock_mkdir, mock_upload, mock_link] + + result = upload_with_rclone(str(test_file), "gdrive:uploads") + + assert result["status"] == "Completed" + assert result["public_url"] is None + + @patch("app.tasks.upload_with_rclone.subprocess.run") + @patch("app.tasks.upload_with_rclone.log_task_progress") + @patch("app.tasks.upload_with_rclone.settings") + def test_upload_link_exception(self, mock_settings, mock_log, mock_run, tmp_path): + """Test that link failure does not fail the upload.""" + mock_settings.workdir = str(tmp_path) + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + rclone_config = tmp_path / "rclone.conf" + rclone_config.write_text("[gdrive]\ntype = drive\n") + + mock_mkdir = Mock() + mock_mkdir.returncode = 0 + mock_upload = Mock() + mock_upload.returncode = 0 + + # First two calls succeed, link raises exception + mock_run.side_effect = [mock_mkdir, mock_upload, subprocess.SubprocessError("link failed")] + + result = upload_with_rclone(str(test_file), "gdrive:uploads") + + assert result["status"] == "Completed" + assert result["public_url"] is None + + @patch("app.tasks.upload_with_rclone.subprocess.run") + @patch("app.tasks.upload_with_rclone.log_task_progress") + @patch("app.tasks.upload_with_rclone.settings") + def test_mkdir_failure(self, mock_settings, mock_log, mock_run, tmp_path): + """Test rclone mkdir failure.""" + mock_settings.workdir = str(tmp_path) + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + rclone_config = tmp_path / "rclone.conf" + rclone_config.write_text("[gdrive]\ntype = drive\n") + + mock_run.side_effect = subprocess.CalledProcessError( + 1, "rclone", stderr=b"mkdir failed" + ) + + with pytest.raises(RuntimeError, match="Rclone error"): + upload_with_rclone(str(test_file), "gdrive:uploads") + + @patch("app.tasks.upload_with_rclone.subprocess.run") + @patch("app.tasks.upload_with_rclone.log_task_progress") + @patch("app.tasks.upload_with_rclone.settings") + def test_upload_command_failure(self, mock_settings, mock_log, mock_run, tmp_path): + """Test rclone copy command failure.""" + mock_settings.workdir = str(tmp_path) + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + rclone_config = tmp_path / "rclone.conf" + rclone_config.write_text("[gdrive]\ntype = drive\n") + + mock_mkdir = Mock() + mock_mkdir.returncode = 0 + + mock_run.side_effect = [ + mock_mkdir, + subprocess.CalledProcessError(1, "rclone", stderr="upload failed"), + ] + + with pytest.raises(RuntimeError, match="Rclone error"): + upload_with_rclone(str(test_file), "gdrive:uploads") + + @patch("app.tasks.upload_with_rclone.log_task_progress") + def test_os_error_handling(self, mock_log, tmp_path): + """Test OSError during rclone execution.""" + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + rclone_config = tmp_path / "rclone.conf" + rclone_config.write_text("[gdrive]\ntype = drive\n") + + with patch("app.tasks.upload_with_rclone.settings") as mock_settings: + mock_settings.workdir = str(tmp_path) + + with patch("app.tasks.upload_with_rclone.subprocess.run", side_effect=OSError("Permission denied")): + with pytest.raises(RuntimeError, match="Error uploading"): + upload_with_rclone(str(test_file), "gdrive:uploads") + + @patch("app.tasks.upload_with_rclone.log_task_progress") + def test_validates_remote_name_special_chars(self, mock_log, tmp_path): + """Test that special characters in remote name are rejected.""" + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + with pytest.raises(ValueError, match="Invalid remote name"): + upload_with_rclone(str(test_file), "rem ote:path") + + @patch("app.tasks.upload_with_rclone.log_task_progress") + def test_validates_remote_name_with_underscore_hyphen(self, mock_log, tmp_path): + """Test that underscores and hyphens in remote names are valid.""" + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + with patch("app.tasks.upload_with_rclone.settings") as mock_settings: + mock_settings.workdir = str(tmp_path) + # No rclone.conf -> ValueError for config not found + with pytest.raises(ValueError, match="Rclone configuration not found"): + upload_with_rclone(str(test_file), "my-remote_1:path") + + +@pytest.mark.unit +class TestSendToAllRcloneDestinations: + """Tests for send_to_all_rclone_destinations task.""" + + def test_file_not_found(self): + """Test raises FileNotFoundError for missing file.""" + with pytest.raises(FileNotFoundError): + send_to_all_rclone_destinations("/nonexistent/file.pdf") + + @patch("app.tasks.upload_with_rclone.upload_with_rclone") + @patch("app.tasks.upload_with_rclone.subprocess.run") + @patch("app.tasks.upload_with_rclone.log_task_progress") + @patch("app.tasks.upload_with_rclone.settings") + def test_successful_queue_to_all_destinations(self, mock_settings, mock_log, mock_run, mock_upload, tmp_path): + """Test queuing uploads to all configured remotes.""" + mock_settings.workdir = str(tmp_path) + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + rclone_config = tmp_path / "rclone.conf" + rclone_config.write_text("[gdrive]\ntype=drive\n[s3]\ntype=s3\n") + + mock_remotes = Mock() + mock_remotes.returncode = 0 + mock_remotes.stdout = "gdrive:\ns3:\n" + mock_run.return_value = mock_remotes + + mock_task = Mock() + mock_task.id = "task-123" + mock_upload.delay.return_value = mock_task + + result = send_to_all_rclone_destinations(str(test_file)) + + assert result["status"] == "Queued" + assert "tasks" in result + assert len(result["tasks"]) == 2 + + @patch("app.tasks.upload_with_rclone.subprocess.run") + @patch("app.tasks.upload_with_rclone.log_task_progress") + @patch("app.tasks.upload_with_rclone.settings") + def test_no_rclone_config(self, mock_settings, mock_log, mock_run, tmp_path): + """Test error when rclone config is missing.""" + mock_settings.workdir = str(tmp_path) + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + with pytest.raises(ValueError, match="Rclone configuration not found"): + send_to_all_rclone_destinations(str(test_file)) + + @patch("app.tasks.upload_with_rclone.subprocess.run") + @patch("app.tasks.upload_with_rclone.log_task_progress") + @patch("app.tasks.upload_with_rclone.settings") + def test_listremotes_failure(self, mock_settings, mock_log, mock_run, tmp_path): + """Test error when listremotes command fails.""" + mock_settings.workdir = str(tmp_path) + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + rclone_config = tmp_path / "rclone.conf" + rclone_config.write_text("[gdrive]\ntype=drive\n") + + mock_run.side_effect = subprocess.SubprocessError("rclone not found") + + with pytest.raises(RuntimeError, match="Error setting up"): + send_to_all_rclone_destinations(str(test_file)) + + @patch("app.tasks.upload_with_rclone.upload_with_rclone") + @patch("app.tasks.upload_with_rclone.subprocess.run") + @patch("app.tasks.upload_with_rclone.log_task_progress") + @patch("app.tasks.upload_with_rclone.settings") + def test_uses_custom_path_settings(self, mock_settings, mock_log, mock_run, mock_upload, tmp_path): + """Test that custom path settings are used for each remote.""" + mock_settings.workdir = str(tmp_path) + mock_settings.rclone_gdrive_path = "Documents/Uploads" + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + rclone_config = tmp_path / "rclone.conf" + rclone_config.write_text("[gdrive]\ntype=drive\n") + + mock_remotes = Mock() + mock_remotes.returncode = 0 + mock_remotes.stdout = "gdrive:\n" + mock_run.return_value = mock_remotes + + mock_task = Mock() + mock_task.id = "task-123" + mock_upload.delay.return_value = mock_task + + result = send_to_all_rclone_destinations(str(test_file)) + + assert result["status"] == "Queued" + # Check the destination includes the custom path + call_args = mock_upload.delay.call_args + assert "Documents/Uploads" in call_args[0][1] + + @patch("app.tasks.upload_with_rclone.subprocess.run") + @patch("app.tasks.upload_with_rclone.log_task_progress") + @patch("app.tasks.upload_with_rclone.settings") + def test_listremotes_nonzero_return_code(self, mock_settings, mock_log, mock_run, tmp_path): + """Test error when listremotes returns non-zero exit code.""" + mock_settings.workdir = str(tmp_path) + + test_file = tmp_path / "test.pdf" + test_file.write_bytes(b"test content") + + rclone_config = tmp_path / "rclone.conf" + rclone_config.write_text("[gdrive]\ntype=drive\n") + + mock_result = Mock() + mock_result.returncode = 1 + mock_result.stderr = "config error" + mock_run.return_value = mock_result + + with pytest.raises(RuntimeError, match="Failed to list rclone remotes"): + send_to_all_rclone_destinations(str(test_file))