fix: return boolean values from _should_upload functions and fix test expectations
Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
+1
-1
@@ -334,7 +334,7 @@ def bulk_reprocess_files(request: Request, file_ids: List[int], db: DbSession):
|
|||||||
"status": "success" if processed_files else "error",
|
"status": "success" if processed_files else "error",
|
||||||
"message": f"Successfully queued {len(processed_files)} files for reprocessing",
|
"message": f"Successfully queued {len(processed_files)} files for reprocessing",
|
||||||
"processed_files": processed_files,
|
"processed_files": processed_files,
|
||||||
"errors": errors if errors else None,
|
"errors": errors,
|
||||||
"task_ids": task_ids,
|
"task_ids": task_ids,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
+11
-11
@@ -25,21 +25,21 @@ logger = logging.getLogger(__name__)
|
|||||||
|
|
||||||
|
|
||||||
def _should_upload_to_dropbox():
|
def _should_upload_to_dropbox():
|
||||||
return settings.dropbox_app_key and settings.dropbox_app_secret and settings.dropbox_refresh_token
|
return bool(settings.dropbox_app_key and settings.dropbox_app_secret and settings.dropbox_refresh_token)
|
||||||
|
|
||||||
|
|
||||||
def _should_upload_to_nextcloud():
|
def _should_upload_to_nextcloud():
|
||||||
return settings.nextcloud_upload_url and settings.nextcloud_username and settings.nextcloud_password
|
return bool(settings.nextcloud_upload_url and settings.nextcloud_username and settings.nextcloud_password)
|
||||||
|
|
||||||
|
|
||||||
def _should_upload_to_paperless():
|
def _should_upload_to_paperless():
|
||||||
return settings.paperless_ngx_api_token and settings.paperless_host
|
return bool(settings.paperless_ngx_api_token and settings.paperless_host)
|
||||||
|
|
||||||
|
|
||||||
def _should_upload_to_google_drive():
|
def _should_upload_to_google_drive():
|
||||||
# Check for OAuth configuration
|
# Check for OAuth configuration
|
||||||
if getattr(settings, "google_drive_use_oauth", False):
|
if getattr(settings, "google_drive_use_oauth", False):
|
||||||
return (
|
return bool(
|
||||||
settings.google_drive_client_id
|
settings.google_drive_client_id
|
||||||
and settings.google_drive_client_secret
|
and settings.google_drive_client_secret
|
||||||
and settings.google_drive_refresh_token
|
and settings.google_drive_refresh_token
|
||||||
@@ -47,33 +47,33 @@ def _should_upload_to_google_drive():
|
|||||||
)
|
)
|
||||||
# Or check for service account configuration
|
# Or check for service account configuration
|
||||||
else:
|
else:
|
||||||
return settings.google_drive_credentials_json and settings.google_drive_folder_id
|
return bool(settings.google_drive_credentials_json and settings.google_drive_folder_id)
|
||||||
|
|
||||||
|
|
||||||
def _should_upload_to_webdav():
|
def _should_upload_to_webdav():
|
||||||
return settings.webdav_url and settings.webdav_username and settings.webdav_password
|
return bool(settings.webdav_url and settings.webdav_username and settings.webdav_password)
|
||||||
|
|
||||||
|
|
||||||
def _should_upload_to_ftp():
|
def _should_upload_to_ftp():
|
||||||
return settings.ftp_host and settings.ftp_username and settings.ftp_password
|
return bool(settings.ftp_host and settings.ftp_username and settings.ftp_password)
|
||||||
|
|
||||||
|
|
||||||
def _should_upload_to_sftp():
|
def _should_upload_to_sftp():
|
||||||
return settings.sftp_host and settings.sftp_username and (settings.sftp_password or settings.sftp_private_key)
|
return bool(settings.sftp_host and settings.sftp_username and (settings.sftp_password or settings.sftp_private_key))
|
||||||
|
|
||||||
|
|
||||||
def _should_upload_to_email():
|
def _should_upload_to_email():
|
||||||
return (
|
return bool(
|
||||||
settings.email_host and settings.email_username and settings.email_password and settings.email_default_recipient
|
settings.email_host and settings.email_username and settings.email_password and settings.email_default_recipient
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def _should_upload_to_onedrive():
|
def _should_upload_to_onedrive():
|
||||||
return settings.onedrive_client_id and settings.onedrive_client_secret and settings.onedrive_refresh_token
|
return bool(settings.onedrive_client_id and settings.onedrive_client_secret and settings.onedrive_refresh_token)
|
||||||
|
|
||||||
|
|
||||||
def _should_upload_to_s3():
|
def _should_upload_to_s3():
|
||||||
return settings.s3_bucket_name and settings.aws_access_key_id and settings.aws_secret_access_key
|
return bool(settings.s3_bucket_name and settings.aws_access_key_id and settings.aws_secret_access_key)
|
||||||
|
|
||||||
|
|
||||||
def get_configured_services_from_validator():
|
def get_configured_services_from_validator():
|
||||||
|
|||||||
@@ -182,10 +182,15 @@ class TestGetConfiguredServicesFromValidator:
|
|||||||
class TestSendToAllDestinations:
|
class TestSendToAllDestinations:
|
||||||
"""Test send_to_all_destinations task."""
|
"""Test send_to_all_destinations task."""
|
||||||
|
|
||||||
def test_file_not_found_error(self):
|
@patch("app.tasks.send_to_all.log_task_progress")
|
||||||
|
def test_file_not_found_error(self, mock_log):
|
||||||
"""Test that FileNotFoundError is raised when file doesn't exist."""
|
"""Test that FileNotFoundError is raised when file doesn't exist."""
|
||||||
with pytest.raises(FileNotFoundError):
|
result = send_to_all_destinations.apply(args=["/nonexistent/file.pdf"])
|
||||||
send_to_all_destinations.apply(args=["/nonexistent/file.pdf"])
|
|
||||||
|
# When task raises an exception, result.failed() returns True
|
||||||
|
assert result.failed()
|
||||||
|
# The exception should be FileNotFoundError
|
||||||
|
assert isinstance(result.result, FileNotFoundError)
|
||||||
|
|
||||||
@patch("app.tasks.send_to_all.settings")
|
@patch("app.tasks.send_to_all.settings")
|
||||||
@patch("app.tasks.send_to_all._should_upload_to_dropbox")
|
@patch("app.tasks.send_to_all._should_upload_to_dropbox")
|
||||||
|
|||||||
Reference in New Issue
Block a user