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

@@ -11,6 +11,7 @@ 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__)
 
@@ -64,17 +65,11 @@ def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_overri
             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}"
+        full_url = join_url(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)}"
+            check_url = join_url(webdav_url, os.path.dirname(path))
             try:
                 response = requests.request(
                     "PROPFIND",
@@ -91,13 +86,7 @@ def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_overri
 
         # 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}"
+        full_url = join_url(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)
@@ -107,12 +96,7 @@ def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_overri
                 if not folder:
                     continue
                 current_path += f"{folder}/"
-                mkdir_url = f"{webdav_url}/{current_path}"
+                mkdir_url = join_url(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",

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

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="<response></response>")
+
+        # 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"