From 82d67c56b34ec5a849bb3fa46aba0182c7f51dcd Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 10 Mar 2026 00:43:00 +0000 Subject: [PATCH] feat(storage): add Apple iCloud Drive storage provider Add iCloud Drive as a new storage destination using the pyicloud library. Includes upload task, configuration, user integration handler, provider status, onboarding support, and comprehensive tests. Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com> --- .env.demo | 9 ++ =2.4.0 | 0 app/celery_worker.py | 1 + app/config.py | 6 + app/models.py | 2 + app/tasks/send_to_all.py | 11 ++ app/tasks/upload_to_icloud.py | 177 ++++++++++++++++++++ app/tasks/upload_to_user_integration.py | 32 ++++ app/utils/config_validator/providers.py | 15 ++ app/utils/settings_service.py | 33 ++++ app/views/onboarding.py | 2 + frontend/templates/files.html | 1 + requirements.txt | 3 + tests/test_send_to_all.py | 1 + tests/test_upload_to_icloud.py | 205 ++++++++++++++++++++++++ 15 files changed, 498 insertions(+) create mode 100644 =2.4.0 create mode 100644 app/tasks/upload_to_icloud.py create mode 100644 tests/test_upload_to_icloud.py diff --git a/.env.demo b/.env.demo index 65f82fdf..21f158e6 100644 --- a/.env.demo +++ b/.env.demo @@ -382,6 +382,15 @@ SFTP_PASSWORD=your_secure_sftp_password SFTP_FOLDER=/Documents/Uploads SFTP_DISABLE_HOST_KEY_VERIFICATION=False # Default is False (secure); set to True only for testing +# iCloud Drive +# Requires an Apple ID with iCloud Drive enabled. +# For accounts with two-factor authentication (most accounts), generate an +# app-specific password at https://appleid.apple.com/account/manage +ICLOUD_USERNAME=your_apple_id@example.com +ICLOUD_PASSWORD=your-app-specific-password +ICLOUD_FOLDER=Documents/Uploads +# ICLOUD_COOKIE_DIRECTORY=/path/to/cookie/dir # Optional: defaults to ~/.pyicloud + # **HTTP Request Settings** # Timeout for HTTP requests - set higher to handle large PDF files (up to 1GB) HTTP_REQUEST_TIMEOUT=120 # Timeout in seconds (default: 120 for large file operations) diff --git a/=2.4.0 b/=2.4.0 new file mode 100644 index 00000000..e69de29b diff --git a/app/celery_worker.py b/app/celery_worker.py index 4881f8a9..fd68c332 100644 --- a/app/celery_worker.py +++ b/app/celery_worker.py @@ -45,6 +45,7 @@ from app.tasks.upload_to_dropbox import upload_to_dropbox # noqa: F401 from app.tasks.upload_to_email import upload_to_email # noqa: F401 from app.tasks.upload_to_ftp import upload_to_ftp # noqa: F401 from app.tasks.upload_to_google_drive import upload_to_google_drive # noqa: F401 +from app.tasks.upload_to_icloud import upload_to_icloud # noqa: F401 from app.tasks.upload_to_nextcloud import upload_to_nextcloud # noqa: F401 from app.tasks.upload_to_onedrive import upload_to_onedrive # noqa: F401 from app.tasks.upload_to_paperless import upload_to_paperless # noqa: F401 diff --git a/app/config.py b/app/config.py index ba6eeb25..6309629c 100644 --- a/app/config.py +++ b/app/config.py @@ -466,6 +466,12 @@ class Settings(BaseSettings): s3_storage_class: Optional[str] = "STANDARD" # Default storage class s3_acl: Optional[str] = "private" # Default ACL + # iCloud Drive settings + icloud_username: Optional[str] = None # Apple ID email address + icloud_password: Optional[str] = None # App-specific password (required for 2FA accounts) + icloud_folder: Optional[str] = None # Target folder path in iCloud Drive (e.g. "Documents/Uploads") + icloud_cookie_directory: Optional[str] = None # Directory for session cookies (default: ~/.pyicloud) + # Uptime Kuma settings uptime_kuma_url: Optional[str] = None uptime_kuma_ping_interval: int = 5 # Default ping interval in minutes diff --git a/app/models.py b/app/models.py index 0cc7b53a..66fb30f3 100644 --- a/app/models.py +++ b/app/models.py @@ -505,6 +505,7 @@ class IntegrationType: EMAIL = "EMAIL" PAPERLESS = "PAPERLESS" RCLONE = "RCLONE" + ICLOUD = "ICLOUD" ALL = { IMAP, @@ -521,6 +522,7 @@ class IntegrationType: EMAIL, PAPERLESS, RCLONE, + ICLOUD, } diff --git a/app/tasks/send_to_all.py b/app/tasks/send_to_all.py index 9a9de6f3..e4ddd5b6 100644 --- a/app/tasks/send_to_all.py +++ b/app/tasks/send_to_all.py @@ -12,6 +12,7 @@ from app.tasks.upload_to_dropbox import upload_to_dropbox from app.tasks.upload_to_email import upload_to_email from app.tasks.upload_to_ftp import upload_to_ftp from app.tasks.upload_to_google_drive import upload_to_google_drive +from app.tasks.upload_to_icloud import upload_to_icloud from app.tasks.upload_to_nextcloud import upload_to_nextcloud from app.tasks.upload_to_onedrive import upload_to_onedrive from app.tasks.upload_to_paperless import upload_to_paperless @@ -79,6 +80,10 @@ def _should_upload_to_s3(): return bool(settings.s3_bucket_name and settings.aws_access_key_id and settings.aws_secret_access_key) +def _should_upload_to_icloud(): + return bool(settings.icloud_username and settings.icloud_password) + + def get_configured_services_from_validator(): """ Use the config validator to determine which services are configured properly. @@ -98,6 +103,7 @@ def get_configured_services_from_validator(): "Email": "email", "OneDrive": "onedrive", "S3 Storage": "s3", + "iCloud Drive": "icloud", } result = {} @@ -206,6 +212,11 @@ def send_to_all_destinations(self, file_path: str, use_validator=True, file_id: "should_upload": _should_upload_to_s3, "upload_func": upload_to_s3, }, + { + "name": "icloud", + "should_upload": _should_upload_to_icloud, + "upload_func": upload_to_icloud, + }, ] # Optionally get configuration status from validator diff --git a/app/tasks/upload_to_icloud.py b/app/tasks/upload_to_icloud.py new file mode 100644 index 00000000..80305544 --- /dev/null +++ b/app/tasks/upload_to_icloud.py @@ -0,0 +1,177 @@ +#!/usr/bin/env python3 + +"""Upload files to Apple iCloud Drive via the pyicloud library. + +This module uses the ``pyicloud`` library to authenticate with Apple's iCloud +service and upload files to iCloud Drive. Because Apple does not offer a public +REST API for iCloud Drive, this integration relies on the *unofficial* +reverse-engineered protocol implemented by ``pyicloud``. + +Requirements +~~~~~~~~~~~~ +* An Apple ID with iCloud Drive enabled. +* An **app-specific password** generated at https://appleid.apple.com (required + when two-factor authentication is active – which is the default for all modern + Apple IDs). +* The ``pyicloud`` Python package (``pip install pyicloud``). + +Configuration +~~~~~~~~~~~~~ +Set the following environment variables (or ``app/config.py`` fields): + +* ``ICLOUD_USERNAME`` – Apple ID email address. +* ``ICLOUD_PASSWORD`` – App-specific password. +* ``ICLOUD_FOLDER`` – Target folder path inside iCloud Drive, using ``/`` as + the separator (e.g. ``Documents/Uploads``). The folder is created + automatically if it does not exist. +* ``ICLOUD_COOKIE_DIRECTORY`` – (Optional) Directory for persisting session + cookies so that re-authentication is avoided between task runs. Defaults to + ``~/.pyicloud``. +""" + +import logging +import os + +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 + +logger = logging.getLogger(__name__) + + +def _get_icloud_api( + username: str, + password: str, + cookie_directory: str | None = None, +): + """Return an authenticated ``PyiCloudService`` instance. + + Args: + username: Apple ID email address. + password: App-specific password. + cookie_directory: Optional directory for session cookies. + + Returns: + An authenticated ``PyiCloudService`` instance. + + Raises: + ImportError: If ``pyicloud`` is not installed. + ValueError: If authentication fails or 2FA is required interactively. + """ + from pyicloud import PyiCloudService # noqa: S404 – trusted first-party usage + + kwargs: dict = {} + if cookie_directory: + kwargs["cookie_directory"] = cookie_directory + + api = PyiCloudService(username, password, **kwargs) + + # If 2SA/2FA is required the user must use an app-specific password instead. + if api.requires_2sa or api.requires_2fa: + raise ValueError( + "iCloud account requires two-factor authentication. " + "Please generate an app-specific password at https://appleid.apple.com " + "and use it as ICLOUD_PASSWORD." + ) + + return api + + +def _navigate_to_folder(drive_root, folder_path: str): + """Navigate into (or create) the folder hierarchy described by *folder_path*. + + Args: + drive_root: The iCloud Drive root node (``api.drive``). + folder_path: ``/``-separated path such as ``Documents/Uploads``. + + Returns: + The drive node representing the target folder. + """ + node = drive_root + if not folder_path: + return node + + parts = [p for p in folder_path.strip("/").split("/") if p] + for part in parts: + children = {child.name: child for child in node.dir()} + if part in children: + node = children[part] + else: + # Create the missing folder + node = node.mkdir(part) + return node + + +@celery.task(base=UploadTaskWithRetry, bind=True) +def upload_to_icloud(self, file_path: str, file_id: int = None, folder_override: str = None): + """Upload a file to Apple iCloud Drive. + + Args: + file_path: Local path to the file to upload. + file_id: Optional ``FileRecord.id`` for progress logging. + folder_override: If provided, overrides the default ``ICLOUD_FOLDER`` + setting for this upload. + """ + task_id = self.request.id + logger.info(f"[{task_id}] Starting iCloud Drive upload: {file_path}") + log_task_progress( + task_id, + "upload_to_icloud", + "in_progress", + f"Uploading to iCloud Drive: {os.path.basename(file_path)}", + file_id=file_id, + ) + + # ------------------------------------------------------------------ + # Validate inputs + # ------------------------------------------------------------------ + 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_icloud", "failure", error_msg, file_id=file_id) + raise FileNotFoundError(error_msg) + + if not settings.icloud_username or not settings.icloud_password: + error_msg = "iCloud credentials are not configured (ICLOUD_USERNAME / ICLOUD_PASSWORD)" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_icloud", "failure", error_msg, file_id=file_id) + raise ValueError(error_msg) + + filename = os.path.basename(file_path) + target_folder = folder_override if folder_override is not None else (settings.icloud_folder or "") + + # ------------------------------------------------------------------ + # Authenticate & upload + # ------------------------------------------------------------------ + try: + api = _get_icloud_api( + settings.icloud_username, + settings.icloud_password, + settings.icloud_cookie_directory, + ) + + folder_node = _navigate_to_folder(api.drive, target_folder) + + with open(file_path, "rb") as fh: + folder_node.upload(fh) + + logger.info(f"[{task_id}] Successfully uploaded {filename} to iCloud Drive folder '{target_folder}'") + log_task_progress( + task_id, + "upload_to_icloud", + "success", + f"Uploaded to iCloud Drive: {filename}", + file_id=file_id, + ) + return { + "status": "Completed", + "file": file_path, + "icloud_folder": target_folder or "/", + } + + except Exception as e: + error_msg = f"Error uploading {filename} to iCloud Drive: {e}" + logger.error(f"[{task_id}] {error_msg}") + log_task_progress(task_id, "upload_to_icloud", "failure", error_msg, file_id=file_id) + raise Exception(error_msg) from e diff --git a/app/tasks/upload_to_user_integration.py b/app/tasks/upload_to_user_integration.py index 1295f6ce..db21701d 100644 --- a/app/tasks/upload_to_user_integration.py +++ b/app/tasks/upload_to_user_integration.py @@ -571,6 +571,37 @@ def _upload_rclone(file_path: str, cfg: dict[str, Any], creds: dict[str, Any], t return {"status": "Completed", "rclone_dest": dest} +def _upload_icloud(file_path: str, cfg: dict[str, Any], creds: dict[str, Any], task_id: str) -> dict[str, Any]: + """Upload *file_path* to iCloud Drive using per-user credentials. + + Expected *cfg* keys: + * ``folder`` – target folder path inside iCloud Drive (e.g. ``Documents/Uploads``). + * ``cookie_directory`` – (optional) path for session cookie persistence. + + Expected *creds* keys: + * ``username`` – Apple ID email address. + * ``password`` – app-specific password. + """ + from app.tasks.upload_to_icloud import _get_icloud_api, _navigate_to_folder + + username = creds.get("username") or "" + password = creds.get("password") or "" + folder = cfg.get("folder") or "" + cookie_directory = cfg.get("cookie_directory") or None + + if not username or not password: + raise ValueError("iCloud integration is missing username or password in credentials") + + api = _get_icloud_api(username, password, cookie_directory) + folder_node = _navigate_to_folder(api.drive, folder) + + with open(file_path, "rb") as fh: + folder_node.upload(fh) + + logger.info("[%s] iCloud Drive upload complete: folder=%s", task_id, folder or "/") + return {"status": "Completed", "icloud_folder": folder or "/"} + + # Map IntegrationType → upload helper _UPLOAD_HANDLERS = { IntegrationType.DROPBOX: _upload_dropbox, @@ -584,6 +615,7 @@ _UPLOAD_HANDLERS = { IntegrationType.PAPERLESS: _upload_paperless, IntegrationType.EMAIL: _upload_email, IntegrationType.RCLONE: _upload_rclone, + IntegrationType.ICLOUD: _upload_icloud, } diff --git a/app/utils/config_validator/providers.py b/app/utils/config_validator/providers.py index db278d0c..9fce74b6 100644 --- a/app/utils/config_validator/providers.py +++ b/app/utils/config_validator/providers.py @@ -373,4 +373,19 @@ def get_provider_status() -> dict[str, dict[str, object]]: }, } + # Check iCloud Drive configuration + providers["iCloud Drive"] = { + "name": "iCloud Drive", + "icon": "fa-brands fa-apple", + "configured": bool(getattr(settings, "icloud_username", None) and getattr(settings, "icloud_password", None)), + "enabled": True, + "description": "Store documents in Apple iCloud Drive", + "details": { + "username": getattr(settings, "icloud_username", "Not set"), + "password": mask_sensitive_value(getattr(settings, "icloud_password", None)), + "folder": getattr(settings, "icloud_folder", "Not set"), + "cookie_directory": getattr(settings, "icloud_cookie_directory", "Not set"), + }, + } + return providers diff --git a/app/utils/settings_service.py b/app/utils/settings_service.py index 7516f8f9..2c0cf456 100644 --- a/app/utils/settings_service.py +++ b/app/utils/settings_service.py @@ -838,6 +838,39 @@ SETTING_METADATA = { "required": False, "restart_required": False, }, + # Storage Providers - iCloud Drive + "icloud_username": { + "category": "Storage Providers", + "description": "Apple ID email address for iCloud Drive authentication", + "type": "string", + "sensitive": False, + "required": False, + "restart_required": False, + }, + "icloud_password": { + "category": "Storage Providers", + "description": "App-specific password for iCloud Drive (generate at https://appleid.apple.com)", + "type": "string", + "sensitive": True, + "required": False, + "restart_required": False, + }, + "icloud_folder": { + "category": "Storage Providers", + "description": "Target folder path in iCloud Drive (e.g. Documents/Uploads)", + "type": "string", + "sensitive": False, + "required": False, + "restart_required": False, + }, + "icloud_cookie_directory": { + "category": "Storage Providers", + "description": "Directory for persisting iCloud session cookies (default: ~/.pyicloud)", + "type": "string", + "sensitive": False, + "required": False, + "restart_required": False, + }, # Storage Providers - AWS S3 "aws_access_key_id": { "category": "Storage Providers", diff --git a/app/views/onboarding.py b/app/views/onboarding.py index 3e551dbf..2b5fa2b2 100644 --- a/app/views/onboarding.py +++ b/app/views/onboarding.py @@ -27,6 +27,7 @@ _DESTINATION_META: list[dict] = [ {"id": "webdav", "name": "WebDAV", "icon": "fas fa-server"}, {"id": "sftp", "name": "SFTP", "icon": "fas fa-terminal"}, {"id": "ftp", "name": "FTP", "icon": "fas fa-server"}, + {"id": "icloud", "name": "iCloud Drive", "icon": "fab fa-apple"}, ] @@ -51,6 +52,7 @@ def _get_configured_destinations(cfg: Settings) -> list[dict]: "webdav": bool(cfg.webdav_url and cfg.webdav_username), "sftp": bool(cfg.sftp_host and cfg.sftp_username), "ftp": bool(cfg.ftp_host and cfg.ftp_username), + "icloud": bool(cfg.icloud_username and cfg.icloud_password), } return [meta for meta in _DESTINATION_META if checks.get(meta["id"], False)] diff --git a/frontend/templates/files.html b/frontend/templates/files.html index c483c0e1..4582ebc4 100644 --- a/frontend/templates/files.html +++ b/frontend/templates/files.html @@ -525,6 +525,7 @@ + diff --git a/requirements.txt b/requirements.txt index 49cca5e3..54995fb0 100644 --- a/requirements.txt +++ b/requirements.txt @@ -34,6 +34,9 @@ boto3>=1.28.0 # SFTP paramiko>=3.4.0 # SSH/SFTP implementation for Python (LGPL license) +# iCloud Drive +pyicloud>=2.4.0 # Unofficial Apple iCloud API client (MIT license) + # Safe XML parsing (protection against XML bomb / XXE attacks) defusedxml>=0.7.1 diff --git a/tests/test_send_to_all.py b/tests/test_send_to_all.py index ac05c29c..1a4bc947 100644 --- a/tests/test_send_to_all.py +++ b/tests/test_send_to_all.py @@ -300,6 +300,7 @@ class TestSendToAllDestinations: mock_sftp.return_value = False mock_email.return_value = False mock_onedrive.return_value = False + mock_icloud.return_value = False mock_dropbox_upload.delay.return_value = MagicMock(id="dropbox-task") mock_s3_upload.delay.return_value = MagicMock(id="s3-task") diff --git a/tests/test_upload_to_icloud.py b/tests/test_upload_to_icloud.py new file mode 100644 index 00000000..72e1e804 --- /dev/null +++ b/tests/test_upload_to_icloud.py @@ -0,0 +1,205 @@ +"""Unit tests for the iCloud Drive upload task and helper functions. + +Tests cover the global upload task (``upload_to_icloud``) as well as the +per-user integration handler (``_upload_icloud`` in +``upload_to_user_integration``). All external calls to ``pyicloud`` are +mocked so tests are fast, hermetic, and free of network access. +""" + +import os +from unittest.mock import MagicMock, patch + +import pytest + +# --------------------------------------------------------------------------- +# Shared helpers +# --------------------------------------------------------------------------- + +TASK_ID = "test-icloud-task-id" + + +def _write_file(path, content: bytes = b"PDF content") -> None: + """Write *content* to *path*, creating parent dirs as needed.""" + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "wb") as fh: + fh.write(content) + + +def _mock_pyicloud_module(mock_api): + """Return a mock ``pyicloud`` module whose ``PyiCloudService`` returns *mock_api*.""" + mock_mod = MagicMock() + mock_mod.PyiCloudService.return_value = mock_api + return mock_mod + + +# --------------------------------------------------------------------------- +# _get_icloud_api +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestGetIcloudApi: + """Tests for the _get_icloud_api helper.""" + + def test_returns_authenticated_api(self): + mock_api = MagicMock() + mock_api.requires_2sa = False + mock_api.requires_2fa = False + + with patch.dict("sys.modules", {"pyicloud": _mock_pyicloud_module(mock_api)}): + from app.tasks.upload_to_icloud import _get_icloud_api + + result = _get_icloud_api("user@example.com", "secret") + + assert result is mock_api + + def test_passes_cookie_directory(self): + mock_api = MagicMock() + mock_api.requires_2sa = False + mock_api.requires_2fa = False + mock_mod = _mock_pyicloud_module(mock_api) + + with patch.dict("sys.modules", {"pyicloud": mock_mod}): + from app.tasks.upload_to_icloud import _get_icloud_api + + _get_icloud_api("user@example.com", "secret", "/tmp/cookies") + + mock_mod.PyiCloudService.assert_called_once_with("user@example.com", "secret", cookie_directory="/tmp/cookies") + + def test_raises_on_2fa_required(self): + mock_api = MagicMock() + mock_api.requires_2sa = False + mock_api.requires_2fa = True + + with patch.dict("sys.modules", {"pyicloud": _mock_pyicloud_module(mock_api)}): + from app.tasks.upload_to_icloud import _get_icloud_api + + with pytest.raises(ValueError, match="two-factor authentication"): + _get_icloud_api("user@example.com", "secret") + + def test_raises_on_2sa_required(self): + mock_api = MagicMock() + mock_api.requires_2sa = True + mock_api.requires_2fa = False + + with patch.dict("sys.modules", {"pyicloud": _mock_pyicloud_module(mock_api)}): + from app.tasks.upload_to_icloud import _get_icloud_api + + with pytest.raises(ValueError, match="two-factor authentication"): + _get_icloud_api("user@example.com", "secret") + + +# --------------------------------------------------------------------------- +# _navigate_to_folder +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestNavigateToFolder: + """Tests for the _navigate_to_folder helper.""" + + def test_empty_path_returns_root(self): + from app.tasks.upload_to_icloud import _navigate_to_folder + + root = MagicMock() + result = _navigate_to_folder(root, "") + assert result is root + + def test_navigates_existing_folders(self): + from app.tasks.upload_to_icloud import _navigate_to_folder + + # Build a mock folder tree: root -> Documents -> Uploads + uploads_node = MagicMock() + uploads_node.name = "Uploads" + + docs_node = MagicMock() + docs_node.name = "Documents" + docs_node.dir.return_value = [uploads_node] + + root = MagicMock() + root.dir.return_value = [docs_node] + + result = _navigate_to_folder(root, "Documents/Uploads") + assert result is uploads_node + + def test_creates_missing_folder(self): + from app.tasks.upload_to_icloud import _navigate_to_folder + + new_folder = MagicMock() + root = MagicMock() + root.dir.return_value = [] # No children + root.mkdir.return_value = new_folder + + result = _navigate_to_folder(root, "NewFolder") + root.mkdir.assert_called_once_with("NewFolder") + assert result is new_folder + + +# --------------------------------------------------------------------------- +# _upload_icloud (user integration handler) +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +class TestUploadIcloudHandler: + """Tests for _upload_icloud handler in upload_to_user_integration.""" + + def _call(self, file_path: str, cfg: dict, creds: dict) -> dict: + from app.tasks.upload_to_user_integration import _upload_icloud + + return _upload_icloud(file_path, cfg, creds, TASK_ID) + + def test_raises_when_credentials_missing(self, tmp_path): + fp = str(tmp_path / "doc.pdf") + _write_file(fp) + with pytest.raises(ValueError, match="username or password"): + self._call(fp, {}, {}) + + def test_raises_when_password_missing(self, tmp_path): + fp = str(tmp_path / "doc.pdf") + _write_file(fp) + with pytest.raises(ValueError, match="username or password"): + self._call(fp, {}, {"username": "user@example.com"}) + + def test_successful_upload(self, tmp_path): + fp = str(tmp_path / "doc.pdf") + _write_file(fp) + + mock_api = MagicMock() + mock_api.requires_2sa = False + mock_api.requires_2fa = False + + # drive.dir() returns nothing -> mkdir will be called + mock_folder = MagicMock() + mock_api.drive.dir.return_value = [] + mock_api.drive.mkdir.return_value = mock_folder + + with patch.dict("sys.modules", {"pyicloud": _mock_pyicloud_module(mock_api)}): + result = self._call( + fp, + {"folder": "Documents"}, + {"username": "user@example.com", "password": "secret"}, + ) + + assert result["status"] == "Completed" + assert result["icloud_folder"] == "Documents" + mock_folder.upload.assert_called_once() + + def test_upload_to_root_when_no_folder(self, tmp_path): + fp = str(tmp_path / "doc.pdf") + _write_file(fp) + + mock_api = MagicMock() + mock_api.requires_2sa = False + mock_api.requires_2fa = False + + with patch.dict("sys.modules", {"pyicloud": _mock_pyicloud_module(mock_api)}): + result = self._call( + fp, + {}, + {"username": "user@example.com", "password": "secret"}, + ) + + assert result["status"] == "Completed" + assert result["icloud_folder"] == "/" + mock_api.drive.upload.assert_called_once()