diff --git a/NOTIFICATION_CONFIG.md b/NOTIFICATION_CONFIG.md index 198786478..15eead98b 100644 --- a/NOTIFICATION_CONFIG.md +++ b/NOTIFICATION_CONFIG.md @@ -267,11 +267,14 @@ app: ### Testing Configuration #### Dry Run Mode -Use dry run to test notifications without sending: -```bash -# This will test notification configuration without actually sending -docker exec -it icloud /bin/sh -c "su-exec abc python src/main.py --dry-run" -``` +Note: ``--dry-run`` (added in this PR / feat/dry-run) is a sync-side +pre-flight that authenticates, summarises what *would* be synced, and +exits without writing files or sending notifications. It is NOT a +notification-test mode — notifications are intentionally suppressed +in dry-run. + +To test notification delivery, use the manual-testing helpers below +(``notify._send_discord_no_throttle`` etc.) rather than ``--dry-run``. #### Manual Testing Test individual notification services: diff --git a/src/main.py b/src/main.py index cf34ba604..8b1abd332 100644 --- a/src/main.py +++ b/src/main.py @@ -2,7 +2,55 @@ __author__ = "Mandar Patil (mandarons@pm.me)" +import argparse + from src import sync if __name__ == "__main__": - sync.sync() + parser = argparse.ArgumentParser( + prog="icloud-docker", + description="iCloud Drive + Photos backup loop. See config.yaml for runtime settings.", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help=( + "Authenticate, summarise what would be synced, then exit " + "without downloading or modifying any files. Useful for " + "verifying credentials + mount paths + config before the " + "real sync loop is allowed to run." + ), + ) + parser.add_argument( + "--check-files", + type=int, + default=None, + metavar="N", + help=( + "Only meaningful with --dry-run. Walks N photos per library " + "and reports per-library counts of would_skip / size_mismatch " + "/ not_found / error against your on-disk tree. Use this " + "BEFORE a real sync to confirm a boredazfcuk → mandarons (or " + "any cross-tool) migration will recognise existing files " + "instead of re-downloading them. Pass 0 to walk every photo " + "(slow on large libraries — recommend 50–200 first)." + ), + ) + args = parser.parse_args() + + # Validate the --check-files / --dry-run combination before handing off + # to sync.sync(). Without these guards, `--check-files 10` (no + # --dry-run) starts the normal sync loop and silently ignores the + # flag, and `--check-files -1` is treated as "walk everything" by + # the migration walkers (since `if sample > 0` is the cap-check) — + # both of which trip up users expecting fail-fast feedback. + if args.check_files is not None: + if not args.dry_run: + parser.error("--check-files requires --dry-run") + if args.check_files < 0: + parser.error( + "--check-files must be a non-negative integer " + "(0 means walk everything, N > 0 caps the walk at N)", + ) + + sync.sync(dry_run=args.dry_run, check_files=args.check_files) diff --git a/src/migration_check.py b/src/migration_check.py new file mode 100644 index 000000000..eeb365712 --- /dev/null +++ b/src/migration_check.py @@ -0,0 +1,404 @@ +"""Per-file dry-run validation — does mandarons see the right files? + +Walks each iCloud photo library AND iCloud Drive and, for a sample +(or all) items, computes the on-disk path mandarons WOULD write to +using the live config (library_destinations, folder_format, +filename_format for photos; mirror-tree for drive) and checks whether +the file is already there with the matching size. + +Used by ``--dry-run --check-files`` from ``main.py`` so users +migrating from a different downloader (e.g. boredazfcuk's +icloud_photos_downloader) can confirm the size-based existence check +will actually find their existing files BEFORE mandarons launches a +real sync. Without this check, a single misconfiguration — +filename_format wrong, folder_format missing, library_destinations +mapping a non-existent key, drive destination pointing at the wrong +mount — silently triggers a full re-download of the user's entire +library. + +Pure read: no downloads, no keyring writes, no cookie writes. +""" + +__author__ = "Mandar Patil (mandarons@pm.me)" + +import os +import unicodedata +from pathlib import Path +from typing import Any +from urllib.parse import unquote + +from src import config_parser, get_logger +from src.photo_path_utils import ( + create_folder_path_if_needed, + generate_photo_filename_with_metadata, +) + +# These two symbols ship in companion PRs (``feat/photos-filename-format-simple`` +# and ``feat/photos-library-destinations``). When those land first the +# real functions are used; when this PR is reviewed/merged in isolation +# the no-op fallbacks let the suite import + run, so the migration-check +# at least authenticates and walks the libraries (just without per- +# library subdirectories or simple-filename naming). The dry-run still +# reports something useful: "would mandarons-default paths line up with +# what's on disk?" +try: + from src.photo_path_utils import set_default_filename_format # type: ignore[attr-defined] +except ( + ImportError +): # pragma: no cover — only when feat/photos-filename-format-simple isn't merged + + def set_default_filename_format(_filename_format: str) -> None: + """No-op fallback — feat/photos-filename-format-simple not merged.""" + + +try: + from src.sync_photos import _library_destination # type: ignore[attr-defined] +except ( + ImportError +): # pragma: no cover — only when feat/photos-library-destinations isn't merged + + def _library_destination( + base_destination: str, + library: str, + library_destinations: dict, + ) -> str: + """Fallback that always returns the base destination. + + feat/photos-library-destinations introduces per-library subdir + mapping; without it, mandarons writes every library to the + single base destination. The migration-check reports against + that same path. + """ + return base_destination + + +LOGGER = get_logger() + + +def _check_one_photo( + photo, + library_dest: str, + folder_format: str | None, +) -> tuple[str, str, int, int]: + """Compute target path + status for a single photo. Returns + ``(status, path, expected_size, actual_size)`` where ``status`` is + one of ``would_skip`` / ``size_mismatch`` / ``not_found`` / + ``error``.""" + try: + file_size = "original" + if file_size not in photo.versions: + return "error", "", 0, 0 + folder_path = create_folder_path_if_needed(library_dest, folder_format, photo) + filename = generate_photo_filename_with_metadata(photo, file_size) + target_path = os.path.join(folder_path, filename) + expected = int(photo.versions[file_size]["size"]) + except Exception as e: + LOGGER.debug( + f"check_migration: failed to compute path for {getattr(photo, 'filename', '?')}: {e!s}", + ) + return "error", "", 0, 0 + + if not os.path.isfile(target_path): + return "not_found", target_path, expected, 0 + try: + actual = os.path.getsize(target_path) + except OSError: + return "error", target_path, expected, 0 + if actual == expected: + return "would_skip", target_path, expected, actual + return "size_mismatch", target_path, expected, actual + + +def check_library( + library, + library_name: str, + photos_base: str, + mapping: dict, + folder_format: str | None, + sample: int, +) -> dict[str, Any]: + """Walk a single library and accumulate per-status counters. + + ``sample=0`` walks every photo (slow on large libraries). + ``sample>0`` walks the first N (newest-first per icloudpy's + iterator). Pagination cost is proportional to the number of + photos walked, so a sample of 200 is usually sub-minute; a sample + of 5000 can take several minutes on a 100K-photo library. + + Note on bias: iCloud's iterator is newest-first, so a small N + skews toward recent photos and gives you no signal about whether + older files (which a migration tool would have downloaded years + ago) will match. Use a sample size proportional to the time + range you care about validating. + """ + library_dest = _library_destination(photos_base, library_name, mapping) + # Match the real sync's default path layout: ``_sync_all_photos_in_library`` + # in src/sync_photos.py iterates ``library.all`` (not + # ``library.albums["All Photos"]``) and writes under + # ``/all//``. Earlier versions + # of this checker walked the album view and reported existing files + # as ``not_found`` because the on-disk path it computed was missing + # the trailing ``/all/`` segment. + check_dest = os.path.join(library_dest, "all") + stats = {"would_skip": 0, "size_mismatch": 0, "not_found": 0, "error": 0} + samples = {"would_skip": [], "size_mismatch": [], "not_found": []} + + seen = 0 + checked = 0 + try: + for photo in library.all: + if sample > 0 and checked >= sample: + break + seen += 1 + checked += 1 + status, path, expected, actual = _check_one_photo( + photo, + check_dest, + folder_format, + ) + stats[status] = stats.get(status, 0) + 1 + if status in samples and len(samples[status]) < 3: + if status == "size_mismatch": + samples[status].append((path, expected, actual)) + else: + samples[status].append((path, expected)) + except Exception as e: + LOGGER.warning(f"check_migration: walk of {library_name} stopped early: {e!s}") + + return { + "library_dest": library_dest, + "checked": checked, + "seen": seen, + "stats": stats, + "samples": samples, + } + + +def check_migration(api, config: dict, sample: int = 0) -> dict[str, Any]: + """Walk every photo library and report what a real sync would do. + + Args: + api: Authenticated ICloudPyService instance. + config: Live config dict (read_config output). + sample: Photos per library to check. ``0`` means all (slow on + large libraries — only use after a small-sample run has + confirmed the mapping looks right). + + Returns: + Dict keyed by library name → per-library result dict (see + ``check_library``). + """ + # Read-only path resolution — never call prepare_*_destination from + # the dry-run path. Those helpers ``os.makedirs`` the destination, + # which would leave stub directories on disk if the user + # misconfigured the mount they're trying to validate (exactly what + # --dry-run is supposed to catch BEFORE writing anything). + photos_base = os.path.join( + config_parser.get_root_destination_path(config=config), + config_parser.get_photos_destination_path(config=config), + ) + # Defensive: feat/photos-library-destinations may not be merged yet. + # Falls back to an empty mapping (every library writes to the same + # photos_base) so this PR is independent of PR 3. + if hasattr(config_parser, "get_photos_library_destinations"): + mapping = config_parser.get_photos_library_destinations(config=config) + else: # pragma: no cover — only when feat/photos-library-destinations isn't merged + mapping = {} + folder_format = config_parser.get_photos_folder_format(config=config) + + # mandarons' sync_photos.sync_photos() normally sets this singleton + # via set_default_filename_format(). When the user invokes us via + # --dry-run we never go through that path, so we have to set it + # ourselves — otherwise every call to + # generate_photo_filename_with_metadata returns the legacy metadata- + # style name regardless of the config setting. + filename_format = (config.get("photos", {}) or {}).get("filename_format") + if filename_format: + set_default_filename_format(filename_format) + + results: dict[str, Any] = {} + for library_name in api.photos.libraries: + LOGGER.info( + f"check_migration: walking library {library_name} (sample={sample or 'all'}) ...", + ) + library = api.photos.libraries[library_name] + results[library_name] = check_library( + library=library, + library_name=library_name, + photos_base=photos_base, + mapping=mapping, + folder_format=folder_format, + sample=sample, + ) + return results + + +def _check_one_drive_file(item, local_path: str) -> tuple[str, str, int, int]: + """Compute status for a single Drive file item. + + Returns ``(status, path, expected_size, actual_size)`` where + ``status`` is one of ``would_skip`` / ``size_mismatch`` / + ``not_found`` / ``error``. + + Handles BOTH on-disk forms a Drive item can take: + - regular file (most items, plus packages that mandarons couldn't + unpack like .key / .jmb — bytes saved flat) + - directory tree (packages mandarons successfully unpacked, e.g. + .band GarageBand projects) + + Size comparison matches mandarons' real-sync ``file_exists`` / + ``package_exists`` semantics: flat-file size for regular files, + sum of contained file sizes for directory packages. + """ + try: + expected = int(item.size) if getattr(item, "size", None) is not None else 0 + except Exception as e: + LOGGER.debug( + f"check_migration: drive item bad size for {getattr(item, 'name', '?')}: {e!s}", + ) + return "error", local_path, 0, 0 + + if os.path.isdir(local_path): + try: + actual = sum( + f.stat().st_size for f in Path(local_path).glob("**/*") if f.is_file() + ) + except OSError: + return "error", local_path, expected, 0 + elif os.path.isfile(local_path): + try: + actual = os.path.getsize(local_path) + except OSError: + return "error", local_path, expected, 0 + else: + return "not_found", local_path, expected, 0 + + if actual == expected: + return "would_skip", local_path, expected, actual + return "size_mismatch", local_path, expected, actual + + +def _walk_drive_recursive( + folder, + destination_path: str, + sample: int, + state: dict, +) -> None: + """Recursively walk a Drive folder, mutating ``state`` in place. + + ``state`` shape: + {'checked': int, 'stats': {...}, 'samples': {...}} + + ``sample > 0`` caps the total file count walked (depth-first across + the tree). ``sample == 0`` walks everything. + + Folders are followed unconditionally; only file items count against + the sample cap (folders themselves aren't validated against disk). + """ + if sample > 0 and state["checked"] >= sample: + return + try: + items_index = folder.dir() + except Exception as e: + LOGGER.debug(f"check_migration: drive folder dir() failed: {e!s}") + return + if not items_index: + return + + for name in items_index: + if sample > 0 and state["checked"] >= sample: + return + try: + item = folder[name] + except Exception as e: + LOGGER.debug(f"check_migration: drive item access failed for {name}: {e!s}") + state["stats"]["error"] = state["stats"].get("error", 0) + 1 + continue + + item_type = getattr(item, "type", None) + if item_type in ("folder", "app_library"): + try: + decoded = unquote(getattr(item, "name", name)) + except Exception: + decoded = name + sub_dest = unicodedata.normalize( + "NFC", + os.path.join(destination_path, decoded), + ) + _walk_drive_recursive(item, sub_dest, sample, state) + elif item_type == "file": + try: + decoded = unquote(getattr(item, "name", name)) + except Exception: + decoded = name + local_path = unicodedata.normalize( + "NFC", + os.path.join(destination_path, decoded), + ) + status, path, expected, actual = _check_one_drive_file(item, local_path) + state["stats"][status] = state["stats"].get(status, 0) + 1 + state["checked"] += 1 + if status in state["samples"] and len(state["samples"][status]) < 3: + if status == "size_mismatch": + state["samples"][status].append((path, expected, actual)) + else: + state["samples"][status].append((path, expected)) + + +def check_drive(drive, drive_destination: str, sample: int) -> dict[str, Any]: + """Walk iCloud Drive and report per-file dry-run status. + + Args: + drive: ``api.drive`` (root drive node from icloudpy). + drive_destination: Local path where mandarons would write Drive + content (matches what ``sync_drive`` uses). + sample: ``0`` walks every file; ``N > 0`` walks up to N files + total (depth-first across the folder tree). A small N gives + quick sanity feedback; a large N (or 0) gives statistical + confidence at the cost of pagination time. + + Returns: + Dict with keys: ``drive_destination``, ``checked``, ``stats``, + ``samples``. Same shape as ``check_library`` minus the + library-specific fields. + """ + state: dict[str, Any] = { + "checked": 0, + "stats": {"would_skip": 0, "size_mismatch": 0, "not_found": 0, "error": 0}, + "samples": {"would_skip": [], "size_mismatch": [], "not_found": []}, + } + try: + _walk_drive_recursive(drive, drive_destination, sample, state) + except Exception as e: + LOGGER.warning(f"check_migration: drive walk stopped early: {e!s}") + return { + "drive_destination": drive_destination, + "checked": state["checked"], + "stats": state["stats"], + "samples": state["samples"], + } + + +def check_drive_migration(api, config: dict, sample: int = 0) -> dict[str, Any] | None: + """Wrapper that resolves drive destination from config then walks. + + Returns None if there's no ``drive:`` section in the config — caller + treats absence as "drive sync would be skipped at real-sync time." + """ + if "drive" not in (config or {}): + return None + try: + # Read-only resolution — see check_migration above for rationale. + drive_destination = os.path.join( + config_parser.get_root_destination_path(config=config), + config_parser.get_drive_destination_path(config=config), + ) + except Exception as e: + LOGGER.warning(f"check_migration: drive destination resolution failed: {e!s}") + return None + LOGGER.info(f"check_migration: walking iCloud Drive (sample={sample or 'all'}) ...") + return check_drive( + drive=api.drive, + drive_destination=drive_destination, + sample=sample, + ) diff --git a/src/sync.py b/src/sync.py index 3dec0ce37..bff9698fd 100644 --- a/src/sync.py +++ b/src/sync.py @@ -120,9 +120,13 @@ def _extract_sync_intervals(config, log_messages: bool = False): photos_sync_interval = 0 if config and "drive" in config: - drive_sync_interval = config_parser.get_drive_sync_interval(config=config, log_messages=log_messages) + drive_sync_interval = config_parser.get_drive_sync_interval( + config=config, log_messages=log_messages, + ) if config and "photos" in config: - photos_sync_interval = config_parser.get_photos_sync_interval(config=config, log_messages=log_messages) + photos_sync_interval = config_parser.get_photos_sync_interval( + config=config, log_messages=log_messages, + ) return drive_sync_interval, photos_sync_interval @@ -164,7 +168,9 @@ def _authenticate_and_get_api(config, username: str): """ server_region = config_parser.get_region(config=config) password = _retrieve_password(username) - return get_api_instance(username=username, password=password, server_region=server_region) + return get_api_instance( + username=username, password=password, server_region=server_region, + ) def _perform_drive_sync(config, api, sync_state: SyncState, drive_sync_interval: int): @@ -289,9 +295,13 @@ def _perform_photos_sync(config, api, sync_state: SyncState, photos_sync_interva stats.photos_downloaded = len(new_files) # Estimate hardlinked photos (approximate) - use_hardlinks = config_parser.get_photos_use_hardlinks(config=config, log_messages=False) + use_hardlinks = config_parser.get_photos_use_hardlinks( + config=config, log_messages=False, + ) if use_hardlinks: - stats.photos_hardlinked = max(0, len(files_after) - len(files_before) - stats.photos_downloaded) + stats.photos_hardlinked = max( + 0, len(files_after) - len(files_before) - stats.photos_downloaded, + ) # Count skipped photos stats.photos_skipped = len(files_before & files_after) @@ -331,6 +341,150 @@ def _perform_photos_sync(config, api, sync_state: SyncState, photos_sync_interva return None +def _perform_dry_run(config, api, check_files: int | None = None) -> None: + """Authenticate-and-enumerate path used when ``--dry-run`` is passed. + + Verifies that the configured credentials, mount paths, and iCloud-side + state are all in working order WITHOUT writing or downloading any + files. Designed as the safety check users run before letting the real + sync loop loose on a new install. + + Logs (at INFO level): + - Drive destination path + root-level item count (when Drive is configured) + - Photos destination path + library names (when Photos is configured) + - When ``check_files`` is not None: per-library would-skip / + size-mismatch / not-found counts (see ``migration_check``). + + Args: + config: Configuration dictionary + api: Authenticated iCloud API instance + check_files: When set (``--check-files=N``), additionally walks + up to N photos per library and reports what a real sync + would do per file. ``0`` walks every photo. ``None`` skips + this check (cheap default for ``--dry-run`` alone). + + Notifications, usage statistics, file writes, file deletions, and the + sync loop itself are all skipped. + """ + LOGGER.info("DRY RUN: authentication succeeded — verifying configured services.") + + if config and "drive" in config: + try: + drive_destination = config_parser.get_drive_destination_path(config=config) + LOGGER.info(f"DRY RUN: Drive destination: {drive_destination}") + root_items = list(api.drive.dir()) + LOGGER.info( + f"DRY RUN: Drive root contains {len(root_items)} item(s) — " + "real sync would walk this tree per `drive.filters`.", + ) + except Exception as e: + LOGGER.warning(f"DRY RUN: Drive enumeration failed: {e!s}") + else: + LOGGER.info( + "DRY RUN: no `drive:` section in config — Drive sync would be skipped.", + ) + + if config and "photos" in config: + try: + photos_destination = config_parser.get_photos_destination_path( + config=config, + ) + LOGGER.info(f"DRY RUN: Photos destination: {photos_destination}") + libraries = ( + list(api.photos.libraries.keys()) + if hasattr(api.photos, "libraries") + else [] + ) + if libraries: + LOGGER.info( + f"DRY RUN: Photos libraries available: {', '.join(libraries)}", + ) + else: + LOGGER.info("DRY RUN: Photos libraries: (none reported by iCloud)") + except Exception as e: + LOGGER.warning(f"DRY RUN: Photos enumeration failed: {e!s}") + else: + LOGGER.info( + "DRY RUN: no `photos:` section in config — Photos sync would be skipped.", + ) + + if check_files is not None: + from src import migration_check + + # Photos walker — per-library counts using mandarons' real path/size + # logic so the report mirrors what a real sync would skip vs download. + if config and "photos" in config: + try: + LOGGER.info( + f"DRY RUN: walking photos for file-existence check " + f"(--check-files={'all' if check_files == 0 else check_files} per library) ...", + ) + results = migration_check.check_migration( + api=api, config=config, sample=check_files, + ) + for library_name, result in results.items(): + stats = result["stats"] + LOGGER.info( + f"DRY RUN: {library_name} (dest {result['library_dest']}): " + f"sampled={result['checked']} " + f"would_skip={stats['would_skip']} " + f"size_mismatch={stats['size_mismatch']} " + f"not_found={stats['not_found']} " + f"errors={stats['error']}", + ) + for status, items in result["samples"].items(): + for item in items: + if status == "size_mismatch": + path, expected, actual = item + LOGGER.info( + f"DRY RUN: sample {status}: {path} (have {actual:,}b, want {expected:,}b)", + ) + else: + path, expected = item + LOGGER.info( + f"DRY RUN: sample {status}: {path} ({expected:,}b)", + ) + except Exception as e: + LOGGER.warning(f"DRY RUN: photos check-files walk failed: {e!s}") + + # Drive walker — same per-file would_skip/size_mismatch/not_found + # report, but walking the Drive tree (no library_destinations, + # mirror-tree layout). Catches misconfigured drive.destination. + if config and "drive" in config: + try: + drive_result = migration_check.check_drive_migration( + api=api, config=config, sample=check_files, + ) + if drive_result is not None: + stats = drive_result["stats"] + LOGGER.info( + f"DRY RUN: Drive (dest {drive_result['drive_destination']}): " + f"sampled={drive_result['checked']} " + f"would_skip={stats['would_skip']} " + f"size_mismatch={stats['size_mismatch']} " + f"not_found={stats['not_found']} " + f"errors={stats['error']}", + ) + for status, items in drive_result["samples"].items(): + for item in items: + if status == "size_mismatch": + path, expected, actual = item + LOGGER.info( + f"DRY RUN: sample {status}: {path} (have {actual:,}b, want {expected:,}b)", + ) + else: + path, expected = item + LOGGER.info( + f"DRY RUN: sample {status}: {path} ({expected:,}b)", + ) + except Exception as e: + LOGGER.warning(f"DRY RUN: drive check-files walk failed: {e!s}") + + LOGGER.info( + "DRY RUN complete — no files were written. Re-run without --dry-run to sync.", + ) + + def _check_services_configured(config): """ Check if any sync services are configured. @@ -356,10 +510,16 @@ def _send_usage_statistics(config, summary: SyncSummary) -> None: # Create anonymized usage data usage_data = { "sync_duration": ( - (summary.sync_end_time - summary.sync_start_time).total_seconds() if summary.sync_end_time else 0 + (summary.sync_end_time - summary.sync_start_time).total_seconds() + if summary.sync_end_time + else 0 + ), + "has_drive_activity": bool( + summary.drive_stats and summary.drive_stats.has_activity(), + ), + "has_photos_activity": bool( + summary.photo_stats and summary.photo_stats.has_activity(), ), - "has_drive_activity": bool(summary.drive_stats and summary.drive_stats.has_activity()), - "has_photos_activity": bool(summary.photo_stats and summary.photo_stats.has_activity()), "has_errors": summary.has_errors(), "timestamp": (summary.sync_end_time.isoformat() if summary.sync_end_time else None), } @@ -427,7 +587,9 @@ def _handle_password_error(config, username: str, sync_state: SyncState): Returns: bool: True if should continue (retry), False if should exit """ - LOGGER.error("Password is not stored in keyring. Please save the password in keyring.") + LOGGER.error( + "Password is not stored in keyring. Please save the password in keyring.", + ) sleep_for = config_parser.get_retry_login_interval(config=config) if sleep_for < 0: @@ -453,7 +615,9 @@ def _log_retry_time(sleep_for: int): Args: sleep_for: Sleep duration in seconds """ - next_sync = (datetime.datetime.now() + datetime.timedelta(seconds=sleep_for)).strftime("%c") + next_sync = ( + datetime.datetime.now() + datetime.timedelta(seconds=sleep_for) + ).strftime("%c") LOGGER.info(f"Retrying login at {next_sync} ...") @@ -482,15 +646,24 @@ def _calculate_next_sync_schedule(config, sync_state: SyncState): sleep_for = sync_state.drive_time_remaining sync_state.enable_sync_drive = True sync_state.enable_sync_photos = False - elif has_drive and has_photos and sync_state.drive_time_remaining <= sync_state.photos_time_remaining: + elif ( + has_drive + and has_photos + and sync_state.drive_time_remaining <= sync_state.photos_time_remaining + ): # Special case: if both timers are equal and large (> 10 seconds), wait for the full interval # This fixes the bug where equal large intervals cause immediate re-sync - if sync_state.drive_time_remaining == sync_state.photos_time_remaining and sync_state.drive_time_remaining > 10: + if ( + sync_state.drive_time_remaining == sync_state.photos_time_remaining + and sync_state.drive_time_remaining > 10 + ): sleep_for = sync_state.drive_time_remaining sync_state.enable_sync_drive = True sync_state.enable_sync_photos = True else: - sleep_for = sync_state.photos_time_remaining - sync_state.drive_time_remaining + sleep_for = ( + sync_state.photos_time_remaining - sync_state.drive_time_remaining + ) sync_state.photos_time_remaining -= sync_state.drive_time_remaining sync_state.enable_sync_drive = True sync_state.enable_sync_photos = False @@ -510,7 +683,9 @@ def _log_next_sync_time(sleep_for: int): Args: sleep_for: Sleep duration in seconds """ - next_sync = (datetime.datetime.now() + datetime.timedelta(seconds=sleep_for)).strftime("%c") + next_sync = ( + datetime.datetime.now() + datetime.timedelta(seconds=sleep_for) + ).strftime("%c") LOGGER.info(f"Resyncing at {next_sync} ...") @@ -550,40 +725,74 @@ def _should_exit_oneshot_mode(config): return should_exit_drive and should_exit_photos -def sync(): +def sync(dry_run: bool = False, check_files: int | None = None): """ Main synchronization loop. Orchestrates the entire sync process by delegating specific responsibilities to focused helper functions. This function coordinates the high-level flow while each helper handles a single concern. + + Args: + dry_run: When True, authenticate and summarise what would be synced, + then exit without writing files, sending notifications, or + entering the sync loop. Useful for verifying credentials, mount + paths, and config before the real loop starts downloading. + check_files: Optional sample size for the per-photo file-existence + check during dry-run. Only meaningful with ``dry_run=True``. + ``None`` skips the check (cheap default). ``0`` walks every + photo (slow on large libraries). Positive N walks N + stride-sampled photos per library. """ sync_state = SyncState() startup_logged = False while True: config = _load_configuration() - alive(config=config) + # Skip telemetry on dry-run: ``alive()`` registers the installation + # and saves the usage cache to disk, both of which violate the + # "no side effects" dry-run contract. (The real sync loop still + # calls it on every iteration as before.) + if not dry_run: + alive(config=config) # Log sync intervals once at startup if not startup_logged: _log_sync_intervals_at_startup(config) startup_logged = True - drive_sync_interval, photos_sync_interval = _extract_sync_intervals(config, log_messages=False) + drive_sync_interval, photos_sync_interval = _extract_sync_intervals( + config, log_messages=False, + ) username = config_parser.get_username(config=config) if config else None if username: try: api = _authenticate_and_get_api(config, username) + # Dry-run path: authenticate, enumerate, log, exit. + # Skips the entire sync + notification + retry pipeline. + if dry_run: + if api.requires_2sa: + LOGGER.info( + "DRY RUN: 2FA required — finish interactive auth first " + "(see README), then re-run with --dry-run.", + ) + else: + _perform_dry_run(config, api, check_files=check_files) + return + if not api.requires_2sa: # Create summary for this sync cycle summary = SyncSummary() # Perform syncs and collect statistics - drive_stats = _perform_drive_sync(config, api, sync_state, drive_sync_interval) - photos_stats = _perform_photos_sync(config, api, sync_state, photos_sync_interval) + drive_stats = _perform_drive_sync( + config, api, sync_state, drive_sync_interval, + ) + photos_stats = _perform_photos_sync( + config, api, sync_state, photos_sync_interval, + ) # Populate summary with statistics summary.drive_stats = drive_stats @@ -605,7 +814,9 @@ def sync(): should_send_notification = False if has_drive_config and has_photos_config: # Both services configured - send notification only when both have synced - should_send_notification = drive_stats is not None and photos_stats is not None + should_send_notification = ( + drive_stats is not None and photos_stats is not None + ) elif has_drive_config and not has_photos_config: # Only drive configured - send when drive synced should_send_notification = drive_stats is not None @@ -617,10 +828,14 @@ def sync(): try: notify.send_sync_summary(config=config, summary=summary) except Exception as e: - LOGGER.debug(f"Failed to send sync summary notification: {e!s}") + LOGGER.debug( + f"Failed to send sync summary notification: {e!s}", + ) if not _check_services_configured(config): - LOGGER.warning("Nothing to sync. Please add drive: and/or photos: section in config.yaml file.") + LOGGER.warning( + "Nothing to sync. Please add drive: and/or photos: section in config.yaml file.", + ) else: if not _handle_2fa_required(config, username, sync_state): break @@ -635,7 +850,9 @@ def sync(): _log_next_sync_time(sleep_for) if _should_exit_oneshot_mode(config): - LOGGER.info("All configured sync intervals are negative, exiting oneshot mode...") + LOGGER.info( + "All configured sync intervals are negative, exiting oneshot mode...", + ) break sleep(sleep_for) diff --git a/tests/test_dry_run.py b/tests/test_dry_run.py new file mode 100644 index 000000000..8260d9221 --- /dev/null +++ b/tests/test_dry_run.py @@ -0,0 +1,428 @@ +"""Tests for the ``--dry-run`` mode (sync.sync(dry_run=True) and the +``_perform_dry_run`` helper).""" + +__author__ = "Mandar Patil (mandarons@pm.me)" + +import logging +import os +import shutil +import unittest +from unittest.mock import MagicMock, patch + +import tests +from src import read_config, sync +from tests import data + + +class TestDryRunPerform(unittest.TestCase): + """Direct tests of ``sync._perform_dry_run`` without going through ``sync.sync``.""" + + def setUp(self): + """Load the test config + a temp root each test.""" + config = read_config(config_path=tests.CONFIG_PATH) + assert isinstance(config, dict) + self.config = config + self.config["app"]["root"] = tests.TEMP_DIR + os.makedirs(tests.TEMP_DIR, exist_ok=True) + + def tearDown(self): + """Remove temp dirs.""" + if os.path.exists(tests.TEMP_DIR): + shutil.rmtree(tests.TEMP_DIR) + + def _make_api(self, drive_items=None, photos_libraries=None): + """Build a minimal API mock with .drive.dir() and .photos.libraries.""" + api = MagicMock() + api.drive.dir.return_value = drive_items or [] + api.photos.libraries = photos_libraries or {} + return api + + def test_drive_destination_and_root_count_logged(self): + """When Drive is configured, log destination path + root item count.""" + api = self._make_api(drive_items=["dir1", "dir2", "file.txt"]) + with self.assertLogs(sync.LOGGER, level=logging.INFO) as cm: + sync._perform_dry_run(config=self.config, api=api) # noqa: SLF001 + joined = "\n".join(cm.output) + self.assertIn("DRY RUN: Drive destination:", joined) + self.assertIn("Drive root contains 3 item(s)", joined) + + def test_photos_libraries_logged(self): + """When Photos is configured, log destination path + library names.""" + api = self._make_api( + photos_libraries={"PrimarySync": object(), "SharedLibrary": object()}, + ) + with self.assertLogs(sync.LOGGER, level=logging.INFO) as cm: + sync._perform_dry_run(config=self.config, api=api) # noqa: SLF001 + joined = "\n".join(cm.output) + self.assertIn("DRY RUN: Photos destination:", joined) + self.assertIn("PrimarySync", joined) + self.assertIn("SharedLibrary", joined) + + def test_drive_enumeration_failure_is_non_fatal(self): + """Exception inside drive.dir() is caught and logged as a warning.""" + api = MagicMock() + api.drive.dir.side_effect = RuntimeError("boom") + api.photos.libraries = {} + with self.assertLogs(sync.LOGGER, level=logging.WARNING) as cm: + sync._perform_dry_run(config=self.config, api=api) # noqa: SLF001 + joined = "\n".join(cm.output) + self.assertIn("Drive enumeration failed", joined) + self.assertIn("boom", joined) + + def test_photos_enumeration_failure_is_non_fatal(self): + """Exception while accessing photos.libraries is caught and logged.""" + + class BoomPhotos: + @property + def libraries(self): + raise RuntimeError("photos boom") # noqa: EM101 + + api = self._make_api(drive_items=[]) + api.photos = BoomPhotos() + with self.assertLogs(sync.LOGGER, level=logging.WARNING) as cm: + sync._perform_dry_run(config=self.config, api=api) # noqa: SLF001 + joined = "\n".join(cm.output) + self.assertIn("Photos enumeration failed", joined) + + def test_completion_line_logged(self): + """Final 'DRY RUN complete' line is always emitted.""" + api = self._make_api() + with self.assertLogs(sync.LOGGER, level=logging.INFO) as cm: + sync._perform_dry_run(config=self.config, api=api) # noqa: SLF001 + joined = "\n".join(cm.output) + self.assertIn("DRY RUN complete", joined) + self.assertIn("--dry-run", joined) + + def test_skipped_services_announced(self): + """When a service is not configured, log that it would be skipped.""" + config = {"app": dict(self.config["app"])} # no drive, no photos + api = self._make_api() + with self.assertLogs(sync.LOGGER, level=logging.INFO) as cm: + sync._perform_dry_run(config=config, api=api) # noqa: SLF001 + joined = "\n".join(cm.output) + self.assertIn("no `drive:` section in config", joined) + self.assertIn("no `photos:` section in config", joined) + + +class TestDryRunCheckFilesIntegration(unittest.TestCase): + """``--check-files N`` triggers the migration_check walk for both + photos and drive. Tests the integration layer in ``_perform_dry_run`` + that wires args.check_files through to migration_check.check_migration + and migration_check.check_drive_migration, then formats the per-status + sample lines for the operator's log.""" + + def setUp(self): + config = read_config(config_path=tests.CONFIG_PATH) + assert isinstance(config, dict) + self.config = config + self.config["app"]["root"] = tests.TEMP_DIR + os.makedirs(tests.TEMP_DIR, exist_ok=True) + + def tearDown(self): + if os.path.exists(tests.TEMP_DIR): + shutil.rmtree(tests.TEMP_DIR) + + def _photos_result(self): + """Fake check_migration result with one library + 3 sample + statuses so every sample-line formatter branch is exercised.""" + return { + "PrimarySync": { + "library_dest": "/photos/personal", + "checked": 5, + "stats": { + "would_skip": 2, + "size_mismatch": 1, + "not_found": 1, + "error": 1, + }, + "samples": { + "would_skip": [("/photos/personal/IMG_1.HEIC", 1000)], + "size_mismatch": [ + ("/photos/personal/IMG_2.HEIC", 1000, 800), + ], + "not_found": [("/photos/personal/IMG_3.HEIC", 500)], + }, + }, + } + + def _drive_result(self): + return { + "drive_destination": "/icloud/drive", + "checked": 4, + "stats": { + "would_skip": 1, + "size_mismatch": 1, + "not_found": 1, + "error": 1, + }, + "samples": { + "would_skip": [("/icloud/drive/a.txt", 100)], + "size_mismatch": [("/icloud/drive/b.txt", 200, 150)], + "not_found": [("/icloud/drive/c.txt", 300)], + }, + } + + def test_check_files_invokes_both_walkers_and_logs_per_status(self): + """Happy path: check_files=10 → call check_migration AND + check_drive_migration, then log per-library + per-drive stats + AND each sample line (would_skip/size_mismatch/not_found).""" + from unittest.mock import MagicMock, patch + + api = MagicMock() + with patch.object( + sync, + "config_parser", + ) as fake_cp, patch( + "src.migration_check.check_migration", + return_value=self._photos_result(), + ) as fake_photos, patch( + "src.migration_check.check_drive_migration", + return_value=self._drive_result(), + ) as fake_drive, self.assertLogs( + sync.LOGGER, level=logging.INFO, + ) as cm: + fake_cp.prepare_drive_destination.return_value = "/icloud/drive" + fake_cp.prepare_photos_destination.return_value = "/icloud/photos" + sync._perform_dry_run( # noqa: SLF001 + config=self.config, + api=api, + check_files=10, + ) + fake_photos.assert_called_once_with(api=api, config=self.config, sample=10) + fake_drive.assert_called_once_with(api=api, config=self.config, sample=10) + joined = "\n".join(cm.output) + self.assertIn("walking photos for file-existence check", joined) + self.assertIn("PrimarySync", joined) + self.assertIn("would_skip=2", joined) + # Sample formatter lines: + self.assertIn("sample would_skip:", joined) + self.assertIn("sample size_mismatch:", joined) + self.assertIn("sample not_found:", joined) + # Drive section: + self.assertIn("sampled=4", joined) + + def test_check_files_zero_means_walk_all(self): + """check_files=0 is passed straight through and the log says + 'all' instead of a number.""" + from unittest.mock import MagicMock, patch + + api = MagicMock() + with patch( + "src.migration_check.check_migration", + return_value={}, + ), patch( + "src.migration_check.check_drive_migration", + return_value=None, + ), self.assertLogs(sync.LOGGER, level=logging.INFO) as cm: + sync._perform_dry_run( # noqa: SLF001 + config=self.config, + api=api, + check_files=0, + ) + joined = "\n".join(cm.output) + self.assertIn("--check-files=all", joined) + + def test_photos_check_files_walk_failure_logged_as_warning(self): + """If migration_check.check_migration raises, the failure is a + WARNING (not a fatal crash) — the rest of the dry-run continues.""" + from unittest.mock import MagicMock, patch + + api = MagicMock() + with patch( + "src.migration_check.check_migration", + side_effect=RuntimeError("photos boom"), + ), patch( + "src.migration_check.check_drive_migration", + return_value=None, + ), self.assertLogs( + sync.LOGGER, level=logging.WARNING, + ) as cm: + sync._perform_dry_run( # noqa: SLF001 + config=self.config, + api=api, + check_files=10, + ) + joined = "\n".join(cm.output) + self.assertIn("photos check-files walk failed", joined) + + def test_drive_check_files_walk_failure_logged_as_warning(self): + """If migration_check.check_drive_migration raises, the failure + is a WARNING — doesn't crash _perform_dry_run.""" + from unittest.mock import MagicMock, patch + + api = MagicMock() + with patch( + "src.migration_check.check_migration", + return_value={}, + ), patch( + "src.migration_check.check_drive_migration", + side_effect=RuntimeError("drive boom"), + ), self.assertLogs(sync.LOGGER, level=logging.WARNING) as cm: + sync._perform_dry_run( # noqa: SLF001 + config=self.config, + api=api, + check_files=10, + ) + joined = "\n".join(cm.output) + self.assertIn("drive check-files walk failed", joined) + + def test_check_files_none_skips_migration_walks(self): + """Default check_files=None → migration_check is NEVER called.""" + from unittest.mock import patch + + api = self._make_api() + with patch( + "src.migration_check.check_migration", + ) as fake_photos, patch( + "src.migration_check.check_drive_migration", + ) as fake_drive: + sync._perform_dry_run(config=self.config, api=api) # noqa: SLF001 + fake_photos.assert_not_called() + fake_drive.assert_not_called() + + def _make_api(self): + from unittest.mock import MagicMock + + api = MagicMock() + api.drive.dir.return_value = [] + api.photos.libraries = {} + return api + + +class TestSyncDryRunIntegration(unittest.TestCase): + """Integration: ``sync.sync(dry_run=True)`` short-circuits the loop.""" + + def setUp(self): + """Reset temp + load config.""" + config = read_config(config_path=tests.CONFIG_PATH) + assert isinstance(config, dict) + self.config = config + self.config["app"]["root"] = tests.TEMP_DIR + os.makedirs(tests.TEMP_DIR, exist_ok=True) + + def tearDown(self): + """Cleanup.""" + if os.path.exists(tests.TEMP_DIR): + shutil.rmtree(tests.TEMP_DIR) + + @patch("src.sync._perform_dry_run") + @patch("src.sync._perform_photos_sync") + @patch("src.sync._perform_drive_sync") + @patch("src.sync._authenticate_and_get_api") + @patch("src.sync.alive") + @patch("src.sync._load_configuration") + def test_dry_run_invokes_perform_dry_run_and_skips_syncs( + self, + mock_load_config, + mock_alive, + mock_auth, + mock_drive_sync, + mock_photos_sync, + mock_perform_dry_run, + ): + """dry_run=True → _perform_dry_run is called; the real syncs are not.""" + mock_load_config.return_value = self.config + # Make get_username return something truthy. + with patch( + "src.config_parser.get_username", return_value=data.AUTHENTICATED_USER, + ): + api = MagicMock() + api.requires_2sa = False + mock_auth.return_value = api + + sync.sync(dry_run=True) + + mock_perform_dry_run.assert_called_once() + mock_drive_sync.assert_not_called() + mock_photos_sync.assert_not_called() + + @patch("src.sync.notify.send_sync_summary") + @patch("src.sync._perform_dry_run") + @patch("src.sync._authenticate_and_get_api") + @patch("src.sync.alive") + @patch("src.sync._load_configuration") + def test_dry_run_does_not_send_notifications( + self, + mock_load_config, + mock_alive, + mock_auth, + mock_perform_dry_run, + mock_notify_send, + ): + """No sync summary notification under dry-run.""" + mock_load_config.return_value = self.config + with patch( + "src.config_parser.get_username", return_value=data.AUTHENTICATED_USER, + ): + api = MagicMock() + api.requires_2sa = False + mock_auth.return_value = api + sync.sync(dry_run=True) + mock_notify_send.assert_not_called() + + @patch("src.sync._perform_dry_run") + @patch("src.sync._authenticate_and_get_api") + @patch("src.sync.sleep") + @patch("src.sync.alive") + @patch("src.sync._load_configuration") + def test_dry_run_does_not_loop( + self, + mock_load_config, + mock_alive, + mock_sleep, + mock_auth, + mock_perform_dry_run, + ): + """sync(dry_run=True) returns after a single authenticate — never sleeps.""" + mock_load_config.return_value = self.config + with patch( + "src.config_parser.get_username", return_value=data.AUTHENTICATED_USER, + ): + api = MagicMock() + api.requires_2sa = False + mock_auth.return_value = api + sync.sync(dry_run=True) + # If we got here without timing out, the loop exited. Sanity-check: + mock_sleep.assert_not_called() + mock_perform_dry_run.assert_called_once() + + @patch("src.sync._perform_dry_run") + @patch("src.sync._authenticate_and_get_api") + @patch("src.sync.alive") + @patch("src.sync._load_configuration") + def test_dry_run_with_2fa_required_skips_perform_dry_run( + self, + mock_load_config, + mock_alive, + mock_auth, + mock_perform_dry_run, + ): + """If 2FA is pending, dry-run logs a message and exits without enumerating.""" + mock_load_config.return_value = self.config + with patch( + "src.config_parser.get_username", return_value=data.AUTHENTICATED_USER, + ): + api = MagicMock() + api.requires_2sa = True + mock_auth.return_value = api + with self.assertLogs(sync.LOGGER, level=logging.INFO) as cm: + sync.sync(dry_run=True) + joined = "\n".join(cm.output) + self.assertIn("DRY RUN: 2FA required", joined) + mock_perform_dry_run.assert_not_called() + + +class TestSyncSignatureBackwardCompat(unittest.TestCase): + """``sync.sync()`` (no args) must still work — dry_run default is False.""" + + def test_dry_run_default_is_false(self): + """Default parameter binding.""" + import inspect + + sig = inspect.signature(sync.sync) + self.assertIn("dry_run", sig.parameters) + self.assertEqual(sig.parameters["dry_run"].default, False) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_migration_check.py b/tests/test_migration_check.py new file mode 100644 index 000000000..45ace4077 --- /dev/null +++ b/tests/test_migration_check.py @@ -0,0 +1,785 @@ +"""Tests for ``src.migration_check.check_library`` — the per-photo +file-existence check driven by ``--dry-run --check-files``.""" + +__author__ = "Mandar Patil (mandarons@pm.me)" + +import os +import tempfile +import unittest +from unittest.mock import MagicMock + +import tests # noqa: F401 — env setup +from src import migration_check + + +def _fake_photo(filename: str, size: int, year: int = 2024, month: int = 1): + """Build a MagicMock that quacks like an icloudpy PhotoAsset.""" + from datetime import datetime + + photo = MagicMock() + photo.filename = filename + photo.created = datetime(year, month, 15) + photo.versions = {"original": {"size": size, "type": "public.heic"}} + photo.id = filename.encode().hex() # stable, unique-ish id for testing + return photo + + +def _fake_library(photos: list): + """Wrap a list of photo mocks in an object shaped like a PhotoLibrary. + + Exposes ``.all`` (what the real sync iterates) AND + ``.albums["All Photos"]`` (legacy callers / older test fixtures). + """ + library = MagicMock() + library.all = photos + library.albums = {"All Photos": photos} + return library + + +# Whether the upstream supports the "simple" filename_format toggle. +# Used as a hint inside setUp — if PR 5 isn't merged, we monkey-patch +# the metadata-format filename generator to return the photo's bare +# `.filename` instead, so the test fixtures (`IMG_N.HEIC`) still match +# the path that check_library computes. Lifts cleanly once +# feat/photos-filename-format-simple is merged upstream. +_SIMPLE_FORMAT_AVAILABLE = hasattr( + __import__("src.photo_path_utils", fromlist=["x"]), + "set_default_filename_format", +) + + +class TestCheckLibrary(unittest.TestCase): + """Behaviour of the per-library walk.""" + + def setUp(self): + if _SIMPLE_FORMAT_AVAILABLE: + # PR 5 merged: use the real simple-format toggle. + migration_check.set_default_filename_format("simple") + self._addCleanup_no_patch = True + else: + # PR 5 NOT merged: monkey-patch the filename generator at the + # migration_check level to return the photo's bare filename. + # Reproduces what "simple" mode would do without depending on + # the PR 5 code path. + from unittest.mock import patch + + self._patcher = patch.object( + migration_check, + "generate_photo_filename_with_metadata", + side_effect=lambda photo, _size: photo.filename, + ) + self._patcher.start() + self.addCleanup(self._patcher.stop) + + def test_empty_library(self): + with tempfile.TemporaryDirectory() as base: + result = migration_check.check_library( + library=_fake_library([]), + library_name="PrimarySync", + photos_base=base, + mapping={}, + folder_format=None, + sample=0, + ) + self.assertEqual(result["stats"]["would_skip"], 0) + self.assertEqual(result["stats"]["not_found"], 0) + self.assertEqual(result["checked"], 0) + + def test_all_photos_present_and_correct_size(self): + with tempfile.TemporaryDirectory() as base: + # Create three photos on disk at the expected paths with the + # expected sizes; check_library should report all would_skip. + photos = [] + # Match the real sync's layout: /all/. + all_dir = os.path.join(base, "all") + os.makedirs(all_dir, exist_ok=True) + for i, size in enumerate([1000, 2000, 3000]): + name = f"IMG_{i}.HEIC" + with open(os.path.join(all_dir, name), "wb") as f: + f.write(b"x" * size) + photos.append(_fake_photo(name, size)) + + result = migration_check.check_library( + library=_fake_library(photos), + library_name="PrimarySync", + photos_base=base, + mapping={}, # empty mapping → fall through to base + folder_format=None, + sample=0, + ) + self.assertEqual(result["stats"]["would_skip"], 3) + self.assertEqual(result["stats"]["size_mismatch"], 0) + self.assertEqual(result["stats"]["not_found"], 0) + self.assertEqual(result["checked"], 3) + + def test_all_photos_missing(self): + with tempfile.TemporaryDirectory() as base: + photos = [_fake_photo(f"IMG_{i}.HEIC", 1000 + i) for i in range(3)] + result = migration_check.check_library( + library=_fake_library(photos), + library_name="PrimarySync", + photos_base=base, + mapping={}, + folder_format=None, + sample=0, + ) + self.assertEqual(result["stats"]["not_found"], 3) + self.assertEqual(result["stats"]["would_skip"], 0) + + def test_size_mismatch_counts_separately_from_not_found(self): + with tempfile.TemporaryDirectory() as base: + # Real sync writes under /all/, so the + # checker looks there too. + all_dir = os.path.join(base, "all") + os.makedirs(all_dir, exist_ok=True) + with open(os.path.join(all_dir, "IMG_a.HEIC"), "wb") as f: + f.write(b"x" * 1000) # matches + with open(os.path.join(all_dir, "IMG_b.HEIC"), "wb") as f: + f.write(b"x" * 500) # wrong size — expected 2000 + + photos = [ + _fake_photo("IMG_a.HEIC", 1000), + _fake_photo("IMG_b.HEIC", 2000), + _fake_photo("IMG_c.HEIC", 3000), # no file on disk + ] + result = migration_check.check_library( + library=_fake_library(photos), + library_name="PrimarySync", + photos_base=base, + mapping={}, + folder_format=None, + sample=0, + ) + self.assertEqual(result["stats"]["would_skip"], 1) + self.assertEqual(result["stats"]["size_mismatch"], 1) + self.assertEqual(result["stats"]["not_found"], 1) + # samples capture the per-status examples + self.assertEqual(len(result["samples"]["would_skip"]), 1) + self.assertEqual(len(result["samples"]["size_mismatch"]), 1) + # size_mismatch sample carries (path, expected, actual) + path, expected, actual = result["samples"]["size_mismatch"][0] + self.assertEqual(expected, 2000) + self.assertEqual(actual, 500) + + def test_sample_caps_walk_at_N(self): + """``sample=N`` walks N stride-sampled photos and stops.""" + with tempfile.TemporaryDirectory() as base: + photos = [_fake_photo(f"IMG_{i}.HEIC", 1000) for i in range(200)] + result = migration_check.check_library( + library=_fake_library(photos), + library_name="PrimarySync", + photos_base=base, + mapping={}, + folder_format=None, + sample=10, + ) + self.assertEqual(result["checked"], 10) + + +def _fake_drive_file(name: str, size: int): + """Build a MagicMock that quacks like an icloudpy Drive file node.""" + node = MagicMock() + node.name = name + node.type = "file" + node.size = size + return node + + +def _fake_drive_folder(name: str, children: dict): + """Build a MagicMock that quacks like an icloudpy Drive folder node. + + ``children`` is a dict mapping child-name → child-mock (file or folder). + """ + node = MagicMock() + node.name = name + node.type = "folder" + node.dir.return_value = list(children.keys()) + node.__getitem__.side_effect = lambda key: children[key] + return node + + +class TestCheckDrive(unittest.TestCase): + """Behaviour of the iCloud Drive walker.""" + + def test_empty_drive(self): + with tempfile.TemporaryDirectory() as base: + drive = _fake_drive_folder("root", {}) + result = migration_check.check_drive( + drive=drive, + drive_destination=base, + sample=0, + ) + self.assertEqual(result["checked"], 0) + self.assertEqual(result["stats"]["would_skip"], 0) + self.assertEqual(result["stats"]["not_found"], 0) + + def test_all_files_present_at_root(self): + with tempfile.TemporaryDirectory() as base: + # Three files on disk with known sizes + for name, size in [("a.txt", 100), ("b.txt", 200), ("c.txt", 300)]: + with open(os.path.join(base, name), "wb") as f: + f.write(b"x" * size) + drive = _fake_drive_folder( + "root", + { + "a.txt": _fake_drive_file("a.txt", 100), + "b.txt": _fake_drive_file("b.txt", 200), + "c.txt": _fake_drive_file("c.txt", 300), + }, + ) + result = migration_check.check_drive( + drive=drive, + drive_destination=base, + sample=0, + ) + self.assertEqual(result["stats"]["would_skip"], 3) + self.assertEqual(result["stats"]["size_mismatch"], 0) + self.assertEqual(result["stats"]["not_found"], 0) + self.assertEqual(result["checked"], 3) + + def test_missing_and_size_mismatch_at_root(self): + with tempfile.TemporaryDirectory() as base: + # a.txt matches; b.txt wrong size; c.txt missing. + with open(os.path.join(base, "a.txt"), "wb") as f: + f.write(b"x" * 100) + with open(os.path.join(base, "b.txt"), "wb") as f: + f.write(b"x" * 50) # expected 200 + drive = _fake_drive_folder( + "root", + { + "a.txt": _fake_drive_file("a.txt", 100), + "b.txt": _fake_drive_file("b.txt", 200), + "c.txt": _fake_drive_file("c.txt", 300), + }, + ) + result = migration_check.check_drive( + drive=drive, + drive_destination=base, + sample=0, + ) + self.assertEqual(result["stats"]["would_skip"], 1) + self.assertEqual(result["stats"]["size_mismatch"], 1) + self.assertEqual(result["stats"]["not_found"], 1) + self.assertEqual(len(result["samples"]["size_mismatch"]), 1) + _, expected, actual = result["samples"]["size_mismatch"][0] + self.assertEqual(expected, 200) + self.assertEqual(actual, 50) + + def test_recurses_into_subfolders(self): + with tempfile.TemporaryDirectory() as base: + # On-disk: base/sub/leaf.txt @ 42 bytes + os.makedirs(os.path.join(base, "sub")) + with open(os.path.join(base, "sub", "leaf.txt"), "wb") as f: + f.write(b"x" * 42) + sub_folder = _fake_drive_folder( + "sub", + {"leaf.txt": _fake_drive_file("leaf.txt", 42)}, + ) + drive = _fake_drive_folder("root", {"sub": sub_folder}) + result = migration_check.check_drive( + drive=drive, + drive_destination=base, + sample=0, + ) + self.assertEqual(result["stats"]["would_skip"], 1) + self.assertEqual(result["checked"], 1) + + def test_unpacked_package_directory_counts_as_skip_when_size_matches(self): + """Mandarons-unpacked packages live as directories on disk; sum + of contained-file sizes must match the package item size.""" + with tempfile.TemporaryDirectory() as base: + # Package directory with two inner files summing to 150 + pkg_path = os.path.join(base, "Project.band") + os.makedirs(pkg_path) + with open(os.path.join(pkg_path, "metadata.plist"), "wb") as f: + f.write(b"x" * 50) + with open(os.path.join(pkg_path, "projectdata"), "wb") as f: + f.write(b"x" * 100) + drive = _fake_drive_folder( + "root", + {"Project.band": _fake_drive_file("Project.band", 150)}, + ) + result = migration_check.check_drive( + drive=drive, + drive_destination=base, + sample=0, + ) + self.assertEqual(result["stats"]["would_skip"], 1) + self.assertEqual(result["stats"]["size_mismatch"], 0) + + def test_sample_caps_drive_walk_at_N(self): + with tempfile.TemporaryDirectory() as base: + children = { + f"f{i}.txt": _fake_drive_file(f"f{i}.txt", 100) for i in range(20) + } + drive = _fake_drive_folder("root", children) + result = migration_check.check_drive( + drive=drive, + drive_destination=base, + sample=5, + ) + self.assertEqual(result["checked"], 5) + + +class TestCheckOnePhotoEdges(unittest.TestCase): + """Branches of ``_check_one_photo`` not covered by happy-path tests.""" + + def setUp(self): + # Same monkey-patch as TestCheckLibrary so the metadata-format + # generator returns the photo's bare filename. + if not _SIMPLE_FORMAT_AVAILABLE: + from unittest.mock import patch + + self._patcher = patch.object( + migration_check, + "generate_photo_filename_with_metadata", + side_effect=lambda photo, _size: photo.filename, + ) + self._patcher.start() + self.addCleanup(self._patcher.stop) + + def test_missing_original_version_returns_error(self): + """A photo with no `original` version key (incomplete CloudKit + record) returns the `error` status without crashing.""" + with tempfile.TemporaryDirectory() as base: + photo = _fake_photo("IMG_0.HEIC", 1000) + photo.versions = {"medium": {"size": 500}} # no "original" + result = migration_check.check_library( + library=_fake_library([photo]), + library_name="PrimarySync", + photos_base=base, + mapping={}, + folder_format=None, + sample=0, + ) + self.assertEqual(result["stats"]["error"], 1) + + def test_path_computation_exception_returns_error(self): + """If the filename generator itself raises (malformed photo, + unicode oddity, …) the per-photo result is `error`, not a + crash that kills the whole walk.""" + from unittest.mock import patch + + with tempfile.TemporaryDirectory() as base: + photo = _fake_photo("IMG_0.HEIC", 1000) + with patch.object( + migration_check, + "generate_photo_filename_with_metadata", + side_effect=RuntimeError("filename boom"), + ): + result = migration_check.check_library( + library=_fake_library([photo]), + library_name="PrimarySync", + photos_base=base, + mapping={}, + folder_format=None, + sample=0, + ) + self.assertEqual(result["stats"]["error"], 1) + + def test_getsize_oserror_returns_error(self): + """``os.path.getsize`` can raise OSError on perms / FS errors + even when ``isfile`` returned True. That edge is reported as + `error`, not size_mismatch.""" + from unittest.mock import patch + + with tempfile.TemporaryDirectory() as base: + name = "IMG_0.HEIC" + all_dir = os.path.join(base, "all") + os.makedirs(all_dir, exist_ok=True) + with open(os.path.join(all_dir, name), "wb") as f: + f.write(b"x" * 1000) + photo = _fake_photo(name, 1000) + with patch("os.path.getsize", side_effect=OSError("perms")): + result = migration_check.check_library( + library=_fake_library([photo]), + library_name="PrimarySync", + photos_base=base, + mapping={}, + folder_format=None, + sample=0, + ) + self.assertEqual(result["stats"]["error"], 1) + + def test_iteration_exception_stops_early_without_crash(self): + """A walk that explodes mid-iteration (transient iCloud error) + is logged and returns partial counts — not a hard fail.""" + + class BoomIterable: + """Iterable that yields one photo, then explodes.""" + + def __iter__(self): + yield _fake_photo("IMG_0.HEIC", 1000) + msg = "iter boom" + raise RuntimeError(msg) + + class BoomLibrary: + def __init__(self): + # check_library iterates ``library.all`` (matching + # sync_photos.py's default path). + self.all = BoomIterable() + self.albums = {"All Photos": self.all} + + with tempfile.TemporaryDirectory() as base: + result = migration_check.check_library( + library=BoomLibrary(), + library_name="PrimarySync", + photos_base=base, + mapping={}, + folder_format=None, + sample=0, + ) + # Got one error (the one photo we saw before boom), then + # the iterator died — checked=1, not None. + self.assertEqual(result["checked"], 1) + + +class TestCheckOneDriveFileEdges(unittest.TestCase): + """Branches of ``_check_one_drive_file`` not covered by happy-path + tests in TestCheckDrive (size guard, getsize OSError, directory + sum OSError, missing-on-disk → not_found).""" + + def test_missing_item_size_attr_is_error(self): + with tempfile.TemporaryDirectory() as base: + item = MagicMock() + item.name = "weird.txt" + item.size = "not a number" # int() raises + status, _path, _expected, _actual = migration_check._check_one_drive_file( # noqa: SLF001 + item, + os.path.join(base, "weird.txt"), + ) + self.assertEqual(status, "error") + + def test_disk_getsize_oserror_returns_error(self): + from unittest.mock import patch + + with tempfile.TemporaryDirectory() as base: + path = os.path.join(base, "file.txt") + with open(path, "wb") as f: + f.write(b"x" * 100) + item = _fake_drive_file("file.txt", 100) + with patch("os.path.getsize", side_effect=OSError("perms")): + status, *_ = migration_check._check_one_drive_file(item, path) # noqa: SLF001 + self.assertEqual(status, "error") + + def test_directory_sum_oserror_returns_error(self): + from unittest.mock import patch + + with tempfile.TemporaryDirectory() as base: + pkg = os.path.join(base, "pkg.band") + os.makedirs(pkg) + item = _fake_drive_file("pkg.band", 100) + with patch( + "pathlib.Path.glob", + side_effect=OSError("perms"), + ): + status, *_ = migration_check._check_one_drive_file(item, pkg) # noqa: SLF001 + self.assertEqual(status, "error") + + +class TestWalkDriveRecursiveEdges(unittest.TestCase): + """Branches of ``_walk_drive_recursive`` not exercised by the + flat-folder happy-path tests in TestCheckDrive.""" + + def test_dir_call_exception_returns_silently(self): + """If folder.dir() raises (transient iCloud error) the walker + logs and returns — doesn't crash the whole sync.""" + folder = MagicMock() + folder.dir.side_effect = RuntimeError("dir boom") + state = { + "checked": 0, + "stats": {"would_skip": 0, "size_mismatch": 0, "not_found": 0, "error": 0}, + "samples": {"would_skip": [], "size_mismatch": [], "not_found": []}, + } + migration_check._walk_drive_recursive( # noqa: SLF001 + folder=folder, + destination_path="/x", + sample=0, + state=state, + ) + self.assertEqual(state["checked"], 0) + + def test_item_getitem_failure_counted_as_error(self): + """``folder[name]`` can raise (item disappeared / permission + change between dir() and getitem). Each failure adds an error + stat — the walk continues.""" + folder = MagicMock() + folder.dir.return_value = ["a", "b"] + folder.__getitem__.side_effect = RuntimeError("item boom") + state = { + "checked": 0, + "stats": {"would_skip": 0, "size_mismatch": 0, "not_found": 0, "error": 0}, + "samples": {"would_skip": [], "size_mismatch": [], "not_found": []}, + } + migration_check._walk_drive_recursive( # noqa: SLF001 + folder=folder, + destination_path="/x", + sample=0, + state=state, + ) + self.assertEqual(state["stats"]["error"], 2) + + def test_recurses_into_subfolders_and_files(self): + """Folder → sub-folder → file chain works. Covers the recursive + branch + the file branch.""" + with tempfile.TemporaryDirectory() as base: + inner_dir = os.path.join(base, "inner") + os.makedirs(inner_dir) + with open(os.path.join(inner_dir, "f.txt"), "wb") as f: + f.write(b"x" * 50) + inner_folder = _fake_drive_folder( + "inner", + {"f.txt": _fake_drive_file("f.txt", 50)}, + ) + inner_folder.name = "inner" + root = _fake_drive_folder("root", {"inner": inner_folder}) + result = migration_check.check_drive( + drive=root, + drive_destination=base, + sample=0, + ) + self.assertEqual(result["checked"], 1) + self.assertEqual(result["stats"]["would_skip"], 1) + + +class TestWalkDriveRecursiveExtraEdges(unittest.TestCase): + """More edge branches inside ``_walk_drive_recursive``.""" + + def test_returns_early_when_sample_already_met(self): + """If the caller passes state with checked >= sample, the walker + returns immediately without calling .dir() at all.""" + folder = MagicMock() + state = { + "checked": 10, + "stats": {"would_skip": 0, "size_mismatch": 0, "not_found": 0, "error": 0}, + "samples": {"would_skip": [], "size_mismatch": [], "not_found": []}, + } + migration_check._walk_drive_recursive( # noqa: SLF001 + folder=folder, + destination_path="/x", + sample=5, + state=state, + ) + folder.dir.assert_not_called() + + def test_folder_unquote_exception_falls_back_to_raw_name(self): + """If urllib unquote raises on a folder name (highly unusual), + the walker uses the raw name and continues recursing.""" + from unittest.mock import patch + + inner = _fake_drive_folder("inner", {}) + inner.name = "name-that-breaks-unquote" + root = _fake_drive_folder("root", {"inner": inner}) + state = { + "checked": 0, + "stats": {"would_skip": 0, "size_mismatch": 0, "not_found": 0, "error": 0}, + "samples": {"would_skip": [], "size_mismatch": [], "not_found": []}, + } + with patch.object( + migration_check, + "unquote", + side_effect=RuntimeError("unquote boom"), + ): + migration_check._walk_drive_recursive( # noqa: SLF001 + folder=root, + destination_path="/x", + sample=0, + state=state, + ) + # No crash; the inner folder was visited (empty so nothing checked). + self.assertEqual(state["checked"], 0) + + def test_file_unquote_exception_falls_back_to_raw_name(self): + """Same fallback for file items.""" + from unittest.mock import patch + + with tempfile.TemporaryDirectory() as base: + # File on disk at a path that matches the raw name fallback. + name = "file.txt" + with open(os.path.join(base, name), "wb") as f: + f.write(b"x" * 50) + root = _fake_drive_folder("root", {name: _fake_drive_file(name, 50)}) + state = { + "checked": 0, + "stats": { + "would_skip": 0, + "size_mismatch": 0, + "not_found": 0, + "error": 0, + }, + "samples": {"would_skip": [], "size_mismatch": [], "not_found": []}, + } + with patch.object( + migration_check, + "unquote", + side_effect=RuntimeError("unquote boom"), + ): + migration_check._walk_drive_recursive( # noqa: SLF001 + folder=root, + destination_path=base, + sample=0, + state=state, + ) + self.assertEqual(state["checked"], 1) + self.assertEqual(state["stats"]["would_skip"], 1) + + +class TestCheckDriveExceptionPath(unittest.TestCase): + """``check_drive``'s try/except — if _walk_drive_recursive raises, + the wrapper logs a warning and returns the partial state instead + of letting the exception bubble out.""" + + def test_walk_exception_logged_and_partial_returned(self): + from unittest.mock import patch + + with tempfile.TemporaryDirectory() as base: + drive = _fake_drive_folder("root", {}) + with patch.object( + migration_check, + "_walk_drive_recursive", + side_effect=RuntimeError("walk boom"), + ): + result = migration_check.check_drive( + drive=drive, + drive_destination=base, + sample=0, + ) + self.assertIn("stats", result) + self.assertEqual(result["checked"], 0) + + +class TestCheckMigrationLibraryDestinationsBranch(unittest.TestCase): + """Cover the `hasattr(config_parser, "get_photos_library_destinations")` + True branch — fires when PR 3 (feat/photos-library-destinations) IS + merged. Since this branch doesn't have PR 3, we monkey-patch the + config_parser to expose the function so the True branch is exercised.""" + + def setUp(self): + from src import read_config + + config = read_config(config_path=tests.CONFIG_PATH) + assert isinstance(config, dict) + self.config = config + self.config["app"]["root"] = tests.TEMP_DIR + os.makedirs(tests.TEMP_DIR, exist_ok=True) + + def tearDown(self): + import shutil + + if os.path.exists(tests.TEMP_DIR): + shutil.rmtree(tests.TEMP_DIR) + + def test_uses_real_library_destinations_when_PR3_is_merged(self): + """When PR 3 is merged, ``get_photos_library_destinations`` exists + and check_migration uses it directly. Coverage-only test.""" + from unittest.mock import MagicMock as _MM + from unittest.mock import patch + + api = _MM() + api.photos.libraries = {"PrimarySync": _fake_library([])} + fake_getter = _MM(return_value={"PrimarySync": "personal"}) + with patch.object( + migration_check.config_parser, + "get_photos_library_destinations", + fake_getter, + create=True, + ): + migration_check.check_migration(api=api, config=self.config, sample=0) + fake_getter.assert_called_once_with(config=self.config) + + +class TestCheckMigrationOrchestrator(unittest.TestCase): + """``check_migration`` iterates api.photos.libraries and routes to + ``check_library`` per library. ``check_drive_migration`` resolves + drive_destination and routes to ``check_drive``.""" + + def setUp(self): + from src import read_config + + config = read_config(config_path=tests.CONFIG_PATH) + assert isinstance(config, dict) + self.config = config + self.config["app"]["root"] = tests.TEMP_DIR + os.makedirs(tests.TEMP_DIR, exist_ok=True) + + def tearDown(self): + import shutil + + if os.path.exists(tests.TEMP_DIR): + shutil.rmtree(tests.TEMP_DIR) + + def test_check_migration_iterates_all_libraries(self): + """Two libraries → result dict has both keys + each value is + the per-library result shape from check_library.""" + api = MagicMock() + api.photos.libraries = { + "PrimarySync": _fake_library([]), + "SharedLibrary": _fake_library([]), + } + results = migration_check.check_migration( + api=api, + config=self.config, + sample=0, + ) + self.assertEqual(set(results.keys()), {"PrimarySync", "SharedLibrary"}) + for r in results.values(): + self.assertIn("library_dest", r) + self.assertIn("stats", r) + + def test_check_migration_sets_filename_format_from_config(self): + """When `photos.filename_format` is set in config, the + orchestrator pushes it through to set_default_filename_format + before walking.""" + from unittest.mock import patch + + api = MagicMock() + api.photos.libraries = {"PrimarySync": _fake_library([])} + self.config["photos"]["filename_format"] = "simple" + with patch.object( + migration_check, + "set_default_filename_format", + ) as fake_setter: + migration_check.check_migration(api=api, config=self.config, sample=0) + fake_setter.assert_called_once_with("simple") + + def test_check_drive_migration_returns_none_when_no_drive_section(self): + config = {"photos": {"destination": "/photos"}} + self.assertIsNone( + migration_check.check_drive_migration(api=MagicMock(), config=config), + ) + + def test_check_drive_migration_returns_none_when_destination_resolution_fails(self): + from unittest.mock import patch + + # check_drive_migration now uses the non-mutating + # get_*_destination_path getters (per Copilot review feedback — + # dry-run mustn't side-effect mkdir). Patch the getter that the + # try-block actually invokes. + with patch.object( + migration_check.config_parser, + "get_root_destination_path", + side_effect=RuntimeError("bad path"), + ): + self.assertIsNone( + migration_check.check_drive_migration( + api=MagicMock(), + config=self.config, + ), + ) + + def test_check_drive_migration_walks_when_drive_configured(self): + """Happy path: drive section present, destination resolves, walk + produces a result dict.""" + api = MagicMock() + api.drive = _fake_drive_folder("root", {}) + result = migration_check.check_drive_migration( + api=api, + config=self.config, + sample=0, + ) + self.assertIsNotNone(result) + self.assertIn("stats", result) + + +if __name__ == "__main__": + unittest.main()