From 80de3b674332d9eeebd966394cbc9bf36538f615 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Mon, 23 Mar 2026 14:04:30 +0000 Subject: [PATCH] Refactor URL creation to use reusable join_url utility Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- app/tasks/upload_to_nextcloud.py | 298 ++++++++++----------- app/utils/network.py | 13 + tests/test_upload_to_nextcloud_join_url.py | 49 ++++ 3 files changed, 203 insertions(+), 157 deletions(-) create mode 100644 tests/test_upload_to_nextcloud_join_url.py diff --git a/app/tasks/upload_to_nextcloud.py b/app/tasks/upload_to_nextcloud.py index 1bb29c75..b8dc52ba 100644 --- a/app/tasks/upload_to_nextcloud.py +++ b/app/tasks/upload_to_nextcloud.py @@ -1,157 +1,141 @@ -#!/usr/bin/env python3 - -import logging -import os - -import requests -from requests.auth import HTTPBasicAuth - -from app.celery_app import celery -from app.config import settings -from app.tasks.retry_config import UploadTaskWithRetry -from app.utils import log_task_progress -from app.utils.filename_utils import extract_remote_path, get_unique_filename - -logger = logging.getLogger(__name__) - - -@celery.task(base=UploadTaskWithRetry, bind=True) -def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_override: str = None): - """ - Upload a file to Nextcloud WebDAV. - - Args: - file_path: Path to the file to upload - file_id: Optional file ID to associate with logs - """ - task_id = self.request.id - logger.info(f"[{task_id}] Starting Nextcloud upload: {file_path}") - log_task_progress( - task_id, - "upload_to_nextcloud", - "in_progress", - f"Uploading to Nextcloud: {os.path.basename(file_path)}", - file_id=file_id, - ) - - if not os.path.exists(file_path): - error_msg = f"File not found: {file_path}" - logger.error(f"[{task_id}] {error_msg}") - log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id) - raise FileNotFoundError(error_msg) - - # For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url' - # This is what's shown in your env view - if not ( - getattr(settings, "nextcloud_upload_url", None) - and getattr(settings, "nextcloud_username", None) - and getattr(settings, "nextcloud_password", None) - ): - logger.info(f"[{task_id}] Nextcloud upload skipped: Missing configuration") - log_task_progress(task_id, "upload_to_nextcloud", "success", "Skipped: Not configured", file_id=file_id) - return {"status": "Skipped", "reason": "Nextcloud settings not configured"} - - filename = os.path.basename(file_path) - - try: - # Prepare WebDAV URL - use nextcloud_upload_url instead of nextcloud_url - webdav_url = settings.nextcloud_upload_url - if not webdav_url.endswith("/"): - webdav_url += "/" - - # Calculate remote path based on local file structure - remote_base = ( - folder_override if folder_override is not None else (getattr(settings, "nextcloud_folder", "") or "") - ) - remote_path = extract_remote_path(file_path, settings.workdir, remote_base) - full_url = f"{webdav_url}/{remote_path}" - - # Remove any double slashes (except in http://) - full_url = full_url.replace("://", "$PLACEHOLDER$") - while "//" in full_url: - full_url = full_url.replace("//", "/") - full_url = full_url.replace("$PLACEHOLDER$", "://") - - # Function to check if file exists in Nextcloud - def check_exists_in_nextcloud(path): - check_url = f"{webdav_url}{os.path.dirname(path)}" - try: - response = requests.request( - "PROPFIND", - check_url, - auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), - headers={"Depth": "1"}, - timeout=10, - ) - - return path in response.text - except Exception: - # If we can't check, assume it doesn't exist - return False - - # Check for potential file collision and get a unique name if needed - remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud) - full_url = f"{webdav_url}/{remote_path}" - - # Fix double slashes again - full_url = full_url.replace("://", "$PLACEHOLDER$") - while "//" in full_url: - full_url = full_url.replace("//", "/") - full_url = full_url.replace("$PLACEHOLDER$", "://") - - # Create necessary parent folders - parent_dirs = os.path.dirname(remote_path) - if parent_dirs: - current_path = "" - for folder in parent_dirs.split("/"): - if not folder: - continue - current_path += f"{folder}/" - mkdir_url = f"{webdav_url}/{current_path}" - # Fix double slashes - mkdir_url = mkdir_url.replace("://", "$PLACEHOLDER$") - while "//" in mkdir_url: - mkdir_url = mkdir_url.replace("//", "/") - mkdir_url = mkdir_url.replace("$PLACEHOLDER$", "://") - - requests.request( - "MKCOL", - mkdir_url, - auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), - timeout=10, - ) - - # Upload the file - logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}") - log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id) - with open(file_path, "rb") as file_data: - response = requests.put( - full_url, - data=file_data, - auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), - headers={"Content-Type": "application/octet-stream"}, - timeout=settings.http_request_timeout, # Use configured timeout for large files - ) - - if response.status_code in (201, 204): # Created or No Content - logger.info(f"[{task_id}] Successfully uploaded {filename} to Nextcloud at {remote_path}") - log_task_progress( - task_id, "upload_to_nextcloud", "success", f"Uploaded to Nextcloud: {remote_path}", file_id=file_id - ) - return { - "status": "Completed", - "file_path": file_path, - "nextcloud_path": remote_path, - "response_code": response.status_code, - } - else: - error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}" - logger.error(f"[{task_id}] {error_msg}") - log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id) - raise Exception(error_msg) - - except Exception as e: - error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}" - logger.error(f"[{task_id}] {error_msg}") - log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id) - raise Exception(error_msg) +#!/usr/bin/env python3 + +import logging +import os + +import requests +from requests.auth import HTTPBasicAuth + +from app.celery_app import celery +from app.config import settings +from app.tasks.retry_config import UploadTaskWithRetry +from app.utils import log_task_progress +from app.utils.filename_utils import extract_remote_path, get_unique_filename +from app.utils.network import join_url + +logger = logging.getLogger(__name__) + + +@celery.task(base=UploadTaskWithRetry, bind=True) +def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_override: str = None): + """ + Upload a file to Nextcloud WebDAV. + + Args: + file_path: Path to the file to upload + file_id: Optional file ID to associate with logs + """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting Nextcloud upload: {file_path}") + log_task_progress( + task_id, + "upload_to_nextcloud", + "in_progress", + f"Uploading to Nextcloud: {os.path.basename(file_path)}", + file_id=file_id, + ) + + if not os.path.exists(file_path): + error_msg = f"File not found: {file_path}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id) + raise FileNotFoundError(error_msg) + + # For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url' + # This is what's shown in your env view + if not ( + getattr(settings, "nextcloud_upload_url", None) + and getattr(settings, "nextcloud_username", None) + and getattr(settings, "nextcloud_password", None) + ): + logger.info(f"[{task_id}] Nextcloud upload skipped: Missing configuration") + log_task_progress(task_id, "upload_to_nextcloud", "success", "Skipped: Not configured", file_id=file_id) + return {"status": "Skipped", "reason": "Nextcloud settings not configured"} + + filename = os.path.basename(file_path) + + try: + # Prepare WebDAV URL - use nextcloud_upload_url instead of nextcloud_url + webdav_url = settings.nextcloud_upload_url + if not webdav_url.endswith("/"): + webdav_url += "/" + + # Calculate remote path based on local file structure + remote_base = ( + folder_override if folder_override is not None else (getattr(settings, "nextcloud_folder", "") or "") + ) + remote_path = extract_remote_path(file_path, settings.workdir, remote_base) + full_url = join_url(webdav_url, remote_path) + + # Function to check if file exists in Nextcloud + def check_exists_in_nextcloud(path): + check_url = join_url(webdav_url, os.path.dirname(path)) + try: + response = requests.request( + "PROPFIND", + check_url, + auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), + headers={"Depth": "1"}, + timeout=10, + ) + + return path in response.text + except Exception: + # If we can't check, assume it doesn't exist + return False + + # Check for potential file collision and get a unique name if needed + remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud) + full_url = join_url(webdav_url, remote_path) + + # Create necessary parent folders + parent_dirs = os.path.dirname(remote_path) + if parent_dirs: + current_path = "" + for folder in parent_dirs.split("/"): + if not folder: + continue + current_path += f"{folder}/" + mkdir_url = join_url(webdav_url, current_path) + + requests.request( + "MKCOL", + mkdir_url, + auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), + timeout=10, + ) + + # Upload the file + logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}") + log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id) + with open(file_path, "rb") as file_data: + response = requests.put( + full_url, + data=file_data, + auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password), + headers={"Content-Type": "application/octet-stream"}, + timeout=settings.http_request_timeout, # Use configured timeout for large files + ) + + if response.status_code in (201, 204): # Created or No Content + logger.info(f"[{task_id}] Successfully uploaded {filename} to Nextcloud at {remote_path}") + log_task_progress( + task_id, "upload_to_nextcloud", "success", f"Uploaded to Nextcloud: {remote_path}", file_id=file_id + ) + return { + "status": "Completed", + "file_path": file_path, + "nextcloud_path": remote_path, + "response_code": response.status_code, + } + else: + error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id) + raise Exception(error_msg) + + except Exception as e: + error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id) + raise Exception(error_msg) diff --git a/app/utils/network.py b/app/utils/network.py index 7ea6d88b..4a389e23 100644 --- a/app/utils/network.py +++ b/app/utils/network.py @@ -32,3 +32,16 @@ def is_private_ip(hostname: str) -> bool: # Log this for debugging logger.warning(f"Could not resolve hostname: {hostname}") return False # Changed from True to False to allow external domains in tests + + +def join_url(base: str, *parts: str) -> str: + """ + Safely join a base URL and multiple path parts. + Handles double slashes while preserving the protocol '://'. + """ + url = "/".join([base, *parts]) + url = url.replace("://", "$PLACEHOLDER$") + while "//" in url: + url = url.replace("//", "/") + url = url.replace("$PLACEHOLDER$", "://") + return url diff --git a/tests/test_upload_to_nextcloud_join_url.py b/tests/test_upload_to_nextcloud_join_url.py new file mode 100644 index 00000000..d5bc1bc9 --- /dev/null +++ b/tests/test_upload_to_nextcloud_join_url.py @@ -0,0 +1,49 @@ +import pytest +from unittest.mock import patch, MagicMock +import os +from app.tasks.upload_to_nextcloud import upload_to_nextcloud + +@pytest.fixture +def mock_settings(): + with patch("app.tasks.upload_to_nextcloud.settings") as mock: + mock.nextcloud_upload_url = "http://nextcloud.local/" + mock.nextcloud_username = "testuser" + mock.nextcloud_password = "testpassword" + mock.nextcloud_folder = "uploads" + mock.workdir = "/tmp/workdir" + mock.http_request_timeout = 30 + yield mock + +@pytest.fixture +def mock_requests(): + with patch("app.tasks.upload_to_nextcloud.requests") as mock: + # Mock PROPFIND to always return false (file doesn't exist) + mock.request.return_value = MagicMock(text="") + + # Mock PUT to return success + put_response = MagicMock() + put_response.status_code = 201 + mock.put.return_value = put_response + yield mock + +def test_upload_to_nextcloud_url_construction(mock_settings, mock_requests): + file_path = "/tmp/workdir/test_file.txt" + + # Create dummy file + os.makedirs("/tmp/workdir", exist_ok=True) + with open(file_path, "w") as f: + f.write("test content") + + # Call the task directly + with patch("app.tasks.upload_to_nextcloud.upload_to_nextcloud.request") as mock_req: + mock_req.id = "test-task-123" + result = upload_to_nextcloud(file_path) + + assert result["status"] == "Completed" + assert result["nextcloud_path"] == "uploads/test_file.txt" + + # Verify requests.put was called with the correct URL + mock_requests.put.assert_called_once() + args, kwargs = mock_requests.put.call_args + url = args[0] + assert url == "http://nextcloud.local/uploads/test_file.txt"