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>
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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)]
|
||||
|
||||
|
||||
@@ -525,6 +525,7 @@
|
||||
<option value="webdav" {% if storage_provider == "webdav" %}selected{% endif %}>WebDAV</option>
|
||||
<option value="ftp" {% if storage_provider == "ftp" %}selected{% endif %}>FTP</option>
|
||||
<option value="sftp" {% if storage_provider == "sftp" %}selected{% endif %}>SFTP</option>
|
||||
<option value="icloud" {% if storage_provider == "icloud" %}selected{% endif %}>iCloud Drive</option>
|
||||
</select>
|
||||
</div>
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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")
|
||||
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user