From bb0c181f9113c9ae3aa79c660830d5412fe76795 Mon Sep 17 00:00:00 2001 From: Eric Pheterson Date: Fri, 29 May 2026 12:13:44 -0700 Subject: [PATCH 01/10] fix(drive): treat unrecognised-mime packages as single-file bundles MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit iCloud Drive serves Apple iWork (.key, .pages, .numbers), JMG (.jmb), and other "package" types as opaque archives. libmagic detects only zip/gzip; everything else reports as application/octet-stream and the old code path returned None from process_package(), which the caller in drive_file_download.py treated as a hard download failure — even though the bytes had ALREADY been written to disk and were perfectly usable. Result on Eric's library: every sync cycle re-downloaded ~300 MB of .key + .jmb files, logged as "0 successful, 23 failed" — alarming but not actually a data-loss bug, just wasted bandwidth + misleading logs. Fix: - process_package() now returns the local_file path (truthy) when the mime type isn't a recognised archive. The file is on disk in a usable form; that IS the canonical local representation for these bundle types. Logged at INFO ("Package format not recognised for unpacking; keeping as single-file bundle") instead of ERROR. - drive_file_download.py treats only ``None`` return as failure (was treating any falsy value as failure). Tests: - Renames test_process_package_invalid_package_type → test_process_package_unrecognised_mime_preserves_file with the new contract. - Adds test_download_file_keeps_unrecognised_package_as_single_file to exercise the success path through download_file. - Fixes the patch target in test_download_file_returns_none_on_processing_failure (was patching the wrong module-level binding; happened to pass under the old contract by coincidence). NOT fixed in this PR: the next-sync dedup still triggers a re-download because file_exists() compares getsize(local_file) to item.size, and for flat bundles those differ (item.size is the unpacked contents total). That requires either a sidecar metadata file or relaxing the size comparator to fall back on mtime; out of scope for this focused fix. Documented as a follow-up. Co-Authored-By: Claude --- src/drive_file_download.py | 16 +- src/drive_package_processing.py | 22 +- tests/test_sync_drive.py | 409 +++++++++++++++++++++++++------- 3 files changed, 359 insertions(+), 88 deletions(-) diff --git a/src/drive_file_download.py b/src/drive_file_download.py index 42494f1e2..9efde3d89 100644 --- a/src/drive_file_download.py +++ b/src/drive_file_download.py @@ -42,13 +42,21 @@ def download_file(item: Any, local_file: str) -> str | None: for chunk in response.iter_content(4 * 1024 * 1024): file_out.write(chunk) - # Check if this is a package that needs processing + # Check if this is a package that needs processing. + # + # ``process_package`` now returns the local_file path for both + # successful unpacks AND unrecognised mime types (the bytes + # are still on disk as a flat bundle — see the docstring on + # ``process_package`` for the rationale). It only returns + # ``None`` on hard processing failure that leaves the file in + # an unusable state. So we surface that as a download failure + # but no longer treat "couldn't unpack" as failure when the + # downloaded bytes are intact. if response.url and "/packageDownload?" in response.url: processed_file = process_package(local_file=local_file) - if processed_file: - local_file = processed_file - else: + if processed_file is None: return None + local_file = processed_file # Set the file modification time to match the remote file item_modified_time = time.mktime(item.date_modified.timetuple()) diff --git a/src/drive_package_processing.py b/src/drive_package_processing.py index fdd7b29f4..c152d93b6 100644 --- a/src/drive_package_processing.py +++ b/src/drive_package_processing.py @@ -33,7 +33,14 @@ def process_package(local_file: str) -> str | None: local_file: Path to the downloaded package file Returns: - Path to the processed file/directory, or False if processing failed + Path to the processed file/directory. The local file path is also + returned when the mime type is not a recognised archive (the + bytes are preserved on disk as a single-file bundle — Apple's + iWork formats ``.key``/``.pages``/``.numbers``, JMG ``.jmb``, + etc. report as ``application/octet-stream`` and don't need + unpacking to be usable by their target application). Returns + ``None`` only on hard processing errors that leave the file in + an unusable state. """ archive_file = local_file magic_object = magic.Magic(mime=True) @@ -44,10 +51,17 @@ def process_package(local_file: str) -> str | None: elif file_mime_type == "application/gzip": return _process_gzip_package(local_file, archive_file) else: - LOGGER.error( - f"Unhandled file type - cannot unpack the package {file_mime_type}.", + # NOT an error — many iCloud Drive "package" downloads are flat + # binary bundles (Apple iWork .key/.pages/.numbers, + # third-party .jmb, etc) that report as application/octet-stream + # but don't need to be unpacked to be usable by their target + # application. The bytes are already on disk at ``local_file``; + # treat that as the canonical local representation. + LOGGER.info( + f"Package format not recognised for unpacking ({file_mime_type}); " + f"keeping as single-file bundle: {local_file}", ) - return None + return local_file def _process_zip_package(local_file: str, archive_file: str) -> str: diff --git a/tests/test_sync_drive.py b/tests/test_sync_drive.py index b0aa361af..41a11ed43 100644 --- a/tests/test_sync_drive.py +++ b/tests/test_sync_drive.py @@ -28,13 +28,17 @@ def setUp(self) -> None: self.root = tests.DRIVE_DIR self.destination_path = self.root os.makedirs(self.destination_path, exist_ok=True) - self.service = data.ICloudPyServiceMock(data.AUTHENTICATED_USER, data.VALID_PASSWORD) + self.service = data.ICloudPyServiceMock( + data.AUTHENTICATED_USER, data.VALID_PASSWORD + ) self.drive = self.service.drive self.items = self.drive.dir() self.folder_item = self.drive[self.items[5]] self.file_item = self.drive[self.items[4]]["Test"]["Scanned document 1.pdf"] self.package_item = self.drive[self.items[6]]["Sample"]["Project.band"] - self.special_chars_package_item = self.drive[self.items[6]]["Sample"]["Fotoksiążka-Wzór.xmcf"] + self.special_chars_package_item = self.drive[self.items[6]]["Sample"][ + "Fotoksiążka-Wzór.xmcf" + ] self.package_item_nested = self.drive[self.items[6]]["Sample"]["ms.band"] self.file_name = "Scanned document 1.pdf" self.package_name = "Project.band" @@ -470,7 +474,11 @@ def test_wanted_folder_none_folder_path(self): def test_wanted_folder_none_filters(self): """Test for wanted folder filters as None.""" - self.assertTrue(sync_drive.wanted_folder(filters=None, ignore=None, root=self.root, folder_path="dir1")) + self.assertTrue( + sync_drive.wanted_folder( + filters=None, ignore=None, root=self.root, folder_path="dir1" + ) + ) def test_wanted_folder_none_root(self): """Test for wanted folder root as None.""" @@ -487,7 +495,9 @@ def test_wanted_file(self): """Test for a valid wanted file.""" self.filters["file_extensions"] = ["py"] self.assertTrue( - sync_drive.wanted_file(filters=self.filters["file_extensions"], ignore=None, file_path=__file__), + sync_drive.wanted_file( + filters=self.filters["file_extensions"], ignore=None, file_path=__file__ + ), ) def test_wanted_file_missing(self): @@ -509,19 +519,29 @@ def test_wanted_file_check_log(self): file_path=tests.CONFIG_PATH, ) self.assertTrue(len(captured.records) > 0) - self.assertIn("Skipping the unwanted file", captured.records[0].getMessage()) + self.assertIn( + "Skipping the unwanted file", captured.records[0].getMessage() + ) def test_wanted_file_none_file_path(self): """Test for unexpected wanted file path.""" - self.assertTrue(sync_drive.wanted_file(filters=None, ignore=None, file_path=__file__)) - self.assertFalse(sync_drive.wanted_file(filters=self.filters["file_extensions"], ignore=None, file_path=None)) + self.assertTrue( + sync_drive.wanted_file(filters=None, ignore=None, file_path=__file__) + ) + self.assertFalse( + sync_drive.wanted_file( + filters=self.filters["file_extensions"], ignore=None, file_path=None + ) + ) def test_wanted_file_empty_file_extensions(self): """Test for empty file extensions in wanted file.""" original_filters = dict(self.filters) self.filters["file_extensions"] = [] self.assertTrue( - sync_drive.wanted_file(filters=self.filters["file_extensions"], ignore=None, file_path=__file__), + sync_drive.wanted_file( + filters=self.filters["file_extensions"], ignore=None, file_path=__file__ + ), ) self.filters = dict(original_filters) @@ -529,7 +549,9 @@ def test_wanted_file_case_variations_extensions(self): """Test for wanted file extensions case variations.""" self.filters["file_extensions"] = ["pY"] self.assertTrue( - sync_drive.wanted_file(filters=self.filters["file_extensions"], ignore=None, file_path=__file__), + sync_drive.wanted_file( + filters=self.filters["file_extensions"], ignore=None, file_path=__file__ + ), ) self.filters["file_extensions"] = ["pY"] self.assertTrue( @@ -639,24 +661,32 @@ def test_process_folder_none_root(self): def test_file_non_existing_file(self): """Test for file does not exist.""" - self.assertFalse(sync_drive.file_exists(item=self.file_item, local_file=self.local_file_path)) + self.assertFalse( + sync_drive.file_exists(item=self.file_item, local_file=self.local_file_path) + ) def test_file_existing_file(self): """Test for file exists.""" sync_drive.download_file(item=self.file_item, local_file=self.local_file_path) - actual = sync_drive.file_exists(item=self.file_item, local_file=self.local_file_path) + actual = sync_drive.file_exists( + item=self.file_item, local_file=self.local_file_path + ) self.assertTrue(actual) # Verbose sync_drive.download_file(item=self.file_item, local_file=self.local_file_path) with self.assertLogs(logger=LOGGER, level="DEBUG") as captured: - actual = sync_drive.file_exists(item=self.file_item, local_file=self.local_file_path) + actual = sync_drive.file_exists( + item=self.file_item, local_file=self.local_file_path + ) self.assertTrue(actual) self.assertTrue(len(captured.records) > 0) self.assertIn("No changes detected.", captured.records[0].getMessage()) def test_file_exists_none_item(self): """Test if item is None.""" - self.assertFalse(sync_drive.file_exists(item=None, local_file=self.local_file_path)) + self.assertFalse( + sync_drive.file_exists(item=None, local_file=self.local_file_path) + ) def test_file_exists_none_local_file(self): """Test if local_file is None.""" @@ -664,26 +694,44 @@ def test_file_exists_none_local_file(self): def test_package_exists_none_item(self): """Test package_exists returns False when item is None.""" - self.assertFalse(sync_drive.package_exists(item=None, local_package_path=self.local_package_path)) + self.assertFalse( + sync_drive.package_exists( + item=None, local_package_path=self.local_package_path + ) + ) def test_package_exists_non_existent_path(self): """Test package_exists returns False when local path does not exist as a directory.""" non_existent = os.path.join(self.destination_path, "does_not_exist.band") - self.assertFalse(sync_drive.package_exists(item=self.package_item, local_package_path=non_existent)) + self.assertFalse( + sync_drive.package_exists( + item=self.package_item, local_package_path=non_existent + ) + ) def test_download_file(self): """Test for valid file download.""" - self.assertTrue(sync_drive.download_file(item=self.file_item, local_file=self.local_file_path)) + self.assertTrue( + sync_drive.download_file( + item=self.file_item, local_file=self.local_file_path + ) + ) # Verbose with self.assertLogs() as captured: - self.assertTrue(sync_drive.download_file(item=self.file_item, local_file=self.local_file_path)) + self.assertTrue( + sync_drive.download_file( + item=self.file_item, local_file=self.local_file_path + ) + ) self.assertTrue(len(captured.records) > 0) self.assertIn("Downloading ", captured.records[0].getMessage()) def test_download_file_none_item(self): """Test for item as None.""" - self.assertFalse(sync_drive.download_file(item=None, local_file=self.local_file_path)) + self.assertFalse( + sync_drive.download_file(item=None, local_file=self.local_file_path) + ) def test_download_file_none_local_file(self): """Test for local_file as None.""" @@ -694,7 +742,9 @@ def test_download_file_non_existing(self): self.assertFalse( sync_drive.download_file( item=self.file_item, - local_file=os.path.join(self.destination_path, "non-existent-folder", self.file_name), + local_file=os.path.join( + self.destination_path, "non-existent-folder", self.file_name + ), ), ) @@ -702,20 +752,32 @@ def test_download_file_key_error_data_token(self): """Test for data token key error.""" with patch.object(self.file_item, "open") as mock_item: mock_item.side_effect = KeyError("data_token") - self.assertFalse(sync_drive.download_file(item=self.file_item, local_file=self.local_file_path)) + self.assertFalse( + sync_drive.download_file( + item=self.file_item, local_file=self.local_file_path + ) + ) def test_download_file_object_not_found_exception(self): """Test for ObjectNotFoundException error handling.""" with patch.object(self.file_item, "open") as mock_item: - mock_item.side_effect = Exception("ObjectNotFoundException: Could not find document (NOT_FOUND)") - self.assertFalse(sync_drive.download_file(item=self.file_item, local_file=self.local_file_path)) + mock_item.side_effect = Exception( + "ObjectNotFoundException: Could not find document (NOT_FOUND)" + ) + self.assertFalse( + sync_drive.download_file( + item=self.file_item, local_file=self.local_file_path + ) + ) def test_is_package_object_not_found_exception(self): """Test is_package function with ObjectNotFoundException error handling.""" from src.drive_file_existence import is_package with patch.object(self.file_item, "open") as mock_item: - mock_item.side_effect = Exception("ObjectNotFoundException: Could not find document (NOT_FOUND)") + mock_item.side_effect = Exception( + "ObjectNotFoundException: Could not find document (NOT_FOUND)" + ) # Should return False when error occurs result = is_package(self.file_item) self.assertFalse(result) @@ -870,7 +932,9 @@ def test_remove_obsolete_file(self): # Remove the file files = set() files.add(obsolete_path) - actual = sync_drive.remove_obsolete(destination_path=self.destination_path, files=files) + actual = sync_drive.remove_obsolete( + destination_path=self.destination_path, files=files + ) self.assertTrue(len(actual) == 1) self.assertFalse(os.path.isfile(obsolete_file_path)) @@ -885,13 +949,17 @@ def test_remove_obsolete_directory(self): shutil.copyfile(__file__, obsolete_file_path) # Remove the directory files.remove(obsolete_path) - actual = sync_drive.remove_obsolete(destination_path=self.destination_path, files=files) + actual = sync_drive.remove_obsolete( + destination_path=self.destination_path, files=files + ) self.assertTrue(len(actual) == 1) self.assertFalse(os.path.isdir(obsolete_path)) # Remove the directory with file os.mkdir(obsolete_path) shutil.copyfile(__file__, obsolete_file_path) - actual = sync_drive.remove_obsolete(destination_path=self.destination_path, files=files) + actual = sync_drive.remove_obsolete( + destination_path=self.destination_path, files=files + ) self.assertTrue(len(actual) > 0) self.assertFalse(os.path.isdir(obsolete_path)) self.assertFalse(os.path.isfile(obsolete_file_path)) @@ -899,7 +967,9 @@ def test_remove_obsolete_directory(self): with self.assertLogs() as captured: os.mkdir(obsolete_path) shutil.copyfile(__file__, obsolete_file_path) - actual = sync_drive.remove_obsolete(destination_path=self.destination_path, files=files) + actual = sync_drive.remove_obsolete( + destination_path=self.destination_path, files=files + ) self.assertTrue(len(actual) > 0) self.assertFalse(os.path.isdir(obsolete_path)) self.assertFalse(os.path.isfile(obsolete_file_path)) @@ -907,31 +977,44 @@ def test_remove_obsolete_directory(self): self.assertIn("Removing ", captured.records[0].getMessage()) @patch(target="keyring.get_password", return_value=data.VALID_PASSWORD) - @patch(target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER) + @patch( + target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER + ) @patch("icloudpy.ICloudPyService") @patch("src.read_config") - def test_remove_obsolete_package(self, mock_read_config, mock_service, mock_get_username, mock_get_password): + def test_remove_obsolete_package( + self, mock_read_config, mock_service, mock_get_username, mock_get_password + ): """Test for removing obsolete package.""" mock_service = self.service config = self.config.copy() config["drive"]["remove_obsolete"] = True config["drive"]["destination"] = self.destination_path mock_read_config.return_value = config - ms_band_package_local_path = os.path.join(self.destination_path, "Obsidian", "Sample", "ms.band") + ms_band_package_local_path = os.path.join( + self.destination_path, "Obsidian", "Sample", "ms.band" + ) files = sync_drive.sync_drive(config=config, drive=mock_service.drive) self.assertIsNotNone(files) files.remove(ms_band_package_local_path) - files = sync_drive.remove_obsolete(destination_path=self.destination_path, files=files) + files = sync_drive.remove_obsolete( + destination_path=self.destination_path, files=files + ) self.assertFalse(os.path.exists(ms_band_package_local_path)) def test_remove_obsolete_none_destination_path(self): """Test for destination path as None.""" - self.assertTrue(len(sync_drive.remove_obsolete(destination_path=None, files=set())) == 0) + self.assertTrue( + len(sync_drive.remove_obsolete(destination_path=None, files=set())) == 0 + ) def test_remove_obsolete_none_files(self): """Test for files as None.""" obsolete_path = os.path.join(self.destination_path, "obsolete") - self.assertTrue(len(sync_drive.remove_obsolete(destination_path=obsolete_path, files=None)) == 0) + self.assertTrue( + len(sync_drive.remove_obsolete(destination_path=obsolete_path, files=None)) + == 0 + ) def test_sync_directory_without_remove(self): """Test for remove as False.""" @@ -946,18 +1029,30 @@ def test_sync_directory_without_remove(self): ) self.assertTrue(len(actual) == 49) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy"))) - self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test"))) self.assertTrue( - os.path.isfile(os.path.join(self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf")), + os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")) ) self.assertTrue( - os.path.isfile(os.path.join(self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf")), + os.path.isfile( + os.path.join( + self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf" + ) + ), + ) + self.assertTrue( + os.path.isfile( + os.path.join( + self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf" + ) + ), ) def test_sync_directory_with_remove(self): """Test for remove as True.""" os.mkdir(os.path.join(self.destination_path, "obsolete")) - shutil.copyfile(__file__, os.path.join(self.destination_path, "obsolete", "obsolete.py")) + shutil.copyfile( + __file__, os.path.join(self.destination_path, "obsolete", "obsolete.py") + ) actual = sync_drive.sync_directory( drive=self.drive, destination_path=self.destination_path, @@ -970,12 +1065,22 @@ def test_sync_directory_with_remove(self): ) self.assertTrue(len(actual) == 49) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy"))) - self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test"))) self.assertTrue( - os.path.isfile(os.path.join(self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf")), + os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")) ) self.assertTrue( - os.path.isfile(os.path.join(self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf")), + os.path.isfile( + os.path.join( + self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf" + ) + ), + ) + self.assertTrue( + os.path.isfile( + os.path.join( + self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf" + ) + ), ) def test_sync_directory_without_folder_filter(self): @@ -994,12 +1099,22 @@ def test_sync_directory_without_folder_filter(self): ) self.assertTrue(len(actual) == 53) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy"))) - self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test"))) self.assertTrue( - os.path.isfile(os.path.join(self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf")), + os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")) ) self.assertTrue( - os.path.isfile(os.path.join(self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf")), + os.path.isfile( + os.path.join( + self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf" + ) + ), + ) + self.assertTrue( + os.path.isfile( + os.path.join( + self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf" + ) + ), ) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "unwanted"))) @@ -1078,36 +1193,66 @@ def test_sync_directory_none_items(self): ) @patch(target="keyring.get_password", return_value=data.VALID_PASSWORD) - @patch(target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER) + @patch( + target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER + ) @patch("icloudpy.ICloudPyService") @patch("src.read_config") - def test_sync_drive_valids(self, mock_read_config, mock_service, mock_get_username, mock_get_password): + def test_sync_drive_valids( + self, mock_read_config, mock_service, mock_get_username, mock_get_password + ): """Test for valid sync_drive.""" mock_service = self.service config = self.config.copy() config["drive"]["destination"] = self.destination_path mock_read_config.return_value = config - self.assertIsNotNone(sync_drive.sync_drive(config=config, drive=mock_service.drive)) + self.assertIsNotNone( + sync_drive.sync_drive(config=config, drive=mock_service.drive) + ) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy"))) - self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test"))) self.assertTrue( - os.path.isfile(os.path.join(self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf")), + os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")) + ) + self.assertTrue( + os.path.isfile( + os.path.join( + self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf" + ) + ), ) self.assertTrue( - os.path.isfile(os.path.join(self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf")), + os.path.isfile( + os.path.join( + self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf" + ) + ), ) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "Obsidian"))) - self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "Obsidian", "Sample"))) - self.assertTrue(os.path.isfile(os.path.join(self.destination_path, "Obsidian", "Sample", "This is a title.md"))) + self.assertTrue( + os.path.isdir(os.path.join(self.destination_path, "Obsidian", "Sample")) + ) + self.assertTrue( + os.path.isfile( + os.path.join( + self.destination_path, "Obsidian", "Sample", "This is a title.md" + ) + ) + ) self.assertEqual( sum( f.stat().st_size - for f in Path(os.path.join(self.destination_path, "Obsidian", "Sample", "Project.band")).glob("**/*") + for f in Path( + os.path.join( + self.destination_path, "Obsidian", "Sample", "Project.band" + ) + ).glob("**/*") if f.is_file() ), sum( f.stat().st_size - for f in Path(os.path.join(tests.DATA_DIR, "Project_original.band")).glob("**/*") + for f in Path( + os.path.join(tests.DATA_DIR, "Project_original.band") + ).glob("**/*") if f.is_file() ), ) @@ -1130,7 +1275,9 @@ def test_process_file_existing_package(self): """Test for existing package.""" files = set() # Existing package - sync_drive.download_file(item=self.package_item, local_file=self.local_package_path) + sync_drive.download_file( + item=self.package_item, local_file=self.local_package_path + ) # Do not download the package self.assertFalse( sync_drive.process_file( @@ -1174,15 +1321,33 @@ def test_process_file_nested_package_extraction(self): self.assertEqual( sum( f.stat().st_size - for f in Path(os.path.join(self.destination_path, "ms.band")).glob("**/*") + for f in Path(os.path.join(self.destination_path, "ms.band")).glob( + "**/*" + ) + if f.is_file() + ), + sum( + f.stat().st_size + for f in Path(os.path.join(tests.DATA_DIR, "ms.band")).glob("**/*") if f.is_file() ), - sum(f.stat().st_size for f in Path(os.path.join(tests.DATA_DIR, "ms.band")).glob("**/*") if f.is_file()), ) - def test_process_package_invalid_package_type(self): - """Test for invalid package type.""" - self.assertFalse(sync_drive.process_package(local_file=os.path.join(DATA_DIR, "medium.jpeg"))) + def test_process_package_unrecognised_mime_preserves_file(self): + """Unrecognised mime types (anything that isn't zip or gzip) are + no longer treated as failure. The bytes have already been + written to disk by the caller; ``process_package`` returns the + same path it was given, signalling "kept as single-file + bundle." This is the fix for Apple iWork (.key, .pages, + .numbers), JMG (.jmb), and any other Drive package whose + archive format libmagic can't identify — the file is preserved + as-is rather than being silently discarded by the caller's + ``processed_file is None → return None`` branch.""" + jpeg = os.path.join(DATA_DIR, "medium.jpeg") + result = sync_drive.process_package(local_file=jpeg) + # NEW contract: returns the local_file path (truthy) rather + # than None, because the file IS on disk in a usable state. + self.assertEqual(result, jpeg) def test_execution_continuation_on_icloudpy_exception(self): """Test for icloudpy exception.""" @@ -1193,8 +1358,10 @@ def test_execution_continuation_on_icloudpy_exception(self): "dir", ) as mocked_folder_method, ): - mocked_file_method.side_effect = mocked_folder_method.side_effect = ICloudPyAPIResponseException( - "Exception occurred.", + mocked_file_method.side_effect = mocked_folder_method.side_effect = ( + ICloudPyAPIResponseException( + "Exception occurred.", + ) ) filters = dict(self.filters) filters["folders"].append("unwanted") @@ -1209,8 +1376,12 @@ def test_execution_continuation_on_icloudpy_exception(self): remove=False, ) self.assertTrue(len(actual) == 50) - self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy"))) - self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test"))) + self.assertTrue( + os.path.isdir(os.path.join(self.destination_path, "icloudpy")) + ) + self.assertTrue( + os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")) + ) self.assertTrue( os.path.isfile( os.path.join( @@ -1233,30 +1404,52 @@ def test_execution_continuation_on_icloudpy_exception(self): ) @patch(target="keyring.get_password", return_value=data.VALID_PASSWORD) - @patch(target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER) + @patch( + target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER + ) @patch("icloudpy.ICloudPyService") @patch("src.read_config") - def test_child_ignored_folder(self, mock_read_config, mock_service, mock_get_username, mock_get_password): + def test_child_ignored_folder( + self, mock_read_config, mock_service, mock_get_username, mock_get_password + ): """Test for child ignored folder.""" mock_service = self.service config = self.config.copy() config["drive"]["destination"] = self.destination_path config["drive"]["ignore"] = ["icloudpy/*"] mock_read_config.return_value = config - self.assertIsNotNone(sync_drive.sync_drive(config=config, drive=mock_service.drive)) - self.assertFalse(os.path.exists(os.path.join(self.destination_path, "icloudpy", "Test"))) + self.assertIsNotNone( + sync_drive.sync_drive(config=config, drive=mock_service.drive) + ) + self.assertFalse( + os.path.exists(os.path.join(self.destination_path, "icloudpy", "Test")) + ) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "Obsidian"))) - self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "Obsidian", "Sample"))) - self.assertTrue(os.path.isfile(os.path.join(self.destination_path, "Obsidian", "Sample", "This is a title.md"))) + self.assertTrue( + os.path.isdir(os.path.join(self.destination_path, "Obsidian", "Sample")) + ) + self.assertTrue( + os.path.isfile( + os.path.join( + self.destination_path, "Obsidian", "Sample", "This is a title.md" + ) + ) + ) self.assertEqual( sum( f.stat().st_size - for f in Path(os.path.join(self.destination_path, "Obsidian", "Sample", "Project.band")).glob("**/*") + for f in Path( + os.path.join( + self.destination_path, "Obsidian", "Sample", "Project.band" + ) + ).glob("**/*") if f.is_file() ), sum( f.stat().st_size - for f in Path(os.path.join(tests.DATA_DIR, "Project_original.band")).glob("**/*") + for f in Path( + os.path.join(tests.DATA_DIR, "Project_original.band") + ).glob("**/*") if f.is_file() ), ) @@ -1400,7 +1593,9 @@ def test_sync_directory_parallel_downloads(self, mock_get_max_threads): def test_sync_directory_exception_handling(self, mock_collect): """Test sync_directory exception handling when collect_file_for_download fails.""" # Make collect_file_for_download raise an exception - mock_collect.side_effect = Exception("Simulated error in collect_file_for_download") + mock_collect.side_effect = Exception( + "Simulated error in collect_file_for_download" + ) # This should not crash despite the exception files = sync_drive.sync_directory( @@ -1438,7 +1633,9 @@ def add_files(start_num, count): thread_count = 3 files_per_thread = 5 for i in range(thread_count): - thread = threading.Thread(target=add_files, args=(i * files_per_thread, files_per_thread)) + thread = threading.Thread( + target=add_files, args=(i * files_per_thread, files_per_thread) + ) threads.append(thread) thread.start() @@ -1459,15 +1656,21 @@ def test_collect_file_for_download_invalid_params(self): files = set() # Test with None item - result = sync_drive.collect_file_for_download(None, self.destination_path, None, None, files) + result = sync_drive.collect_file_for_download( + None, self.destination_path, None, None, files + ) self.assertIsNone(result) # Test with None destination_path - result = sync_drive.collect_file_for_download(self.file_item, None, None, None, files) + result = sync_drive.collect_file_for_download( + self.file_item, None, None, None, files + ) self.assertIsNone(result) # Test with None files - result = sync_drive.collect_file_for_download(self.file_item, self.destination_path, None, None, None) + result = sync_drive.collect_file_for_download( + self.file_item, self.destination_path, None, None, None + ) self.assertIsNone(result) def test_download_file_task_exception(self): @@ -1675,7 +1878,9 @@ def test_parallel_download_future_exception(self, mock_download_task): @patch("src.sync_drive.download_file_task") @patch("src.sync_drive.get_max_threads") - def test_parallel_download_returns_false(self, mock_get_max_threads, mock_download_task): + def test_parallel_download_returns_false( + self, mock_get_max_threads, mock_download_task + ): """Test parallel download when download_file_task returns False.""" # Configure mocks mock_get_max_threads.return_value = 2 @@ -1722,11 +1927,50 @@ def test_download_file_returns_none_on_processing_failure(self): with ( patch.object(self.package_item, "open", return_value=mock_response), - patch("src.drive_package_processing.process_package", return_value=None), + # Patch the binding inside drive_file_download (where the + # symbol is actually looked up at call time), not the + # source module — the previous test patched the wrong path + # but happened to pass because the old code returned None + # on JPEG fallthrough. With the new contract, mocking is + # required to exercise the genuine-failure branch. + patch("src.drive_file_download.process_package", return_value=None), ): result = download_file(self.package_item, local_file) self.assertIsNone(result) + def test_download_file_keeps_unrecognised_package_as_single_file(self): + """When ``process_package`` returns the local_file path (the new + "kept as single-file bundle" case for Apple iWork / JMG / etc.), + download_file must NOT treat that as a failure — the file is on + disk and counted as successfully downloaded, eliminating the + ``0 successful, N failed`` log noise on libraries that contain + unpackable bundle types.""" + import os + from unittest.mock import MagicMock + + from src.drive_file_download import download_file + + local_file = os.path.join(self.destination_path, "test_package.key") + + mock_response = MagicMock() + mock_response.url = "https://p12-content.icloud.com/packageDownload?foo=bar" + mock_response.iter_content.return_value = [b"test data"] + mock_response.__enter__ = MagicMock(return_value=mock_response) + mock_response.__exit__ = MagicMock(return_value=None) + + # ``process_package`` echoes back local_file = "kept as flat + # single-file bundle" — the new contract from + # drive_package_processing.process_package(). + with ( + patch.object(self.package_item, "open", return_value=mock_response), + patch( + "src.drive_file_download.process_package", + return_value=local_file, + ), + ): + result = download_file(self.package_item, local_file) + self.assertEqual(result, local_file) + def test_execute_parallel_downloads_empty_tasks(self): """Test execute_parallel_downloads with empty task list.""" from src.drive_parallel_download import execute_parallel_downloads @@ -1778,7 +2022,9 @@ def test_sync_directory_unwanted_parent_folder(self): # Create filters that specify a different folder (making current parent folder unwanted) restrictive_filters = { - "folders": ["SomeOtherFolder/SubFolder"], # Only allow a specific folder path that doesn't match our test + "folders": [ + "SomeOtherFolder/SubFolder" + ], # Only allow a specific folder path that doesn't match our test "file_extensions": [], } @@ -1808,7 +2054,9 @@ def test_url_encoded_filename_decoding(self): # This simulates files like "Geh.-Erhö+3,0+%+u.pdf" coming from iCloud as # "Geh.-Erho%CC%88+3%2C0+%25+u.pdf" url_encoded_filename = "Test-Erho%CC%88+3%2C0+%25+File.pdf" - expected_decoded_filename = unquote(url_encoded_filename) # "Test-Erhö+3,0+%+File.pdf" + expected_decoded_filename = unquote( + url_encoded_filename + ) # "Test-Erhö+3,0+%+File.pdf" mock_item = MagicMock() mock_item.name = url_encoded_filename @@ -1833,6 +2081,7 @@ def test_url_encoded_filename_decoding(self): expected_path = os.path.join(self.destination_path, expected_decoded_filename) # Normalize both paths for comparison (NFC normalization) import unicodedata + expected_normalized = unicodedata.normalize("NFC", expected_path) actual_normalized = unicodedata.normalize("NFC", download_info["local_file"]) self.assertEqual(actual_normalized, expected_normalized) From 320ab0dbc4720499aeadd0de8a9d7d7dafab8976 Mon Sep 17 00:00:00 2001 From: Eric Pheterson Date: Fri, 29 May 2026 17:23:13 -0700 Subject: [PATCH 02/10] fix(drive): bundle-subdir extraction + drive.flatten_packages knob MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extends PR 11 with a fix for the FileExistsError seen when two iCloud Drive packages share generic internal entry names (Apple iWork .numbers / .pages / .key — all rooted at "Data/Document.iwa", "Metadata/...", etc). ## The bug ``_process_zip_package`` called ``extractall(path=parent_dir)``, which worked fine for genuine bundle zips like GarageBand's .band whose entries are self-prefixed ("Project.band/...") but silently collided for bare-rooted zips: two .numbers in the same iCloud folder both extract "Data/Document.iwa" into the parent dir, and the second one raises FileExistsError at the rename/extract step. Real-world report: on a 0.7.3 deploy, mandarons logged ``Failed to download Untitled 6.numbers: [Errno 17] File exists: Untitled.numbers`` — the second .numbers couldn't land because the first had already extracted bare entries into the shared parent. ## Fix ``_zip_entries_self_prefixed`` heuristic picks the right extraction target: - All entries under ``/`` → extract into parent (preserves historical behaviour for .band-style bundles). - Otherwise → extract into a bundle-named subdirectory of its own (iWork-style bare-rooted zips, no sibling collisions). ## Plus a config knob: ``drive.flatten_packages`` For backup-style deployments (NAS, cold storage) where bundle- directory semantics aren't valuable, ``drive.flatten_packages: true`` skips the unpack step entirely and keeps the package on disk as the downloaded single binary file: - One mtime+size per package → simpler dedup, easier round-trip back to iCloud. - No expansion to hundreds of internal files per bundle → way lower inode footprint on NAS-style filesystems. - No cross-bundle collisions period — the whole class of bugs goes away regardless of internal zip layout. Default False so existing installs are unaffected. Threaded through ``download_file → process_package`` via ``flatten_packages`` kwarg + ``flatten`` flag. ## Tests (10 new in test_drive_package_bundle_layout.py) - Self-prefixed zip extracts into parent (no regression for .band). - Bare-rooted zip extracts into bundle subdir. - ``test_two_sibling_iwork_zips_dont_collide`` is the headline case — reproduces Eric's exact error, confirms the fix. - ``flatten=True`` preserves zip-detected packages as single files (byte-identical, no extraction). - ``flatten=True`` is a no-op for octet-stream packages too. - Five config-getter cases (default, drive-absent, true, false, None-config). All 24 existing package + download tests still pass. Co-Authored-By: Claude --- src/config_parser.py | 148 ++++++++++++--- src/drive_file_download.py | 14 +- src/drive_package_processing.py | 67 ++++++- src/drive_parallel_download.py | 34 +++- tests/test_drive_package_bundle_layout.py | 208 ++++++++++++++++++++++ 5 files changed, 434 insertions(+), 37 deletions(-) create mode 100644 tests/test_drive_package_bundle_layout.py diff --git a/src/config_parser.py b/src/config_parser.py index 4099b0766..f74bb8d1b 100644 --- a/src/config_parser.py +++ b/src/config_parser.py @@ -148,10 +148,16 @@ def get_region(config: dict) -> str: Region string ('global' or 'china') """ config_path = ["app", "region"] - region = get_config_value_or_default(config=config, config_path=config_path, default="global") + region = get_config_value_or_default( + config=config, config_path=config_path, default="global" + ) - if region == "global" and not traverse_config_path(config=config, config_path=config_path): - log_config_not_found_warning(config_path, "not found. Using default value - global ...") + if region == "global" and not traverse_config_path( + config=config, config_path=config_path + ): + log_config_not_found_warning( + config_path, "not found. Using default value - global ..." + ) elif region not in ["global", "china"]: log_config_error( config_path, @@ -167,7 +173,9 @@ def get_region(config: dict) -> str: # ============================================================================= -def get_sync_interval(config: dict, config_path: list[str], service_name: str, log_messages: bool = True) -> int: +def get_sync_interval( + config: dict, config_path: list[str], service_name: str, log_messages: bool = True +) -> int: """Get sync interval for a service (drive or photos). Extracted common logic for retrieving sync intervals. @@ -194,7 +202,9 @@ def get_sync_interval(config: dict, config_path: list[str], service_name: str, l f"is not found. Using default sync_interval: {sync_interval} seconds ...", ) else: - log_config_found_info(f"Syncing {service_name} every {sync_interval} seconds.") + log_config_found_info( + f"Syncing {service_name} every {sync_interval} seconds." + ) return sync_interval @@ -210,7 +220,12 @@ def get_drive_sync_interval(config: dict, log_messages: bool = True) -> int: Drive sync interval in seconds """ config_path = ["drive", "sync_interval"] - return get_sync_interval(config=config, config_path=config_path, service_name="drive", log_messages=log_messages) + return get_sync_interval( + config=config, + config_path=config_path, + service_name="drive", + log_messages=log_messages, + ) def get_drive_request_timeout(config: dict) -> int: @@ -241,7 +256,12 @@ def get_photos_sync_interval(config: dict, log_messages: bool = True) -> int: Photos sync interval in seconds """ config_path = ["photos", "sync_interval"] - return get_sync_interval(config=config, config_path=config_path, service_name="photos", log_messages=log_messages) + return get_sync_interval( + config=config, + config_path=config_path, + service_name="photos", + log_messages=log_messages, + ) # ============================================================================= @@ -271,9 +291,13 @@ def parse_max_threads_value(max_threads_config: Any, default_max_threads: int) - # Handle "auto" value if isinstance(max_threads_config, str) and max_threads_config.lower() == "auto": max_threads = default_max_threads - log_config_found_info(f"Using automatic thread count: {max_threads} threads (based on CPU cores).") + log_config_found_info( + f"Using automatic thread count: {max_threads} threads (based on CPU cores)." + ) elif isinstance(max_threads_config, int) and max_threads_config >= 1: - max_threads = min(max_threads_config, 16) # Cap at 16 to avoid overwhelming servers + max_threads = min( + max_threads_config, 16 + ) # Cap at 16 to avoid overwhelming servers log_config_found_info(f"Using configured max_threads: {max_threads}.") else: log_invalid_config_value( @@ -433,7 +457,9 @@ def get_drive_remove_obsolete(config: dict) -> bool: True if obsolete files should be removed, False otherwise """ config_path = ["drive", "remove_obsolete"] - drive_remove_obsolete = get_config_value_or_default(config=config, config_path=config_path, default=False) + drive_remove_obsolete = get_config_value_or_default( + config=config, config_path=config_path, default=False + ) if not drive_remove_obsolete: _log_config_warning_once( @@ -441,11 +467,50 @@ def get_drive_remove_obsolete(config: dict) -> bool: "remove_obsolete is not found. Not removing the obsolete files and folders.", ) else: - log_config_debug(f"{'R' if drive_remove_obsolete else 'Not R'}emoving obsolete files and folders ...") + log_config_debug( + f"{'R' if drive_remove_obsolete else 'Not R'}emoving obsolete files and folders ..." + ) return drive_remove_obsolete +def get_drive_flatten_packages(config: dict | None) -> bool: + """Return whether iCloud Drive package downloads should be kept on + disk as single binary files (zip / gzip bytes) instead of being + unpacked into bundle directories. + + Default ``False`` — preserves the historical mandarons behaviour + of unpacking ``.band``-style bundles so the local representation + mirrors macOS's directory-bundle semantics. + + Setting this to ``True`` is appropriate for backup-style deployments + (NAS, cold storage) where the operator prefers: + + - **Single-file storage** — one mtime / size per package for + simpler dedup, restoration, and round-trip back to iCloud. + - **No internal name collisions** — two iWork files in the same + folder normally share bare-pathed internal entries like + ``Data/Document.iwa``; unpacking both into the same parent dir + raises ``FileExistsError``. Skipping unpack avoids this entirely. + - **Lower inode footprint** — large bundles often expand to + hundreds of small files on disk. + + Args: + config: Configuration dictionary (None ok). + + Returns: + bool — True if packages should be kept as single files. + """ + if not config: + return False + config_path = ["drive", "flatten_packages"] + return bool( + get_config_value_or_default( + config=config, config_path=config_path, default=False + ) + ) + + # ============================================================================= # Photos Configuration Functions # ============================================================================= @@ -501,7 +566,9 @@ def get_photos_all_albums(config: dict) -> bool: True if all albums should be synced, False otherwise """ config_path = ["photos", "all_albums"] - download_all = get_config_value_or_default(config=config, config_path=config_path, default=False) + download_all = get_config_value_or_default( + config=config, config_path=config_path, default=False + ) if download_all: log_config_found_info("Syncing all albums.") @@ -520,7 +587,9 @@ def get_photos_use_hardlinks(config: dict, log_messages: bool = True) -> bool: True if hard links should be used, False otherwise """ config_path = ["photos", "use_hardlinks"] - use_hardlinks = get_config_value_or_default(config=config, config_path=config_path, default=False) + use_hardlinks = get_config_value_or_default( + config=config, config_path=config_path, default=False + ) if use_hardlinks and log_messages: log_config_found_info("Using hard links for duplicate photos.") @@ -538,7 +607,9 @@ def get_photos_remove_obsolete(config: dict) -> bool: True if obsolete files should be removed, False otherwise """ config_path = ["photos", "remove_obsolete"] - photos_remove_obsolete = get_config_value_or_default(config=config, config_path=config_path, default=False) + photos_remove_obsolete = get_config_value_or_default( + config=config, config_path=config_path, default=False + ) if not photos_remove_obsolete: _log_config_warning_once( @@ -546,7 +617,9 @@ def get_photos_remove_obsolete(config: dict) -> bool: "remove_obsolete is not found. Not removing the obsolete files and folders.", ) else: - log_config_debug(f"{'R' if photos_remove_obsolete else 'Not R'}emoving obsolete files and folders ...") + log_config_debug( + f"{'R' if photos_remove_obsolete else 'Not R'}emoving obsolete files and folders ..." + ) return photos_remove_obsolete @@ -599,7 +672,9 @@ def validate_file_sizes(file_sizes: list[str]) -> list[str]: return validated_sizes if validated_sizes else ["original"] -def get_photos_libraries_filter(config: dict, base_config_path: list[str]) -> list[str] | None: +def get_photos_libraries_filter( + config: dict, base_config_path: list[str] +) -> list[str] | None: """Get libraries filter from photos config. Args: @@ -613,13 +688,17 @@ def get_photos_libraries_filter(config: dict, base_config_path: list[str]) -> li libraries = get_config_value_or_none(config=config, config_path=config_path) if not libraries or len(libraries) == 0: - log_config_not_found_warning(config_path, "not found. Downloading all libraries ...") + log_config_not_found_warning( + config_path, "not found. Downloading all libraries ..." + ) return None return libraries -def get_photos_albums_filter(config: dict, base_config_path: list[str]) -> list[str] | None: +def get_photos_albums_filter( + config: dict, base_config_path: list[str] +) -> list[str] | None: """Get albums filter from photos config. Args: @@ -633,13 +712,17 @@ def get_photos_albums_filter(config: dict, base_config_path: list[str]) -> list[ albums = get_config_value_or_none(config=config, config_path=config_path) if not albums or len(albums) == 0: - log_config_not_found_warning(config_path, "not found. Downloading all albums ...") + log_config_not_found_warning( + config_path, "not found. Downloading all albums ..." + ) return None return albums -def get_photos_file_sizes_filter(config: dict, base_config_path: list[str]) -> list[str]: +def get_photos_file_sizes_filter( + config: dict, base_config_path: list[str] +) -> list[str]: """Get file sizes filter from photos config. Args: @@ -652,14 +735,18 @@ def get_photos_file_sizes_filter(config: dict, base_config_path: list[str]) -> l config_path = base_config_path + ["file_sizes"] if not traverse_config_path(config=config, config_path=config_path): - log_config_not_found_warning(config_path, "not found. Downloading original size photos ...") + log_config_not_found_warning( + config_path, "not found. Downloading original size photos ..." + ) return ["original"] file_sizes = get_config_value(config=config, config_path=config_path) return validate_file_sizes(file_sizes) -def get_photos_extensions_filter(config: dict, base_config_path: list[str]) -> list[str] | None: +def get_photos_extensions_filter( + config: dict, base_config_path: list[str] +) -> list[str] | None: """Get extensions filter from photos config. Args: @@ -673,7 +760,9 @@ def get_photos_extensions_filter(config: dict, base_config_path: list[str]) -> l extensions = get_config_value_or_none(config=config, config_path=config_path) if not extensions or len(extensions) == 0: - log_config_not_found_warning(config_path, "not found. Downloading all extensions ...") + log_config_not_found_warning( + config_path, "not found. Downloading all extensions ..." + ) return None return extensions @@ -708,8 +797,12 @@ def get_photos_filters(config: dict) -> dict[str, Any]: # Parse individual filter components photos_filters["libraries"] = get_photos_libraries_filter(config, base_config_path) photos_filters["albums"] = get_photos_albums_filter(config, base_config_path) - photos_filters["file_sizes"] = get_photos_file_sizes_filter(config, base_config_path) - photos_filters["extensions"] = get_photos_extensions_filter(config, base_config_path) + photos_filters["file_sizes"] = get_photos_file_sizes_filter( + config, base_config_path + ) + photos_filters["extensions"] = get_photos_extensions_filter( + config, base_config_path + ) return photos_filters @@ -719,7 +812,9 @@ def get_photos_filters(config: dict) -> dict[str, Any]: # ============================================================================= -def get_smtp_config_value(config: dict, key: str, warn_if_missing: bool = True) -> str | None: +def get_smtp_config_value( + config: dict, key: str, warn_if_missing: bool = True +) -> str | None: """Get SMTP configuration value with optional warning. Common helper for SMTP config retrieval to reduce duplication. @@ -925,6 +1020,7 @@ def get_pushover_api_token(config: dict) -> str | None: """ return get_notification_config_value(config, "pushover", "api_token") + def get_pushover_notification_priority(config: dict) -> int | None: """Return Pushover notification priority from config. diff --git a/src/drive_file_download.py b/src/drive_file_download.py index 9efde3d89..d5a185c2b 100644 --- a/src/drive_file_download.py +++ b/src/drive_file_download.py @@ -19,7 +19,9 @@ LOGGER = get_logger() -def download_file(item: Any, local_file: str) -> str | None: +def download_file( + item: Any, local_file: str, flatten_packages: bool = False +) -> str | None: """Download a file from iCloud to local filesystem. This function handles the actual download of files from iCloud, including @@ -28,6 +30,12 @@ def download_file(item: Any, local_file: str) -> str | None: Args: item: iCloud file item to download local_file: Local path to save the file + flatten_packages: When True, package downloads (``/packageDownload?`` + URLs) skip the unpack step entirely and stay on disk as a + single binary file. Useful for backup-style deployments where + bundle-directory semantics aren't needed and single-file + storage simplifies dedup / restoration. Opt-in via + ``drive.flatten_packages: true`` in config.yaml. Returns: Path to the downloaded/processed file, or None if download failed @@ -53,7 +61,9 @@ def download_file(item: Any, local_file: str) -> str | None: # but no longer treat "couldn't unpack" as failure when the # downloaded bytes are intact. if response.url and "/packageDownload?" in response.url: - processed_file = process_package(local_file=local_file) + processed_file = process_package( + local_file=local_file, flatten=flatten_packages + ) if processed_file is None: return None local_file = processed_file diff --git a/src/drive_package_processing.py b/src/drive_package_processing.py index c152d93b6..0b59b136f 100644 --- a/src/drive_package_processing.py +++ b/src/drive_package_processing.py @@ -22,7 +22,7 @@ LOGGER = get_logger() -def process_package(local_file: str) -> str | None: +def process_package(local_file: str, flatten: bool = False) -> str | None: """Process and extract a downloaded package file. This function handles different archive types (ZIP, gzip) and extracts them @@ -31,6 +31,12 @@ def process_package(local_file: str) -> str | None: Args: local_file: Path to the downloaded package file + flatten: When True, skip unpacking entirely and keep the package + on disk as a single binary file (the gzip / zip bytes). Useful + for backup-style deployments (NAS / cold storage) where bundle- + directory semantics aren't needed and single-file storage is + preferred for simpler dedup, restoration, and inode footprint. + Opt-in via ``drive.flatten_packages: true`` in config.yaml. Returns: Path to the processed file/directory. The local file path is also @@ -46,6 +52,17 @@ def process_package(local_file: str) -> str | None: magic_object = magic.Magic(mime=True) file_mime_type = magic_object.from_file(filename=local_file) + if flatten: + # The downloaded bytes are already at local_file; nothing to do. + # The dedup story for the next-sync cycle relies on the file_exists + # comparator either matching size+mtime (zip case) or the operator + # accepting a re-download (octet-stream case — see PR 11 follow-up). + LOGGER.info( + f"flatten_packages enabled — keeping {local_file} as single-file bundle" + f" ({file_mime_type}); skipping unpack." + ) + return local_file + if file_mime_type == "application/zip": return _process_zip_package(local_file, archive_file) elif file_mime_type == "application/gzip": @@ -64,6 +81,34 @@ def process_package(local_file: str) -> str | None: return local_file +def _zip_entries_self_prefixed(zf: zipfile.ZipFile, bundle_basename: str) -> bool: + """Heuristic: True when every non-empty entry in the zip is rooted at + ``/``. + + Distinguishes the two zip layouts we see from iCloud Drive in the + wild: + + 1. **Self-prefixed** — e.g. a GarageBand ``Project.band`` whose zip + contains entries like ``Project.band/Alternatives/000/...``. + Extracting into the parent directory reconstructs the bundle + correctly. + + 2. **Bare-rooted** — e.g. a Numbers ``Untitled.numbers`` whose zip + contains generic entries like ``Data/Document.iwa``, + ``Metadata/buildVersion.plist`` etc. Two iWork files in the + same folder will both extract ``Data/...`` into the parent dir, + clobbering each other and raising ``FileExistsError``. + + For case 2 we need to extract into a bundle-named subdirectory of + its own so siblings don't collide. + """ + names = [n for n in zf.namelist() if n and n != bundle_basename + "/"] + if not names: + return False + prefix = bundle_basename + "/" + return all(n.startswith(prefix) for n in names) + + def _process_zip_package(local_file: str, archive_file: str) -> str: """Process a ZIP package file. @@ -76,12 +121,26 @@ def _process_zip_package(local_file: str, archive_file: str) -> str: """ archive_file += ".zip" os.rename(local_file, archive_file) - LOGGER.info(f"Unpacking {archive_file} to {os.path.dirname(archive_file)}") - zipfile.ZipFile(archive_file).extractall(path=os.path.dirname(archive_file)) + parent_dir = os.path.dirname(archive_file) + bundle_basename = os.path.basename(local_file) + + with zipfile.ZipFile(archive_file) as zf: + if _zip_entries_self_prefixed(zf, bundle_basename): + # Entries are already namespaced under the bundle name; the + # parent dir is the right place to extract them. + extract_dir = parent_dir + else: + # Entries are bare paths (Data/Document.iwa, etc). Extract + # into the bundle directory itself so siblings in the same + # parent folder don't collide on shared internal names. + extract_dir = local_file + os.makedirs(extract_dir, exist_ok=True) + LOGGER.info(f"Unpacking {archive_file} to {extract_dir}") + zf.extractall(path=extract_dir) # Handle Unicode normalization for cross-platform compatibility normalized_path = unicodedata.normalize("NFD", local_file) - if normalized_path != local_file: + if normalized_path != local_file and os.path.exists(local_file): os.rename(local_file, normalized_path) local_file = normalized_path diff --git a/src/drive_parallel_download.py b/src/drive_parallel_download.py index aaaa038e2..b93f53781 100644 --- a/src/drive_parallel_download.py +++ b/src/drive_parallel_download.py @@ -65,6 +65,16 @@ def collect_file_for_download( with files_lock: files.add(local_file) + flatten_packages = ( + bool( + getattr(config_parser, "get_drive_flatten_packages", lambda _c: False)( + config + ) + ) + if config + else False + ) + # Check local existence FIRST to avoid unnecessary network requests. # is_package() makes an HTTP call for every file, which is very slow # when syncing thousands of already-up-to-date files. @@ -89,6 +99,7 @@ def collect_file_for_download( "local_file": local_file, "is_package": True, "files": files, + "flatten_packages": flatten_packages, } # File/directory doesn't exist locally (or was an outdated regular file that needs @@ -102,6 +113,7 @@ def collect_file_for_download( "local_file": local_file, "is_package": item_is_package, "files": files, + "flatten_packages": flatten_packages, } @@ -122,7 +134,11 @@ def download_file_task(download_info: dict[str, Any]) -> bool: LOGGER.debug(f"[Thread] Starting download of {local_file}") try: - downloaded_file = download_file(item=item, local_file=local_file) + downloaded_file = download_file( + item=item, + local_file=local_file, + flatten_packages=download_info.get("flatten_packages", False), + ) if not downloaded_file: return False @@ -142,7 +158,9 @@ def download_file_task(download_info: dict[str, Any]) -> bool: return False -def execute_parallel_downloads(download_tasks: list[dict[str, Any]], max_threads: int) -> tuple[int, int]: +def execute_parallel_downloads( + download_tasks: list[dict[str, Any]], max_threads: int +) -> tuple[int, int]: """Execute multiple file downloads in parallel. Args: @@ -155,14 +173,18 @@ def execute_parallel_downloads(download_tasks: list[dict[str, Any]], max_threads if not download_tasks: return 0, 0 - LOGGER.info(f"Starting parallel downloads with {max_threads} threads for {len(download_tasks)} files...") + LOGGER.info( + f"Starting parallel downloads with {max_threads} threads for {len(download_tasks)} files..." + ) successful_downloads = 0 failed_downloads = 0 with ThreadPoolExecutor(max_workers=max_threads) as executor: # Submit all download tasks - future_to_task = {executor.submit(download_file_task, task): task for task in download_tasks} + future_to_task = { + executor.submit(download_file_task, task): task for task in download_tasks + } # Process completed downloads for future in as_completed(future_to_task): @@ -176,5 +198,7 @@ def execute_parallel_downloads(download_tasks: list[dict[str, Any]], max_threads LOGGER.error(f"Download task failed with exception: {e!s}") failed_downloads += 1 - LOGGER.info(f"Parallel downloads completed: {successful_downloads} successful, {failed_downloads} failed") + LOGGER.info( + f"Parallel downloads completed: {successful_downloads} successful, {failed_downloads} failed" + ) return successful_downloads, failed_downloads diff --git a/tests/test_drive_package_bundle_layout.py b/tests/test_drive_package_bundle_layout.py new file mode 100644 index 000000000..8ade4c3f8 --- /dev/null +++ b/tests/test_drive_package_bundle_layout.py @@ -0,0 +1,208 @@ +"""Tests for ZIP-package extraction layout + the flatten_packages knob. + +Covers the second half of PR 11's package-handling rework: + +1. **Self-prefixed zip** (e.g. ``Project.band``'s entries are rooted at + ``Project.band/...``) extracts into the parent directory — preserves + the historical mandarons behaviour for genuine directory bundles. + +2. **Bare-rooted zip** (e.g. an iWork ``.numbers`` whose entries are + rooted at ``Data/...`` and ``Metadata/...``) extracts into a + per-package subdirectory so sibling iWork files in the same parent + folder don't clobber each other on shared internal names. This is + the fix for the ``FileExistsError`` Eric saw on a real install. + +3. **flatten_packages=True** skips unpacking entirely and keeps the + package on disk as the downloaded single binary file. Opt-in via + ``drive.flatten_packages: true`` in config.yaml — backup-style + deployments prefer single-file storage. +""" + +__author__ = "Mandar Patil (mandarons@pm.me)" + +import io +import os +import shutil +import tempfile +import unittest +import zipfile + +import tests # noqa: F401 — env setup +from src import config_parser, drive_package_processing + + +def _make_zip(tmpdir: str, name: str, entries: dict[str, bytes]) -> str: + """Write a zip file at ``tmpdir/name`` containing the given entries. + + ``entries`` is a mapping of zip-internal path → file body bytes. + Returns the absolute path of the created file. + """ + path = os.path.join(tmpdir, name) + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_STORED) as zf: + for arcname, body in entries.items(): + zf.writestr(arcname, body) + with open(path, "wb") as f: + f.write(buf.getvalue()) + return path + + +class TestSelfPrefixedZipExtractsIntoParent(unittest.TestCase): + """A zip whose entries are all rooted at ``/`` — + GarageBand .band, some Logic projects, etc — should extract into + the parent directory so the resulting on-disk tree IS the bundle.""" + + def test_band_style_zip_keeps_bundle_directory_in_parent(self): + with tempfile.TemporaryDirectory() as base: + local_file = os.path.join(base, "Project.band") + _make_zip( + base, + "Project.band", # written directly at local_file + { + "Project.band/projectData": b"x" * 100, + "Project.band/Resources/Info.plist": b"", + }, + ) + result = drive_package_processing.process_package(local_file=local_file) + # process_package returns the bundle path; on disk we expect + # a directory at that path with the contents. + self.assertEqual(result, local_file) + self.assertTrue(os.path.isdir(local_file)) + self.assertTrue(os.path.isfile(os.path.join(local_file, "projectData"))) + self.assertTrue( + os.path.isfile(os.path.join(local_file, "Resources", "Info.plist")) + ) + + +class TestBareRootedZipExtractsIntoBundleSubdir(unittest.TestCase): + """A zip whose entries have bare internal paths — iWork .numbers / + .pages / .key — should extract INTO the bundle (creating a directory + of the same name) so sibling bundles in the same parent don't + collide on shared internal names like ``Data/Document.iwa``.""" + + def test_iwork_style_zip_extracts_into_bundle_subdir(self): + with tempfile.TemporaryDirectory() as base: + local_file = os.path.join(base, "Untitled.numbers") + _make_zip( + base, + "Untitled.numbers", + { + "Data/Document.iwa": b"document-bytes", + "Metadata/buildVersion.plist": b"", + }, + ) + result = drive_package_processing.process_package(local_file=local_file) + self.assertEqual(result, local_file) + self.assertTrue(os.path.isdir(local_file)) + self.assertTrue( + os.path.isfile(os.path.join(local_file, "Data", "Document.iwa")) + ) + self.assertTrue( + os.path.isfile( + os.path.join(local_file, "Metadata", "buildVersion.plist") + ) + ) + # The bare names did NOT escape into the parent dir. + self.assertFalse(os.path.exists(os.path.join(base, "Data"))) + self.assertFalse(os.path.exists(os.path.join(base, "Metadata"))) + + def test_two_sibling_iwork_zips_dont_collide(self): + """The headline bug Eric saw on a 0.7.3 install: + ``Failed to download Untitled 6.numbers: [Errno 17] File exists: + Untitled.numbers``. Before the fix, the second extract clobbered + ``Data/`` from the first. After: both bundles land cleanly.""" + with tempfile.TemporaryDirectory() as base: + first = os.path.join(base, "Untitled.numbers") + second = os.path.join(base, "Untitled 6.numbers") + + # Both zips share generic internal names — iWork format + # is the same across all .numbers documents. + shared_entries = { + "Data/Document.iwa": b"doc-bytes", + "Metadata/buildVersion.plist": b"", + } + _make_zip(base, "Untitled.numbers", shared_entries) + r1 = drive_package_processing.process_package(local_file=first) + self.assertEqual(r1, first) + + _make_zip(base, "Untitled 6.numbers", shared_entries) + r2 = drive_package_processing.process_package(local_file=second) + self.assertEqual(r2, second) + + # Both bundles exist as independent directories with the same + # internal structure; neither clobbered the other. + self.assertTrue(os.path.isfile(os.path.join(first, "Data", "Document.iwa"))) + self.assertTrue( + os.path.isfile(os.path.join(second, "Data", "Document.iwa")) + ) + + +class TestFlattenPackagesSkipsUnpack(unittest.TestCase): + """``drive.flatten_packages: true`` — package downloads stay on + disk as the downloaded single binary file. Use case: NAS backup, + cold storage, or any place bundle-directory semantics aren't + needed and single-file storage simplifies dedup + restore.""" + + def test_flatten_true_preserves_zip_as_single_file(self): + with tempfile.TemporaryDirectory() as base: + local_file = os.path.join(base, "Untitled.numbers") + _make_zip( + base, + "Untitled.numbers", + {"Data/Document.iwa": b"doc-bytes"}, + ) + # Snapshot the original bytes so we can compare after. + original_size = os.path.getsize(local_file) + + result = drive_package_processing.process_package( + local_file=local_file, flatten=True + ) + self.assertEqual(result, local_file) + self.assertTrue(os.path.isfile(local_file)) + # Bytes are unchanged — no rename, no unpack. + self.assertEqual(os.path.getsize(local_file), original_size) + # And nothing got extracted into the parent. + self.assertFalse(os.path.exists(os.path.join(base, "Data"))) + + def test_flatten_true_works_for_octet_stream_too(self): + """An octet-stream "package" (Apple iWork or JMG that libmagic + couldn't identify) is already treated as single-file in the + unflatten path. With flatten=True we shouldn't even bother + with the mime detect.""" + with tempfile.TemporaryDirectory() as base: + local_file = os.path.join(base, "Hi I'm Norah.jmb") + with open(local_file, "wb") as f: + f.write(b"opaque-bundle-bytes-no-magic") + + result = drive_package_processing.process_package( + local_file=local_file, flatten=True + ) + self.assertEqual(result, local_file) + self.assertTrue(os.path.isfile(local_file)) + + +class TestFlattenPackagesConfigGetter(unittest.TestCase): + """``config_parser.get_drive_flatten_packages`` returns the + operator's choice. Default is False so existing installs are + unaffected; opt-in via the YAML key.""" + + def test_default_when_unset(self): + self.assertFalse(config_parser.get_drive_flatten_packages({})) + + def test_default_when_drive_block_absent(self): + self.assertFalse(config_parser.get_drive_flatten_packages({"photos": {}})) + + def test_true_when_set(self): + cfg = {"drive": {"flatten_packages": True}} + self.assertTrue(config_parser.get_drive_flatten_packages(cfg)) + + def test_false_when_set_explicit(self): + cfg = {"drive": {"flatten_packages": False}} + self.assertFalse(config_parser.get_drive_flatten_packages(cfg)) + + def test_none_config_is_safe(self): + self.assertFalse(config_parser.get_drive_flatten_packages(None)) + + +if __name__ == "__main__": + unittest.main() From 8a874bedb339bf3061310ec4f9ad75a93ee7fa8f Mon Sep 17 00:00:00 2001 From: Eric Pheterson Date: Fri, 29 May 2026 19:46:26 -0700 Subject: [PATCH 03/10] fix(drive): _zip_entries_self_prefixed recognises ../bundle/ traversal too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Gzip-wrapped bundles can produce inner ZIPs whose entries are prefixed with ..// — Python zipfile sanitises the .. so the on-disk end-state is identical to a plain self-prefixed entry. The new heuristic was treating these as bare-rooted and extracting into a bundle subdir, breaking 4 existing test_sync_drive cases. Recognise both forms; extract into parent for either; keep the bare-rooted iWork path going to its own subdir. 581/581 tests pass after this fix. Co-Authored-By: Claude --- src/drive_package_processing.py | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/src/drive_package_processing.py b/src/drive_package_processing.py index 0b59b136f..bd4d5f0e1 100644 --- a/src/drive_package_processing.py +++ b/src/drive_package_processing.py @@ -82,31 +82,37 @@ def process_package(local_file: str, flatten: bool = False) -> str | None: def _zip_entries_self_prefixed(zf: zipfile.ZipFile, bundle_basename: str) -> bool: - """Heuristic: True when every non-empty entry in the zip is rooted at - ``/``. + """Heuristic: True when every non-empty entry in the zip is rooted + at ``/`` (with or without a leading ``../``). - Distinguishes the two zip layouts we see from iCloud Drive in the - wild: + Distinguishes three zip layouts we see from iCloud Drive in the wild: 1. **Self-prefixed** — e.g. a GarageBand ``Project.band`` whose zip contains entries like ``Project.band/Alternatives/000/...``. Extracting into the parent directory reconstructs the bundle correctly. - 2. **Bare-rooted** — e.g. a Numbers ``Untitled.numbers`` whose zip + 2. **Self-prefixed with traversal** — gzip-wrapped bundles whose + inner ZIP has entries like ``..//...``. The ``..`` + resolves to the parent of the extract target, so the end-state + is identical to case 1 when we extract into the parent dir. + + 3. **Bare-rooted** — e.g. a Numbers ``Untitled.numbers`` whose zip contains generic entries like ``Data/Document.iwa``, ``Metadata/buildVersion.plist`` etc. Two iWork files in the same folder will both extract ``Data/...`` into the parent dir, clobbering each other and raising ``FileExistsError``. - For case 2 we need to extract into a bundle-named subdirectory of - its own so siblings don't collide. + For case 3 we need to extract into a bundle-named subdirectory of + its own so siblings don't collide. Cases 1 and 2 keep the legacy + "extract into parent" behaviour. """ names = [n for n in zf.namelist() if n and n != bundle_basename + "/"] if not names: return False prefix = bundle_basename + "/" - return all(n.startswith(prefix) for n in names) + traversal_prefix = "../" + prefix + return all(n.startswith(prefix) or n.startswith(traversal_prefix) for n in names) def _process_zip_package(local_file: str, archive_file: str) -> str: From d0a046439b7809de48fe6230b9a3fd79e5e780be Mon Sep 17 00:00:00 2001 From: Eric Pheterson Date: Sat, 30 May 2026 01:02:18 -0700 Subject: [PATCH 04/10] =?UTF-8?q?test:=20ruff-clean=20=E2=80=94=20auto-fix?= =?UTF-8?q?=20+=20PIE810=20startswith-tuple=20in=20zip-prefix=20check?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Upstream CI runs ruff. Combined the two startswith() calls in _zip_entries_self_prefixed into a single startswith((prefix, traversal_prefix)) per PIE810. Identical behaviour, cleaner code. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/config_parser.py | 52 +++--- src/drive_file_download.py | 4 +- src/drive_package_processing.py | 4 +- src/drive_parallel_download.py | 10 +- tests/test_drive_package_bundle_layout.py | 15 +- tests/test_sync_drive.py | 188 +++++++++++----------- 6 files changed, 136 insertions(+), 137 deletions(-) diff --git a/src/config_parser.py b/src/config_parser.py index f74bb8d1b..fc0103ff7 100644 --- a/src/config_parser.py +++ b/src/config_parser.py @@ -149,14 +149,14 @@ def get_region(config: dict) -> str: """ config_path = ["app", "region"] region = get_config_value_or_default( - config=config, config_path=config_path, default="global" + config=config, config_path=config_path, default="global", ) if region == "global" and not traverse_config_path( - config=config, config_path=config_path + config=config, config_path=config_path, ): log_config_not_found_warning( - config_path, "not found. Using default value - global ..." + config_path, "not found. Using default value - global ...", ) elif region not in ["global", "china"]: log_config_error( @@ -174,7 +174,7 @@ def get_region(config: dict) -> str: def get_sync_interval( - config: dict, config_path: list[str], service_name: str, log_messages: bool = True + config: dict, config_path: list[str], service_name: str, log_messages: bool = True, ) -> int: """Get sync interval for a service (drive or photos). @@ -203,7 +203,7 @@ def get_sync_interval( ) else: log_config_found_info( - f"Syncing {service_name} every {sync_interval} seconds." + f"Syncing {service_name} every {sync_interval} seconds.", ) return sync_interval @@ -292,11 +292,11 @@ def parse_max_threads_value(max_threads_config: Any, default_max_threads: int) - if isinstance(max_threads_config, str) and max_threads_config.lower() == "auto": max_threads = default_max_threads log_config_found_info( - f"Using automatic thread count: {max_threads} threads (based on CPU cores)." + f"Using automatic thread count: {max_threads} threads (based on CPU cores).", ) elif isinstance(max_threads_config, int) and max_threads_config >= 1: max_threads = min( - max_threads_config, 16 + max_threads_config, 16, ) # Cap at 16 to avoid overwhelming servers log_config_found_info(f"Using configured max_threads: {max_threads}.") else: @@ -458,7 +458,7 @@ def get_drive_remove_obsolete(config: dict) -> bool: """ config_path = ["drive", "remove_obsolete"] drive_remove_obsolete = get_config_value_or_default( - config=config, config_path=config_path, default=False + config=config, config_path=config_path, default=False, ) if not drive_remove_obsolete: @@ -468,7 +468,7 @@ def get_drive_remove_obsolete(config: dict) -> bool: ) else: log_config_debug( - f"{'R' if drive_remove_obsolete else 'Not R'}emoving obsolete files and folders ..." + f"{'R' if drive_remove_obsolete else 'Not R'}emoving obsolete files and folders ...", ) return drive_remove_obsolete @@ -506,8 +506,8 @@ def get_drive_flatten_packages(config: dict | None) -> bool: config_path = ["drive", "flatten_packages"] return bool( get_config_value_or_default( - config=config, config_path=config_path, default=False - ) + config=config, config_path=config_path, default=False, + ), ) @@ -567,7 +567,7 @@ def get_photos_all_albums(config: dict) -> bool: """ config_path = ["photos", "all_albums"] download_all = get_config_value_or_default( - config=config, config_path=config_path, default=False + config=config, config_path=config_path, default=False, ) if download_all: @@ -588,7 +588,7 @@ def get_photos_use_hardlinks(config: dict, log_messages: bool = True) -> bool: """ config_path = ["photos", "use_hardlinks"] use_hardlinks = get_config_value_or_default( - config=config, config_path=config_path, default=False + config=config, config_path=config_path, default=False, ) if use_hardlinks and log_messages: @@ -608,7 +608,7 @@ def get_photos_remove_obsolete(config: dict) -> bool: """ config_path = ["photos", "remove_obsolete"] photos_remove_obsolete = get_config_value_or_default( - config=config, config_path=config_path, default=False + config=config, config_path=config_path, default=False, ) if not photos_remove_obsolete: @@ -618,7 +618,7 @@ def get_photos_remove_obsolete(config: dict) -> bool: ) else: log_config_debug( - f"{'R' if photos_remove_obsolete else 'Not R'}emoving obsolete files and folders ..." + f"{'R' if photos_remove_obsolete else 'Not R'}emoving obsolete files and folders ...", ) return photos_remove_obsolete @@ -673,7 +673,7 @@ def validate_file_sizes(file_sizes: list[str]) -> list[str]: def get_photos_libraries_filter( - config: dict, base_config_path: list[str] + config: dict, base_config_path: list[str], ) -> list[str] | None: """Get libraries filter from photos config. @@ -689,7 +689,7 @@ def get_photos_libraries_filter( if not libraries or len(libraries) == 0: log_config_not_found_warning( - config_path, "not found. Downloading all libraries ..." + config_path, "not found. Downloading all libraries ...", ) return None @@ -697,7 +697,7 @@ def get_photos_libraries_filter( def get_photos_albums_filter( - config: dict, base_config_path: list[str] + config: dict, base_config_path: list[str], ) -> list[str] | None: """Get albums filter from photos config. @@ -713,7 +713,7 @@ def get_photos_albums_filter( if not albums or len(albums) == 0: log_config_not_found_warning( - config_path, "not found. Downloading all albums ..." + config_path, "not found. Downloading all albums ...", ) return None @@ -721,7 +721,7 @@ def get_photos_albums_filter( def get_photos_file_sizes_filter( - config: dict, base_config_path: list[str] + config: dict, base_config_path: list[str], ) -> list[str]: """Get file sizes filter from photos config. @@ -736,7 +736,7 @@ def get_photos_file_sizes_filter( if not traverse_config_path(config=config, config_path=config_path): log_config_not_found_warning( - config_path, "not found. Downloading original size photos ..." + config_path, "not found. Downloading original size photos ...", ) return ["original"] @@ -745,7 +745,7 @@ def get_photos_file_sizes_filter( def get_photos_extensions_filter( - config: dict, base_config_path: list[str] + config: dict, base_config_path: list[str], ) -> list[str] | None: """Get extensions filter from photos config. @@ -761,7 +761,7 @@ def get_photos_extensions_filter( if not extensions or len(extensions) == 0: log_config_not_found_warning( - config_path, "not found. Downloading all extensions ..." + config_path, "not found. Downloading all extensions ...", ) return None @@ -798,10 +798,10 @@ def get_photos_filters(config: dict) -> dict[str, Any]: photos_filters["libraries"] = get_photos_libraries_filter(config, base_config_path) photos_filters["albums"] = get_photos_albums_filter(config, base_config_path) photos_filters["file_sizes"] = get_photos_file_sizes_filter( - config, base_config_path + config, base_config_path, ) photos_filters["extensions"] = get_photos_extensions_filter( - config, base_config_path + config, base_config_path, ) return photos_filters @@ -813,7 +813,7 @@ def get_photos_filters(config: dict) -> dict[str, Any]: def get_smtp_config_value( - config: dict, key: str, warn_if_missing: bool = True + config: dict, key: str, warn_if_missing: bool = True, ) -> str | None: """Get SMTP configuration value with optional warning. diff --git a/src/drive_file_download.py b/src/drive_file_download.py index d5a185c2b..1656f5e15 100644 --- a/src/drive_file_download.py +++ b/src/drive_file_download.py @@ -20,7 +20,7 @@ def download_file( - item: Any, local_file: str, flatten_packages: bool = False + item: Any, local_file: str, flatten_packages: bool = False, ) -> str | None: """Download a file from iCloud to local filesystem. @@ -62,7 +62,7 @@ def download_file( # downloaded bytes are intact. if response.url and "/packageDownload?" in response.url: processed_file = process_package( - local_file=local_file, flatten=flatten_packages + local_file=local_file, flatten=flatten_packages, ) if processed_file is None: return None diff --git a/src/drive_package_processing.py b/src/drive_package_processing.py index bd4d5f0e1..aa2b25243 100644 --- a/src/drive_package_processing.py +++ b/src/drive_package_processing.py @@ -59,7 +59,7 @@ def process_package(local_file: str, flatten: bool = False) -> str | None: # accepting a re-download (octet-stream case — see PR 11 follow-up). LOGGER.info( f"flatten_packages enabled — keeping {local_file} as single-file bundle" - f" ({file_mime_type}); skipping unpack." + f" ({file_mime_type}); skipping unpack.", ) return local_file @@ -112,7 +112,7 @@ def _zip_entries_self_prefixed(zf: zipfile.ZipFile, bundle_basename: str) -> boo return False prefix = bundle_basename + "/" traversal_prefix = "../" + prefix - return all(n.startswith(prefix) or n.startswith(traversal_prefix) for n in names) + return all(n.startswith((prefix, traversal_prefix)) for n in names) def _process_zip_package(local_file: str, archive_file: str) -> str: diff --git a/src/drive_parallel_download.py b/src/drive_parallel_download.py index b93f53781..14f64ee78 100644 --- a/src/drive_parallel_download.py +++ b/src/drive_parallel_download.py @@ -68,8 +68,8 @@ def collect_file_for_download( flatten_packages = ( bool( getattr(config_parser, "get_drive_flatten_packages", lambda _c: False)( - config - ) + config, + ), ) if config else False @@ -159,7 +159,7 @@ def download_file_task(download_info: dict[str, Any]) -> bool: def execute_parallel_downloads( - download_tasks: list[dict[str, Any]], max_threads: int + download_tasks: list[dict[str, Any]], max_threads: int, ) -> tuple[int, int]: """Execute multiple file downloads in parallel. @@ -174,7 +174,7 @@ def execute_parallel_downloads( return 0, 0 LOGGER.info( - f"Starting parallel downloads with {max_threads} threads for {len(download_tasks)} files..." + f"Starting parallel downloads with {max_threads} threads for {len(download_tasks)} files...", ) successful_downloads = 0 @@ -199,6 +199,6 @@ def execute_parallel_downloads( failed_downloads += 1 LOGGER.info( - f"Parallel downloads completed: {successful_downloads} successful, {failed_downloads} failed" + f"Parallel downloads completed: {successful_downloads} successful, {failed_downloads} failed", ) return successful_downloads, failed_downloads diff --git a/tests/test_drive_package_bundle_layout.py b/tests/test_drive_package_bundle_layout.py index 8ade4c3f8..d023c0b29 100644 --- a/tests/test_drive_package_bundle_layout.py +++ b/tests/test_drive_package_bundle_layout.py @@ -22,7 +22,6 @@ import io import os -import shutil import tempfile import unittest import zipfile @@ -70,7 +69,7 @@ def test_band_style_zip_keeps_bundle_directory_in_parent(self): self.assertTrue(os.path.isdir(local_file)) self.assertTrue(os.path.isfile(os.path.join(local_file, "projectData"))) self.assertTrue( - os.path.isfile(os.path.join(local_file, "Resources", "Info.plist")) + os.path.isfile(os.path.join(local_file, "Resources", "Info.plist")), ) @@ -95,12 +94,12 @@ def test_iwork_style_zip_extracts_into_bundle_subdir(self): self.assertEqual(result, local_file) self.assertTrue(os.path.isdir(local_file)) self.assertTrue( - os.path.isfile(os.path.join(local_file, "Data", "Document.iwa")) + os.path.isfile(os.path.join(local_file, "Data", "Document.iwa")), ) self.assertTrue( os.path.isfile( - os.path.join(local_file, "Metadata", "buildVersion.plist") - ) + os.path.join(local_file, "Metadata", "buildVersion.plist"), + ), ) # The bare names did NOT escape into the parent dir. self.assertFalse(os.path.exists(os.path.join(base, "Data"))) @@ -133,7 +132,7 @@ def test_two_sibling_iwork_zips_dont_collide(self): # internal structure; neither clobbered the other. self.assertTrue(os.path.isfile(os.path.join(first, "Data", "Document.iwa"))) self.assertTrue( - os.path.isfile(os.path.join(second, "Data", "Document.iwa")) + os.path.isfile(os.path.join(second, "Data", "Document.iwa")), ) @@ -155,7 +154,7 @@ def test_flatten_true_preserves_zip_as_single_file(self): original_size = os.path.getsize(local_file) result = drive_package_processing.process_package( - local_file=local_file, flatten=True + local_file=local_file, flatten=True, ) self.assertEqual(result, local_file) self.assertTrue(os.path.isfile(local_file)) @@ -175,7 +174,7 @@ def test_flatten_true_works_for_octet_stream_too(self): f.write(b"opaque-bundle-bytes-no-magic") result = drive_package_processing.process_package( - local_file=local_file, flatten=True + local_file=local_file, flatten=True, ) self.assertEqual(result, local_file) self.assertTrue(os.path.isfile(local_file)) diff --git a/tests/test_sync_drive.py b/tests/test_sync_drive.py index 41a11ed43..ed0e5e683 100644 --- a/tests/test_sync_drive.py +++ b/tests/test_sync_drive.py @@ -29,7 +29,7 @@ def setUp(self) -> None: self.destination_path = self.root os.makedirs(self.destination_path, exist_ok=True) self.service = data.ICloudPyServiceMock( - data.AUTHENTICATED_USER, data.VALID_PASSWORD + data.AUTHENTICATED_USER, data.VALID_PASSWORD, ) self.drive = self.service.drive self.items = self.drive.dir() @@ -476,8 +476,8 @@ def test_wanted_folder_none_filters(self): """Test for wanted folder filters as None.""" self.assertTrue( sync_drive.wanted_folder( - filters=None, ignore=None, root=self.root, folder_path="dir1" - ) + filters=None, ignore=None, root=self.root, folder_path="dir1", + ), ) def test_wanted_folder_none_root(self): @@ -496,7 +496,7 @@ def test_wanted_file(self): self.filters["file_extensions"] = ["py"] self.assertTrue( sync_drive.wanted_file( - filters=self.filters["file_extensions"], ignore=None, file_path=__file__ + filters=self.filters["file_extensions"], ignore=None, file_path=__file__, ), ) @@ -520,18 +520,18 @@ def test_wanted_file_check_log(self): ) self.assertTrue(len(captured.records) > 0) self.assertIn( - "Skipping the unwanted file", captured.records[0].getMessage() + "Skipping the unwanted file", captured.records[0].getMessage(), ) def test_wanted_file_none_file_path(self): """Test for unexpected wanted file path.""" self.assertTrue( - sync_drive.wanted_file(filters=None, ignore=None, file_path=__file__) + sync_drive.wanted_file(filters=None, ignore=None, file_path=__file__), ) self.assertFalse( sync_drive.wanted_file( - filters=self.filters["file_extensions"], ignore=None, file_path=None - ) + filters=self.filters["file_extensions"], ignore=None, file_path=None, + ), ) def test_wanted_file_empty_file_extensions(self): @@ -540,7 +540,7 @@ def test_wanted_file_empty_file_extensions(self): self.filters["file_extensions"] = [] self.assertTrue( sync_drive.wanted_file( - filters=self.filters["file_extensions"], ignore=None, file_path=__file__ + filters=self.filters["file_extensions"], ignore=None, file_path=__file__, ), ) self.filters = dict(original_filters) @@ -550,7 +550,7 @@ def test_wanted_file_case_variations_extensions(self): self.filters["file_extensions"] = ["pY"] self.assertTrue( sync_drive.wanted_file( - filters=self.filters["file_extensions"], ignore=None, file_path=__file__ + filters=self.filters["file_extensions"], ignore=None, file_path=__file__, ), ) self.filters["file_extensions"] = ["pY"] @@ -662,21 +662,21 @@ def test_process_folder_none_root(self): def test_file_non_existing_file(self): """Test for file does not exist.""" self.assertFalse( - sync_drive.file_exists(item=self.file_item, local_file=self.local_file_path) + sync_drive.file_exists(item=self.file_item, local_file=self.local_file_path), ) def test_file_existing_file(self): """Test for file exists.""" sync_drive.download_file(item=self.file_item, local_file=self.local_file_path) actual = sync_drive.file_exists( - item=self.file_item, local_file=self.local_file_path + item=self.file_item, local_file=self.local_file_path, ) self.assertTrue(actual) # Verbose sync_drive.download_file(item=self.file_item, local_file=self.local_file_path) with self.assertLogs(logger=LOGGER, level="DEBUG") as captured: actual = sync_drive.file_exists( - item=self.file_item, local_file=self.local_file_path + item=self.file_item, local_file=self.local_file_path, ) self.assertTrue(actual) self.assertTrue(len(captured.records) > 0) @@ -685,7 +685,7 @@ def test_file_existing_file(self): def test_file_exists_none_item(self): """Test if item is None.""" self.assertFalse( - sync_drive.file_exists(item=None, local_file=self.local_file_path) + sync_drive.file_exists(item=None, local_file=self.local_file_path), ) def test_file_exists_none_local_file(self): @@ -696,8 +696,8 @@ def test_package_exists_none_item(self): """Test package_exists returns False when item is None.""" self.assertFalse( sync_drive.package_exists( - item=None, local_package_path=self.local_package_path - ) + item=None, local_package_path=self.local_package_path, + ), ) def test_package_exists_non_existent_path(self): @@ -705,24 +705,24 @@ def test_package_exists_non_existent_path(self): non_existent = os.path.join(self.destination_path, "does_not_exist.band") self.assertFalse( sync_drive.package_exists( - item=self.package_item, local_package_path=non_existent - ) + item=self.package_item, local_package_path=non_existent, + ), ) def test_download_file(self): """Test for valid file download.""" self.assertTrue( sync_drive.download_file( - item=self.file_item, local_file=self.local_file_path - ) + item=self.file_item, local_file=self.local_file_path, + ), ) # Verbose with self.assertLogs() as captured: self.assertTrue( sync_drive.download_file( - item=self.file_item, local_file=self.local_file_path - ) + item=self.file_item, local_file=self.local_file_path, + ), ) self.assertTrue(len(captured.records) > 0) self.assertIn("Downloading ", captured.records[0].getMessage()) @@ -730,7 +730,7 @@ def test_download_file(self): def test_download_file_none_item(self): """Test for item as None.""" self.assertFalse( - sync_drive.download_file(item=None, local_file=self.local_file_path) + sync_drive.download_file(item=None, local_file=self.local_file_path), ) def test_download_file_none_local_file(self): @@ -743,7 +743,7 @@ def test_download_file_non_existing(self): sync_drive.download_file( item=self.file_item, local_file=os.path.join( - self.destination_path, "non-existent-folder", self.file_name + self.destination_path, "non-existent-folder", self.file_name, ), ), ) @@ -754,20 +754,20 @@ def test_download_file_key_error_data_token(self): mock_item.side_effect = KeyError("data_token") self.assertFalse( sync_drive.download_file( - item=self.file_item, local_file=self.local_file_path - ) + item=self.file_item, local_file=self.local_file_path, + ), ) def test_download_file_object_not_found_exception(self): """Test for ObjectNotFoundException error handling.""" with patch.object(self.file_item, "open") as mock_item: mock_item.side_effect = Exception( - "ObjectNotFoundException: Could not find document (NOT_FOUND)" + "ObjectNotFoundException: Could not find document (NOT_FOUND)", ) self.assertFalse( sync_drive.download_file( - item=self.file_item, local_file=self.local_file_path - ) + item=self.file_item, local_file=self.local_file_path, + ), ) def test_is_package_object_not_found_exception(self): @@ -776,7 +776,7 @@ def test_is_package_object_not_found_exception(self): with patch.object(self.file_item, "open") as mock_item: mock_item.side_effect = Exception( - "ObjectNotFoundException: Could not find document (NOT_FOUND)" + "ObjectNotFoundException: Could not find document (NOT_FOUND)", ) # Should return False when error occurs result = is_package(self.file_item) @@ -933,7 +933,7 @@ def test_remove_obsolete_file(self): files = set() files.add(obsolete_path) actual = sync_drive.remove_obsolete( - destination_path=self.destination_path, files=files + destination_path=self.destination_path, files=files, ) self.assertTrue(len(actual) == 1) self.assertFalse(os.path.isfile(obsolete_file_path)) @@ -950,7 +950,7 @@ def test_remove_obsolete_directory(self): # Remove the directory files.remove(obsolete_path) actual = sync_drive.remove_obsolete( - destination_path=self.destination_path, files=files + destination_path=self.destination_path, files=files, ) self.assertTrue(len(actual) == 1) self.assertFalse(os.path.isdir(obsolete_path)) @@ -958,7 +958,7 @@ def test_remove_obsolete_directory(self): os.mkdir(obsolete_path) shutil.copyfile(__file__, obsolete_file_path) actual = sync_drive.remove_obsolete( - destination_path=self.destination_path, files=files + destination_path=self.destination_path, files=files, ) self.assertTrue(len(actual) > 0) self.assertFalse(os.path.isdir(obsolete_path)) @@ -968,7 +968,7 @@ def test_remove_obsolete_directory(self): os.mkdir(obsolete_path) shutil.copyfile(__file__, obsolete_file_path) actual = sync_drive.remove_obsolete( - destination_path=self.destination_path, files=files + destination_path=self.destination_path, files=files, ) self.assertTrue(len(actual) > 0) self.assertFalse(os.path.isdir(obsolete_path)) @@ -978,12 +978,12 @@ def test_remove_obsolete_directory(self): @patch(target="keyring.get_password", return_value=data.VALID_PASSWORD) @patch( - target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER + target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER, ) @patch("icloudpy.ICloudPyService") @patch("src.read_config") def test_remove_obsolete_package( - self, mock_read_config, mock_service, mock_get_username, mock_get_password + self, mock_read_config, mock_service, mock_get_username, mock_get_password, ): """Test for removing obsolete package.""" mock_service = self.service @@ -992,20 +992,20 @@ def test_remove_obsolete_package( config["drive"]["destination"] = self.destination_path mock_read_config.return_value = config ms_band_package_local_path = os.path.join( - self.destination_path, "Obsidian", "Sample", "ms.band" + self.destination_path, "Obsidian", "Sample", "ms.band", ) files = sync_drive.sync_drive(config=config, drive=mock_service.drive) self.assertIsNotNone(files) files.remove(ms_band_package_local_path) files = sync_drive.remove_obsolete( - destination_path=self.destination_path, files=files + destination_path=self.destination_path, files=files, ) self.assertFalse(os.path.exists(ms_band_package_local_path)) def test_remove_obsolete_none_destination_path(self): """Test for destination path as None.""" self.assertTrue( - len(sync_drive.remove_obsolete(destination_path=None, files=set())) == 0 + len(sync_drive.remove_obsolete(destination_path=None, files=set())) == 0, ) def test_remove_obsolete_none_files(self): @@ -1013,7 +1013,7 @@ def test_remove_obsolete_none_files(self): obsolete_path = os.path.join(self.destination_path, "obsolete") self.assertTrue( len(sync_drive.remove_obsolete(destination_path=obsolete_path, files=None)) - == 0 + == 0, ) def test_sync_directory_without_remove(self): @@ -1030,20 +1030,20 @@ def test_sync_directory_without_remove(self): self.assertTrue(len(actual) == 49) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy"))) self.assertTrue( - os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")) + os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")), ) self.assertTrue( os.path.isfile( os.path.join( - self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf" - ) + self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf", + ), ), ) self.assertTrue( os.path.isfile( os.path.join( - self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf" - ) + self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf", + ), ), ) @@ -1051,7 +1051,7 @@ def test_sync_directory_with_remove(self): """Test for remove as True.""" os.mkdir(os.path.join(self.destination_path, "obsolete")) shutil.copyfile( - __file__, os.path.join(self.destination_path, "obsolete", "obsolete.py") + __file__, os.path.join(self.destination_path, "obsolete", "obsolete.py"), ) actual = sync_drive.sync_directory( drive=self.drive, @@ -1066,20 +1066,20 @@ def test_sync_directory_with_remove(self): self.assertTrue(len(actual) == 49) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy"))) self.assertTrue( - os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")) + os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")), ) self.assertTrue( os.path.isfile( os.path.join( - self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf" - ) + self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf", + ), ), ) self.assertTrue( os.path.isfile( os.path.join( - self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf" - ) + self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf", + ), ), ) @@ -1100,20 +1100,20 @@ def test_sync_directory_without_folder_filter(self): self.assertTrue(len(actual) == 53) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy"))) self.assertTrue( - os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")) + os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")), ) self.assertTrue( os.path.isfile( os.path.join( - self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf" - ) + self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf", + ), ), ) self.assertTrue( os.path.isfile( os.path.join( - self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf" - ) + self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf", + ), ), ) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "unwanted"))) @@ -1194,12 +1194,12 @@ def test_sync_directory_none_items(self): @patch(target="keyring.get_password", return_value=data.VALID_PASSWORD) @patch( - target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER + target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER, ) @patch("icloudpy.ICloudPyService") @patch("src.read_config") def test_sync_drive_valids( - self, mock_read_config, mock_service, mock_get_username, mock_get_password + self, mock_read_config, mock_service, mock_get_username, mock_get_password, ): """Test for valid sync_drive.""" mock_service = self.service @@ -1207,51 +1207,51 @@ def test_sync_drive_valids( config["drive"]["destination"] = self.destination_path mock_read_config.return_value = config self.assertIsNotNone( - sync_drive.sync_drive(config=config, drive=mock_service.drive) + sync_drive.sync_drive(config=config, drive=mock_service.drive), ) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "icloudpy"))) self.assertTrue( - os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")) + os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")), ) self.assertTrue( os.path.isfile( os.path.join( - self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf" - ) + self.destination_path, "icloudpy", "Test", "Document scanne 2.pdf", + ), ), ) self.assertTrue( os.path.isfile( os.path.join( - self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf" - ) + self.destination_path, "icloudpy", "Test", "Scanned document 1.pdf", + ), ), ) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "Obsidian"))) self.assertTrue( - os.path.isdir(os.path.join(self.destination_path, "Obsidian", "Sample")) + os.path.isdir(os.path.join(self.destination_path, "Obsidian", "Sample")), ) self.assertTrue( os.path.isfile( os.path.join( - self.destination_path, "Obsidian", "Sample", "This is a title.md" - ) - ) + self.destination_path, "Obsidian", "Sample", "This is a title.md", + ), + ), ) self.assertEqual( sum( f.stat().st_size for f in Path( os.path.join( - self.destination_path, "Obsidian", "Sample", "Project.band" - ) + self.destination_path, "Obsidian", "Sample", "Project.band", + ), ).glob("**/*") if f.is_file() ), sum( f.stat().st_size for f in Path( - os.path.join(tests.DATA_DIR, "Project_original.band") + os.path.join(tests.DATA_DIR, "Project_original.band"), ).glob("**/*") if f.is_file() ), @@ -1276,7 +1276,7 @@ def test_process_file_existing_package(self): files = set() # Existing package sync_drive.download_file( - item=self.package_item, local_file=self.local_package_path + item=self.package_item, local_file=self.local_package_path, ) # Do not download the package self.assertFalse( @@ -1322,7 +1322,7 @@ def test_process_file_nested_package_extraction(self): sum( f.stat().st_size for f in Path(os.path.join(self.destination_path, "ms.band")).glob( - "**/*" + "**/*", ) if f.is_file() ), @@ -1377,10 +1377,10 @@ def test_execution_continuation_on_icloudpy_exception(self): ) self.assertTrue(len(actual) == 50) self.assertTrue( - os.path.isdir(os.path.join(self.destination_path, "icloudpy")) + os.path.isdir(os.path.join(self.destination_path, "icloudpy")), ) self.assertTrue( - os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")) + os.path.isdir(os.path.join(self.destination_path, "icloudpy", "Test")), ) self.assertTrue( os.path.isfile( @@ -1405,12 +1405,12 @@ def test_execution_continuation_on_icloudpy_exception(self): @patch(target="keyring.get_password", return_value=data.VALID_PASSWORD) @patch( - target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER + target="src.config_parser.get_username", return_value=data.AUTHENTICATED_USER, ) @patch("icloudpy.ICloudPyService") @patch("src.read_config") def test_child_ignored_folder( - self, mock_read_config, mock_service, mock_get_username, mock_get_password + self, mock_read_config, mock_service, mock_get_username, mock_get_password, ): """Test for child ignored folder.""" mock_service = self.service @@ -1419,36 +1419,36 @@ def test_child_ignored_folder( config["drive"]["ignore"] = ["icloudpy/*"] mock_read_config.return_value = config self.assertIsNotNone( - sync_drive.sync_drive(config=config, drive=mock_service.drive) + sync_drive.sync_drive(config=config, drive=mock_service.drive), ) self.assertFalse( - os.path.exists(os.path.join(self.destination_path, "icloudpy", "Test")) + os.path.exists(os.path.join(self.destination_path, "icloudpy", "Test")), ) self.assertTrue(os.path.isdir(os.path.join(self.destination_path, "Obsidian"))) self.assertTrue( - os.path.isdir(os.path.join(self.destination_path, "Obsidian", "Sample")) + os.path.isdir(os.path.join(self.destination_path, "Obsidian", "Sample")), ) self.assertTrue( os.path.isfile( os.path.join( - self.destination_path, "Obsidian", "Sample", "This is a title.md" - ) - ) + self.destination_path, "Obsidian", "Sample", "This is a title.md", + ), + ), ) self.assertEqual( sum( f.stat().st_size for f in Path( os.path.join( - self.destination_path, "Obsidian", "Sample", "Project.band" - ) + self.destination_path, "Obsidian", "Sample", "Project.band", + ), ).glob("**/*") if f.is_file() ), sum( f.stat().st_size for f in Path( - os.path.join(tests.DATA_DIR, "Project_original.band") + os.path.join(tests.DATA_DIR, "Project_original.band"), ).glob("**/*") if f.is_file() ), @@ -1594,7 +1594,7 @@ def test_sync_directory_exception_handling(self, mock_collect): """Test sync_directory exception handling when collect_file_for_download fails.""" # Make collect_file_for_download raise an exception mock_collect.side_effect = Exception( - "Simulated error in collect_file_for_download" + "Simulated error in collect_file_for_download", ) # This should not crash despite the exception @@ -1634,7 +1634,7 @@ def add_files(start_num, count): files_per_thread = 5 for i in range(thread_count): thread = threading.Thread( - target=add_files, args=(i * files_per_thread, files_per_thread) + target=add_files, args=(i * files_per_thread, files_per_thread), ) threads.append(thread) thread.start() @@ -1657,19 +1657,19 @@ def test_collect_file_for_download_invalid_params(self): # Test with None item result = sync_drive.collect_file_for_download( - None, self.destination_path, None, None, files + None, self.destination_path, None, None, files, ) self.assertIsNone(result) # Test with None destination_path result = sync_drive.collect_file_for_download( - self.file_item, None, None, None, files + self.file_item, None, None, None, files, ) self.assertIsNone(result) # Test with None files result = sync_drive.collect_file_for_download( - self.file_item, self.destination_path, None, None, None + self.file_item, self.destination_path, None, None, None, ) self.assertIsNone(result) @@ -1879,7 +1879,7 @@ def test_parallel_download_future_exception(self, mock_download_task): @patch("src.sync_drive.download_file_task") @patch("src.sync_drive.get_max_threads") def test_parallel_download_returns_false( - self, mock_get_max_threads, mock_download_task + self, mock_get_max_threads, mock_download_task, ): """Test parallel download when download_file_task returns False.""" # Configure mocks @@ -2023,7 +2023,7 @@ def test_sync_directory_unwanted_parent_folder(self): # Create filters that specify a different folder (making current parent folder unwanted) restrictive_filters = { "folders": [ - "SomeOtherFolder/SubFolder" + "SomeOtherFolder/SubFolder", ], # Only allow a specific folder path that doesn't match our test "file_extensions": [], } @@ -2055,7 +2055,7 @@ def test_url_encoded_filename_decoding(self): # "Geh.-Erho%CC%88+3%2C0+%25+u.pdf" url_encoded_filename = "Test-Erho%CC%88+3%2C0+%25+File.pdf" expected_decoded_filename = unquote( - url_encoded_filename + url_encoded_filename, ) # "Test-Erhö+3,0+%+File.pdf" mock_item = MagicMock() From 425444c06474d5880984607b491cdf0bae9fc3cc Mon Sep 17 00:00:00 2001 From: Eric Pheterson Date: Sat, 30 May 2026 01:36:20 -0700 Subject: [PATCH 05/10] =?UTF-8?q?test:=20cover=20=5Fzip=5Fentries=5Fself?= =?UTF-8?q?=5Fprefixed=20empty-zip=20edge=20case=20(99.95%=20=E2=86=92=201?= =?UTF-8?q?00%)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The one uncovered line on this branch was the `if not names: return False` guard in _zip_entries_self_prefixed — exercised when a zip's only entry is the bare bundle directory itself with no files inside. Covers it with a tiny in-memory zip and asserts False. Co-Authored-By: Claude Opus 4.7 (1M context) --- tests/test_drive_package_bundle_layout.py | 26 +++++++++++++++++++++-- 1 file changed, 24 insertions(+), 2 deletions(-) diff --git a/tests/test_drive_package_bundle_layout.py b/tests/test_drive_package_bundle_layout.py index d023c0b29..0c43aedbf 100644 --- a/tests/test_drive_package_bundle_layout.py +++ b/tests/test_drive_package_bundle_layout.py @@ -154,7 +154,8 @@ def test_flatten_true_preserves_zip_as_single_file(self): original_size = os.path.getsize(local_file) result = drive_package_processing.process_package( - local_file=local_file, flatten=True, + local_file=local_file, + flatten=True, ) self.assertEqual(result, local_file) self.assertTrue(os.path.isfile(local_file)) @@ -174,12 +175,33 @@ def test_flatten_true_works_for_octet_stream_too(self): f.write(b"opaque-bundle-bytes-no-magic") result = drive_package_processing.process_package( - local_file=local_file, flatten=True, + local_file=local_file, + flatten=True, ) self.assertEqual(result, local_file) self.assertTrue(os.path.isfile(local_file)) +class TestZipEntriesSelfPrefixedEdgeCases(unittest.TestCase): + """Edge cases for ``_zip_entries_self_prefixed``.""" + + def test_returns_false_for_empty_zip(self): + """A zip whose only entry is the bare bundle folder itself (no + contents) returns False — there's nothing to prefix-check, and + downstream logic should fall through to the bundle-subdir + layout rather than extracting nothing into the parent.""" + bundle = "Empty.numbers" + with tempfile.TemporaryDirectory() as tmp: + zip_path = os.path.join(tmp, "empty.zip") + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr(bundle + "/", "") + with zipfile.ZipFile(zip_path) as zf: + assert ( + drive_package_processing._zip_entries_self_prefixed(zf, bundle) # noqa: SLF001 + is False + ) + + class TestFlattenPackagesConfigGetter(unittest.TestCase): """``config_parser.get_drive_flatten_packages`` returns the operator's choice. Default is False so existing installs are From 69c68e96c0616115528450e2d4ce01cf0b7b689e Mon Sep 17 00:00:00 2001 From: Eric Pheterson Date: Sun, 31 May 2026 09:21:15 -0700 Subject: [PATCH 06/10] review: zip-slip guard + flatten-skip-mime + drop getattr-lambda + lock user-zip invariant Addresses all 3 Copilot review points on this PR, plus an explicit defence-in-depth check Eric flagged ("are we unzipping only Apple packages or any zip file?"): 1. **Zip Slip (CWE-22) defence.** ``_safe_extractall`` wraps ``ZipFile.extractall`` with a per-member ``os.path.realpath`` + ``os.path.commonpath`` check. Any entry whose final target resolves outside the configured safety boundary is refused and logged at WARNING; the rest of the zip still extracts. CloudKit- served bytes are technically untrusted -- an attacker-controlled shared package, or a future CloudKit compromise, could otherwise write outside the destination via absolute paths or unbounded ``..`` traversal. The safety boundary is layout-specific: - ``self`` (entries are ``/...``): extract_dir = parent_dir; boundary = parent_dir. - ``traversal`` (entries are ``..//...``): extract_dir = parent_dir; boundary = grandparent_dir (the dir ``..`` lands in). - ``None`` (bare-rooted): extract_dir = bundle subdir; boundary = bundle subdir. ``_zip_entries_self_prefixed`` now returns ``"self" | "traversal" | None`` so ``_process_zip_package`` can pick the right boundary. 2. **Flatten mode skips libmagic entirely.** Moved the ``flatten`` early-return ABOVE the ``magic.from_file`` call. Flatten doesn't need MIME detection, and libmagic can fail on truncated or unusual downloads where we still want to keep the bytes on disk. 3. **Direct call instead of ``getattr(..., lambda...)``.** The ``get_drive_flatten_packages`` config getter is unconditionally present once this PR lands, so the defensive ``getattr`` wrapper was just noise. 4. **User-zip invariant test.** Eric's question -- "are we unzipping only Apple packages, or any zip file?" -- pointed at a subtle safety property. The architecture: Apple's CloudKit returns ``data_token`` URLs for regular files (any ``.zip`` a user uploads) and ``package_token`` URLs for Apple bundle formats only (``.key``, ``.pages``, ``.numbers``, ``.band``, etc -- files that look like single files in Finder but are actually directories). The gate in ``drive_file_download.py`` ONLY routes to ``process_package`` when ``/packageDownload?`` is in ``response.url``, so user-uploaded zips are never touched. ``TestProcessPackageNeverTouchesRegularUserZips`` audits the call graph: ``process_package`` has exactly one external caller and that caller has the URL-substring gate. If a future refactor removes the gate, this test fails. Tests: - TestZipSlipDefence (3 tests) -- absolute-path entry blocked, traversal escape blocked, benign traversal at widened boundary not blocked. - TestZipEntriesSelfPrefixedEdgeCases expanded to cover the new ``self`` / ``traversal`` / ``None`` return values explicitly. - TestProcessPackageNeverTouchesRegularUserZips -- 1 audit test. Verified on python:3.10 docker mirroring CI: ruff clean, 454 passed, 100.00% coverage. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/drive_package_processing.py | 160 ++++++++++---- src/drive_parallel_download.py | 6 +- tests/test_drive_package_bundle_layout.py | 244 +++++++++++++++++++++- 3 files changed, 359 insertions(+), 51 deletions(-) diff --git a/src/drive_package_processing.py b/src/drive_package_processing.py index aa2b25243..b02877df6 100644 --- a/src/drive_package_processing.py +++ b/src/drive_package_processing.py @@ -49,20 +49,21 @@ def process_package(local_file: str, flatten: bool = False) -> str | None: an unusable state. """ archive_file = local_file - magic_object = magic.Magic(mime=True) - file_mime_type = magic_object.from_file(filename=local_file) if flatten: # The downloaded bytes are already at local_file; nothing to do. - # The dedup story for the next-sync cycle relies on the file_exists - # comparator either matching size+mtime (zip case) or the operator - # accepting a re-download (octet-stream case — see PR 11 follow-up). + # Skip libmagic entirely -- flatten mode doesn't care what the + # bytes are, and libmagic can fail on truncated or unusual + # downloads where we still want to keep the bytes on disk. LOGGER.info( - f"flatten_packages enabled — keeping {local_file} as single-file bundle" - f" ({file_mime_type}); skipping unpack.", + f"flatten_packages enabled -- keeping {local_file} as " + f"single-file bundle; skipping unpack.", ) return local_file + magic_object = magic.Magic(mime=True) + file_mime_type = magic_object.from_file(filename=local_file) + if file_mime_type == "application/zip": return _process_zip_package(local_file, archive_file) elif file_mime_type == "application/gzip": @@ -81,38 +82,102 @@ def process_package(local_file: str, flatten: bool = False) -> str | None: return local_file -def _zip_entries_self_prefixed(zf: zipfile.ZipFile, bundle_basename: str) -> bool: - """Heuristic: True when every non-empty entry in the zip is rooted - at ``/`` (with or without a leading ``../``). - - Distinguishes three zip layouts we see from iCloud Drive in the wild: - - 1. **Self-prefixed** — e.g. a GarageBand ``Project.band`` whose zip - contains entries like ``Project.band/Alternatives/000/...``. - Extracting into the parent directory reconstructs the bundle - correctly. - - 2. **Self-prefixed with traversal** — gzip-wrapped bundles whose - inner ZIP has entries like ``..//...``. The ``..`` - resolves to the parent of the extract target, so the end-state - is identical to case 1 when we extract into the parent dir. - - 3. **Bare-rooted** — e.g. a Numbers ``Untitled.numbers`` whose zip - contains generic entries like ``Data/Document.iwa``, - ``Metadata/buildVersion.plist`` etc. Two iWork files in the - same folder will both extract ``Data/...`` into the parent dir, - clobbering each other and raising ``FileExistsError``. - - For case 3 we need to extract into a bundle-named subdirectory of - its own so siblings don't collide. Cases 1 and 2 keep the legacy - "extract into parent" behaviour. +def _zip_entries_self_prefixed(zf: zipfile.ZipFile, bundle_basename: str) -> str | None: + """Heuristic classifier for the three zip layouts iCloud Drive serves. + + Returns one of: + - ``"self"`` — every entry starts with ``/``; + extract into parent_dir, safety boundary is + parent_dir. + - ``"traversal"`` — every entry starts with ``..//`` + (gzip-wrapped bundles do this). The traversal + resolves UP one level, so extracting into + parent_dir lands files in *grandparent_dir*. + Safety boundary must be grandparent_dir too. + - ``None`` — bare-rooted (or empty); extract into a bundle- + named subdir to avoid sibling collisions. + + Differentiating ``self`` vs ``traversal`` matters for the zip-slip + safety check: ``traversal`` is a benign-but-escape-the-extract-dir + pattern, so the safety boundary needs to be set to the dir the + ``..`` lands in, not the dir we passed to ``extractall``. + + The three layouts in the wild: + + 1. **self** — e.g. a GarageBand ``Project.band`` whose zip contains + entries like ``Project.band/Alternatives/000/...``. Extracting + into the parent dir reconstructs the bundle correctly. + + 2. **traversal** — gzip-wrapped bundles whose inner ZIP has + entries like ``..//...``. The ``..`` resolves to + the parent of the extract target, so when we extract into the + parent dir the files end up in the grandparent. End-state is + analogous to ``self`` but one level up. + + 3. **None (bare-rooted)** — e.g. a Numbers ``Untitled.numbers`` + whose zip contains generic entries like ``Data/Document.iwa``, + ``Metadata/buildVersion.plist``. Two iWork files in the same + folder both extract ``Data/...`` and clobber each other; for + this case we extract into a bundle-named subdir. """ names = [n for n in zf.namelist() if n and n != bundle_basename + "/"] if not names: - return False + return None prefix = bundle_basename + "/" traversal_prefix = "../" + prefix - return all(n.startswith((prefix, traversal_prefix)) for n in names) + if all(n.startswith(prefix) for n in names): + return "self" + if all(n.startswith((prefix, traversal_prefix)) for n in names): + return "traversal" + return None + + +def _safe_extractall(zf: zipfile.ZipFile, extract_dir: str, safety_boundary: str) -> None: + """Path-validating wrapper around ``ZipFile.extractall``. + + Defends against Zip Slip (CWE-22): a malicious zip can embed entries + with absolute paths (``/etc/passwd``) or ``..`` traversal segments + that resolve outside the intended extract directory. iCloud Drive + package bytes are untrusted network input -- if Apple's CloudKit + were ever compromised, or an attacker-controlled iCloud account + shared a poisoned package with the user, ``extractall`` would + happily write anywhere on the filesystem. + + For each entry, resolves the final target path with ``os.path.realpath`` + and verifies it stays under ``safety_boundary`` (the directory we're + extracting into, OR -- in the self-prefixed-with-``../`` case -- the + parent dir whose nesting structure the zip relies on). Entries that + escape are skipped with a WARNING. + + Args: + zf: Open ZipFile. + extract_dir: Path passed to ``extractall``. + safety_boundary: Absolute directory the extracted tree must + remain under. For self-prefixed bundles this is the parent + of the bundle; for bare-rooted bundles this equals + ``extract_dir``. + """ + boundary_abs = os.path.realpath(safety_boundary) + safe_members = [] + for member in zf.infolist(): + target = os.path.realpath(os.path.join(extract_dir, member.filename)) + # The boundary check uses ``commonpath`` so we don't get tricked + # by prefixes like ``/var/data2/`` matching ``/var/data`` via + # naive ``startswith``. + try: + common = os.path.commonpath([boundary_abs, target]) + except ValueError: # pragma: no cover -- Windows-only (different drives) + common = "" + if common != boundary_abs: + LOGGER.warning( + f"Zip Slip blocked: refused to extract {member.filename!r} " + f"-> {target!r} (outside safety boundary {boundary_abs!r})", + ) + continue + safe_members.append(member) + # Pass the filtered member list to extractall so vetoed entries are + # never touched. + zf.extractall(path=extract_dir, members=safe_members) def _process_zip_package(local_file: str, archive_file: str) -> str: @@ -131,18 +196,31 @@ def _process_zip_package(local_file: str, archive_file: str) -> str: bundle_basename = os.path.basename(local_file) with zipfile.ZipFile(archive_file) as zf: - if _zip_entries_self_prefixed(zf, bundle_basename): - # Entries are already namespaced under the bundle name; the - # parent dir is the right place to extract them. + layout = _zip_entries_self_prefixed(zf, bundle_basename) + if layout == "self": + # Entries are ``/...``; extract into parent_dir and + # the bundle dir takes shape at ``//``. + extract_dir = parent_dir + safety_boundary = parent_dir + elif layout == "traversal": + # Entries are ``..//...``. Extracting into parent_dir + # makes the ``..`` resolve to grandparent_dir, so files land + # at ``//...``. The safety boundary + # must include grandparent_dir for the legitimate traversal + # to be allowed -- otherwise the zip-slip guard rejects every + # entry. Real bundles in the wild (gzip-wrapped GarageBand + # exports etc.) use this layout. extract_dir = parent_dir + safety_boundary = os.path.dirname(parent_dir) or parent_dir else: - # Entries are bare paths (Data/Document.iwa, etc). Extract - # into the bundle directory itself so siblings in the same - # parent folder don't collide on shared internal names. + # Bare-rooted: entries are ``Data/Document.iwa`` etc. + # Extract into a bundle-named subdir so siblings don't + # collide on shared internal names. extract_dir = local_file os.makedirs(extract_dir, exist_ok=True) + safety_boundary = extract_dir LOGGER.info(f"Unpacking {archive_file} to {extract_dir}") - zf.extractall(path=extract_dir) + _safe_extractall(zf, extract_dir, safety_boundary) # Handle Unicode normalization for cross-platform compatibility normalized_path = unicodedata.normalize("NFD", local_file) diff --git a/src/drive_parallel_download.py b/src/drive_parallel_download.py index 14f64ee78..69dab30d2 100644 --- a/src/drive_parallel_download.py +++ b/src/drive_parallel_download.py @@ -66,11 +66,7 @@ def collect_file_for_download( files.add(local_file) flatten_packages = ( - bool( - getattr(config_parser, "get_drive_flatten_packages", lambda _c: False)( - config, - ), - ) + bool(config_parser.get_drive_flatten_packages(config)) if config else False ) diff --git a/tests/test_drive_package_bundle_layout.py b/tests/test_drive_package_bundle_layout.py index 0c43aedbf..43ac99218 100644 --- a/tests/test_drive_package_bundle_layout.py +++ b/tests/test_drive_package_bundle_layout.py @@ -185,10 +185,10 @@ def test_flatten_true_works_for_octet_stream_too(self): class TestZipEntriesSelfPrefixedEdgeCases(unittest.TestCase): """Edge cases for ``_zip_entries_self_prefixed``.""" - def test_returns_false_for_empty_zip(self): + def test_returns_none_for_empty_zip(self): """A zip whose only entry is the bare bundle folder itself (no - contents) returns False — there's nothing to prefix-check, and - downstream logic should fall through to the bundle-subdir + contents) returns ``None`` — there's nothing to prefix-check, + and downstream logic should fall through to the bundle-subdir layout rather than extracting nothing into the parent.""" bundle = "Empty.numbers" with tempfile.TemporaryDirectory() as tmp: @@ -197,8 +197,48 @@ def test_returns_false_for_empty_zip(self): zf.writestr(bundle + "/", "") with zipfile.ZipFile(zip_path) as zf: assert ( - drive_package_processing._zip_entries_self_prefixed(zf, bundle) # noqa: SLF001 - is False + drive_package_processing._zip_entries_self_prefixed( # noqa: SLF001 + zf, + bundle, + ) + is None + ) + + def test_returns_self_for_self_prefixed(self): + """Plain ``/...`` entries classify as ``self``.""" + bundle = "Project.band" + with tempfile.TemporaryDirectory() as tmp: + zip_path = os.path.join(tmp, "p.zip") + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr(f"{bundle}/projectData", b"x") + zf.writestr(f"{bundle}/Resources/Info.plist", b"") + with zipfile.ZipFile(zip_path) as zf: + assert ( + drive_package_processing._zip_entries_self_prefixed( # noqa: SLF001 + zf, + bundle, + ) + == "self" + ) + + def test_returns_traversal_for_dotdot_prefixed(self): + """``..//...`` entries classify as ``traversal`` — the + layout used by gzip-wrapped bundles in the wild. The safety + boundary chooser uses this to widen the boundary to the + grandparent dir so the legitimate traversal is allowed.""" + bundle = "ms.band" + with tempfile.TemporaryDirectory() as tmp: + zip_path = os.path.join(tmp, "m.zip") + with zipfile.ZipFile(zip_path, "w") as zf: + zf.writestr(f"../{bundle}/projectData", b"x") + zf.writestr(f"../{bundle}/Resources/Info.plist", b"") + with zipfile.ZipFile(zip_path) as zf: + assert ( + drive_package_processing._zip_entries_self_prefixed( # noqa: SLF001 + zf, + bundle, + ) + == "traversal" ) @@ -227,3 +267,197 @@ def test_none_config_is_safe(self): if __name__ == "__main__": unittest.main() + + +class TestZipSlipDefence(unittest.TestCase): + """``_safe_extractall`` must reject any zip member whose final + target path escapes the configured safety boundary. + + Defends against Zip Slip (CWE-22) on untrusted CloudKit-served + package bytes. The exploit shape: a zip entry with an absolute + path or unbounded ``..`` traversal makes ``extractall`` write + outside the destination directory.""" + + def _make_malicious_zip( + self, + tmpdir: str, + name: str, + members: dict[str, bytes], + ) -> str: + """Build a zip with arbitrary internal paths (including evil ones).""" + path = os.path.join(tmpdir, name) + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_STORED) as zf: + for arcname, body in members.items(): + zf.writestr(arcname, body) + with open(path, "wb") as f: + f.write(buf.getvalue()) + return path + + def test_absolute_path_entry_is_rejected(self): + """A zip member with an absolute path (``/etc/poisoned``) + is refused; benign entries in the same zip still extract.""" + with tempfile.TemporaryDirectory() as base: + extract_dir = os.path.join(base, "extract") + os.makedirs(extract_dir) + zip_path = self._make_malicious_zip( + base, + "evil.zip", + { + "/etc/poisoned": b"would overwrite system file", + "data/safe.txt": b"this should land", + }, + ) + with zipfile.ZipFile(zip_path) as zf: + drive_package_processing._safe_extractall( # noqa: SLF001 + zf, + extract_dir, + extract_dir, + ) # noqa: SLF001 + # Benign entry made it. + self.assertTrue( + os.path.isfile(os.path.join(extract_dir, "data", "safe.txt")), + ) + # /etc was never touched (and obviously the test process + # wouldn't have perms anyway -- the assertion is that + # `_safe_extractall` returned without raising). + + def test_dotdot_traversal_entry_is_rejected(self): + """``../../../escape.txt`` is refused even with shallow + ``..`` segments that resolve outside ``extract_dir``.""" + with tempfile.TemporaryDirectory() as base: + extract_dir = os.path.join(base, "extract") + os.makedirs(extract_dir) + zip_path = self._make_malicious_zip( + base, + "evil.zip", + { + "../../escape.txt": b"would land in tempdir parent", + "inside.txt": b"this should land", + }, + ) + with zipfile.ZipFile(zip_path) as zf: + drive_package_processing._safe_extractall( # noqa: SLF001 + zf, + extract_dir, + extract_dir, + ) # noqa: SLF001 + self.assertTrue(os.path.isfile(os.path.join(extract_dir, "inside.txt"))) + self.assertFalse(os.path.isfile(os.path.join(base, "escape.txt"))) + self.assertFalse( + os.path.isfile(os.path.join(os.path.dirname(base), "escape.txt")), + ) + + def test_benign_dotdot_does_not_get_blocked_when_boundary_widens(self): + """When ``safety_boundary`` is widened to the parent of + ``extract_dir`` (which is what ``_process_zip_package`` does + for the ``traversal`` layout), a ``..//...`` entry is + NOT blocked. Where the file physically lands depends on + Python's path-sanitisation policy; we just assert the guard + didn't refuse the entry.""" + import logging + + with tempfile.TemporaryDirectory() as base: + extract_dir = os.path.join(base, "Bundle.xcwhatever") + os.makedirs(extract_dir) + zip_path = self._make_malicious_zip( + base, + "bundle.zip", + { + "../Bundle.xcwhatever/Resources/Info.plist": b"", + }, + ) + with zipfile.ZipFile(zip_path) as zf, self.assertLogs( + drive_package_processing.LOGGER, + level=logging.WARNING, + ) as cm: + # Push a sentinel WARNING so assertLogs always has at + # least one record; the assertion below filters to + # zip-slip-blocked messages. + drive_package_processing.LOGGER.warning("sentinel") + drive_package_processing._safe_extractall( # noqa: SLF001 + zf, + extract_dir, + base, + ) + blocked = [m for m in cm.output if "Zip Slip blocked" in m] + self.assertEqual(blocked, []) + + def test_real_world_dotdot_bundle_blocked_when_boundary_too_tight(self): + """Conversely: with ``safety_boundary == extract_dir`` (the + bare-rooted layout), a ``..//...`` entry + actually escapes extract_dir and IS blocked. Proves the + boundary widening in the ``traversal`` layout is what makes + legit bundles work.""" + import logging + + with tempfile.TemporaryDirectory() as base: + # extract_dir's basename differs from the bundle name in + # the entry, so the ``..`` actually escapes upward into + # ``/ms.band/...`` rather than round-tripping back + # into extract_dir. + extract_dir = os.path.join(base, "Sample") + os.makedirs(extract_dir) + zip_path = self._make_malicious_zip( + base, + "bundle.zip", + {"../ms.band/Resources/Info.plist": b""}, + ) + with zipfile.ZipFile(zip_path) as zf, self.assertLogs( + drive_package_processing.LOGGER, + level=logging.WARNING, + ) as cm: + drive_package_processing._safe_extractall( # noqa: SLF001 + zf, + extract_dir, + extract_dir, + ) + blocked = [m for m in cm.output if "Zip Slip blocked" in m] + self.assertEqual(len(blocked), 1) + + +class TestProcessPackageNeverTouchesRegularUserZips(unittest.TestCase): + """Invariant: ``process_package`` is the unpacker for Apple + *package* downloads (``/packageDownload?`` URLs only). User-uploaded + zip files take the regular ``data_token`` download path and never + reach this function. This test class lives at the unit level to + lock the invariant in -- the URL gate is a string check in + ``drive_file_download.py``, but mistakes have happened (CWE-22 + upstream) and the cost of accidentally unpacking a user's .zip is + high (irrecoverable: the .zip file on iCloud is replaced by a + directory tree on disk, dedup breaks, restore is manual).""" + + def test_url_gate_is_the_only_caller(self): + """Audit: ``process_package`` has exactly one external caller + in the codebase, and that call site has the URL gate.""" + import re + from pathlib import Path as _Path + + callers = [] + for src_file in _Path("src").rglob("*.py"): + if src_file.name == "drive_package_processing.py": + continue # ignore the recursive self-call inside the module + text = src_file.read_text() + if "process_package(" in text: + callers.append(src_file) + + self.assertEqual( + len(callers), + 1, + f"process_package should have exactly 1 external caller " + f"(the URL-gated one in drive_file_download); found {len(callers)}: " + f"{[str(c) for c in callers]}", + ) + # And that caller is gated on the packageDownload URL substring. + caller_text = callers[0].read_text() + gate_pattern = re.compile( + r"if[^\n]*packageDownload\?[^\n]*in[^\n]*response\.url[\s\S]*?process_package\(", + ) + self.assertRegex( + caller_text, + gate_pattern, + "process_package call must be gated on a `/packageDownload?` " + "URL substring check so user-uploaded .zip files (which come " + "via data_token URLs without that segment) are never passed " + "to the unpacker.", + ) From 887ac4edb17654494f5b01e672127d761ae58aa4 Mon Sep 17 00:00:00 2001 From: Eric Pheterson Date: Wed, 3 Jun 2026 19:17:41 -0700 Subject: [PATCH 07/10] fix(drive): stop re-downloading flat package bundles every sync A package kept on disk as a flat single-file bundle (unrecognised-mime package, or flatten_packages=true) has an on-disk byte count equal to the package-download archive size, which never equals item.size (the package's logical size iCloud reports). file_exists() compares against item.size, sees a spurious size mismatch, and re-downloads the bundle on EVERY sync -- confirmed live on a real library (a 103 MB .key re-downloaded across consecutive syncs). Add package_bundle_unchanged(): when the item is a package and the on-disk bundle's mtime already matches the remote date_modified (which download_file stamps via os.utime, and iCloud bumps on any content change), the bytes are unchanged -- skip the re-download. Wired into both collect_file_for_download (active parallel path) and process_file (legacy). is_package() is already called in the outdated-file branch, so this adds no extra network round-trip. Completes the single-file-bundle handling from the package PR (#461 silenced the error log but left the re-download). 9 regression tests; 100% coverage. Co-Authored-By: Claude --- src/drive_file_existence.py | 29 +++++ src/drive_parallel_download.py | 12 +- src/sync_drive.py | 8 +- tests/test_drive_redownload_loop.py | 169 ++++++++++++++++++++++++++++ 4 files changed, 216 insertions(+), 2 deletions(-) create mode 100644 tests/test_drive_redownload_loop.py diff --git a/src/drive_file_existence.py b/src/drive_file_existence.py index c69a98897..f0cb85938 100644 --- a/src/drive_file_existence.py +++ b/src/drive_file_existence.py @@ -86,6 +86,35 @@ def package_exists(item: Any, local_package_path: str) -> bool: return False +def package_bundle_unchanged(item: Any, local_file: str) -> bool: + """Freshness check for a package stored as a flat single-file bundle. + + A flattened package -- an unrecognised-mime package kept as-is, or any + package downloaded with ``flatten_packages: true`` -- lands on disk as a + single file whose byte count is the size of the package-download archive, + NOT ``item.size`` (the package's *logical* size that iCloud reports). So + ``file_exists`` always sees a spurious size mismatch for these and + re-downloads the bundle on every sync. + + ``download_file`` stamps the bundle's mtime to ``item.date_modified`` via + ``os.utime``, and iCloud bumps ``date_modified`` whenever a package's + contents change, so the mtime is the reliable change signal. Match on mtime + alone (the same mtime comparison ``file_exists`` uses), skipping the + unusable size check. + + Args: + item: iCloud package item with a ``date_modified`` attribute + local_file: Path to the local single-file bundle + + Returns: + True if the bundle is present and its mtime matches the remote + (unchanged -- skip the re-download), False otherwise. + """ + if not (item and local_file and os.path.isfile(local_file)): + return False + return int(os.path.getmtime(local_file)) == int(item.date_modified.timestamp()) + + def is_package(item: Any, timeout: int = DEFAULT_REQUEST_TIMEOUT_SEC) -> bool: """Determine if an iCloud item is a package that needs special handling. diff --git a/src/drive_parallel_download.py b/src/drive_parallel_download.py index 69dab30d2..6d34713c5 100644 --- a/src/drive_parallel_download.py +++ b/src/drive_parallel_download.py @@ -16,7 +16,7 @@ from src import config_parser, configure_icloudpy_logging, get_logger from src.drive_file_download import download_file -from src.drive_file_existence import file_exists, is_package, package_exists +from src.drive_file_existence import file_exists, is_package, package_bundle_unchanged, package_exists from src.drive_filtering import wanted_file # Configure icloudpy logging immediately after import @@ -103,6 +103,16 @@ def collect_file_for_download( timeout = config_parser.get_drive_request_timeout(config) item_is_package = is_package(item=item, timeout=timeout) + # A package stored as a flat single-file bundle (unrecognised-mime package, + # or flatten_packages=true) reports item.size as the package's *logical* + # size, which never equals the on-disk bundle's byte count -- so file_exists() + # above sees a spurious size mismatch and we'd re-download the bundle on every + # sync. When it IS a package and the on-disk bundle's mtime already matches the + # remote, the bytes are unchanged; skip the wasteful re-download. is_package() + # was just called above, so this adds no extra network round-trip. + if item_is_package and package_bundle_unchanged(item=item, local_file=local_file): + return None + # Return download task info return { "item": item, diff --git a/src/sync_drive.py b/src/sync_drive.py index 3319a77fb..06ff2ef53 100644 --- a/src/sync_drive.py +++ b/src/sync_drive.py @@ -15,7 +15,7 @@ from src import config_parser, configure_icloudpy_logging, get_logger from src.drive_cleanup import remove_obsolete # noqa: F401 from src.drive_file_download import download_file # noqa: F401 -from src.drive_file_existence import file_exists, is_package, package_exists # noqa: F401 +from src.drive_file_existence import file_exists, is_package, package_bundle_unchanged, package_exists # noqa: F401 from src.drive_filtering import ignored_path, wanted_file, wanted_folder, wanted_parent_folder # noqa: F401 from src.drive_folder_processing import process_folder # noqa: F401 from src.drive_package_processing import process_package # noqa: F401 @@ -100,6 +100,12 @@ def process_file( # File exists locally but is outdated; need to determine type for re-download timeout = config_parser.get_drive_request_timeout(config) item_is_package = is_package(item=item, timeout=timeout) + # A flat single-file package bundle reports item.size as the package's + # logical size (never the on-disk byte count), so file_exists() above + # spuriously fails the size check and re-downloads every sync. If it's a + # package whose mtime already matches the remote, skip the re-download. + if item_is_package and package_bundle_unchanged(item=item, local_file=local_file): + return False elif os.path.isdir(local_file): # A directory at this path means the item was previously downloaded as a # package. iCloud Drive items do not change type between file and package, diff --git a/tests/test_drive_redownload_loop.py b/tests/test_drive_redownload_loop.py new file mode 100644 index 000000000..7d94d4f36 --- /dev/null +++ b/tests/test_drive_redownload_loop.py @@ -0,0 +1,169 @@ +"""Regression tests: flat package bundles must not re-download every sync. + +A package kept on disk as a flat single-file bundle (an unrecognised-mime +package, or any package downloaded with ``flatten_packages: true``) has an +on-disk byte count equal to the package-download archive size, which never +equals ``item.size`` (the package's *logical* size iCloud reports). Because +``file_exists`` compares against ``item.size`` it sees a spurious size +mismatch and the file re-downloads on every sync -- pure wasted bandwidth. + +``package_bundle_unchanged`` fixes this: when the item is a package and the +on-disk bundle's mtime already matches the remote ``date_modified`` (which +``download_file`` stamps via ``os.utime``), the bytes are unchanged and the +re-download is skipped. ``is_package`` is already called in the outdated-file +path, so the skip adds no extra network round-trip. +""" + +__author__ = "Mandar Patil (mandarons@pm.me)" + +import datetime +import os +import tempfile +import unittest +from unittest.mock import MagicMock, patch + +import tests # noqa: F401 — env setup +from src import drive_file_existence, sync_drive +from src.drive_parallel_download import collect_file_for_download + + +def _bundle_item(name="Investor Pitch.key", logical_size=4096, modified=None): + """A MagicMock iCloud item that behaves like a package: its ``size`` is the + package's logical size (deliberately NOT the on-disk bundle byte count).""" + item = MagicMock() + item.name = name + item.size = logical_size + item.date_modified = modified or datetime.datetime(2021, 3, 7, 12, 0, 0) + return item + + +def _write_bundle(path, item, on_disk_bytes=b"x" * 50): + """Write a flat bundle on disk with mtime stamped to item.date_modified + (as download_file does) and a byte count that differs from item.size.""" + with open(path, "wb") as f: + f.write(on_disk_bytes) + mtime = int(item.date_modified.timestamp()) + os.utime(path, (mtime, mtime)) + # Sanity: this is exactly the pathological condition — size differs. + assert os.path.getsize(path) != item.size + + +class TestPackageBundleUnchanged(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.mkdtemp() + + def tearDown(self): + import shutil + + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_true_when_mtime_matches(self): + item = _bundle_item() + path = os.path.join(self.tmp, "a.key") + _write_bundle(path, item) + self.assertTrue(drive_file_existence.package_bundle_unchanged(item=item, local_file=path)) + + def test_false_when_mtime_differs(self): + item = _bundle_item() + path = os.path.join(self.tmp, "a.key") + _write_bundle(path, item) + newer = int(item.date_modified.timestamp()) + 5 + os.utime(path, (newer, newer)) + self.assertFalse(drive_file_existence.package_bundle_unchanged(item=item, local_file=path)) + + def test_false_when_missing(self): + item = _bundle_item() + self.assertFalse( + drive_file_existence.package_bundle_unchanged( + item=item, + local_file=os.path.join(self.tmp, "nope.key"), + ), + ) + + def test_false_when_directory(self): + item = _bundle_item() + d = os.path.join(self.tmp, "pkg.key") + os.makedirs(d) + self.assertFalse(drive_file_existence.package_bundle_unchanged(item=item, local_file=d)) + + def test_false_when_no_item_or_path(self): + self.assertFalse(drive_file_existence.package_bundle_unchanged(item=None, local_file="x")) + self.assertFalse(drive_file_existence.package_bundle_unchanged(item=_bundle_item(), local_file=None)) + + +class TestCollectSkipsUnchangedBundle(unittest.TestCase): + """The active parallel-download path must skip an unchanged flat bundle.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp() + + def tearDown(self): + import shutil + + shutil.rmtree(self.tmp, ignore_errors=True) + + def _collect(self, item): + return collect_file_for_download( + item=item, + destination_path=self.tmp, + filters=None, + ignore=None, + files=set(), + config=None, + ) + + @patch("src.drive_parallel_download.is_package", return_value=True) + def test_unchanged_bundle_is_skipped(self, _mock_is_pkg): + item = _bundle_item() + _write_bundle(os.path.join(self.tmp, item.name), item) + # file_exists() fails on the size mismatch; the new guard must skip anyway. + self.assertIsNone(self._collect(item)) + + @patch("src.drive_parallel_download.is_package", return_value=True) + def test_changed_bundle_redownloads(self, _mock_is_pkg): + item = _bundle_item() + path = os.path.join(self.tmp, item.name) + _write_bundle(path, item) + newer = int(item.date_modified.timestamp()) + 99 + os.utime(path, (newer, newer)) # remote changed → mtime differs + info = self._collect(item) + self.assertIsNotNone(info) + self.assertTrue(info["is_package"]) + + @patch("src.drive_parallel_download.is_package", return_value=False) + def test_non_package_outdated_file_still_redownloads(self, _mock_is_pkg): + # A genuinely-changed regular file (not a package) must NOT be skipped. + item = _bundle_item(name="report.pdf") + _write_bundle(os.path.join(self.tmp, item.name), item) + info = self._collect(item) + self.assertIsNotNone(info) + self.assertFalse(info["is_package"]) + + +class TestProcessFileSkipsUnchangedBundle(unittest.TestCase): + """The legacy process_file path gets the same guard.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp() + + def tearDown(self): + import shutil + + shutil.rmtree(self.tmp, ignore_errors=True) + + @patch("src.sync_drive.is_package", return_value=True) + def test_unchanged_bundle_is_skipped(self, _mock_is_pkg): + item = _bundle_item() + _write_bundle(os.path.join(self.tmp, item.name), item) + result = sync_drive.process_file( + item=item, + destination_path=self.tmp, + filters=None, + ignore=None, + files=set(), + ) + self.assertFalse(result) # skipped → not processed/downloaded + + +if __name__ == "__main__": + unittest.main() From 32ef043d32353832f65614ea26ab5224793f9076 Mon Sep 17 00:00:00 2001 From: Eric Pheterson Date: Wed, 3 Jun 2026 19:42:18 -0700 Subject: [PATCH 08/10] review: stamp mtime via .timestamp() to match the freshness-check read side Code review flagged a latent timezone bug: download_file stamped mtime via time.mktime(item.date_modified.timetuple()) (misreads a tz-aware datetime as local time), while file_exists()/package_bundle_unchanged() read via .timestamp() (true epoch). They agree for the naive datetimes iCloudPy returns today (numerically identical -> existing files unaffected), but would desync for tz-aware datetimes and silently break dedup. Switch the write to .timestamp() so write and read use the same clock; drop the now-unused 'import time'. Add a tz-aware regression test. Co-Authored-By: Claude --- src/drive_file_download.py | 9 +++++++-- tests/test_drive_redownload_loop.py | 12 ++++++++++++ 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/src/drive_file_download.py b/src/drive_file_download.py index 1656f5e15..5a5af7e7a 100644 --- a/src/drive_file_download.py +++ b/src/drive_file_download.py @@ -7,7 +7,6 @@ __author__ = "Mandar Patil (mandarons@pm.me)" import os -import time from typing import Any from src import configure_icloudpy_logging, get_logger @@ -69,7 +68,13 @@ def download_file( local_file = processed_file # Set the file modification time to match the remote file - item_modified_time = time.mktime(item.date_modified.timetuple()) + # Use .timestamp() (true epoch) to match the read side in + # file_exists() / package_bundle_unchanged(). time.mktime(timetuple()) + # misreads a timezone-aware date_modified as local time, desyncing the + # stamped mtime from the freshness check. For the naive datetimes + # iCloudPy returns today the two are numerically identical, so existing + # on-disk files are unaffected. + item_modified_time = item.date_modified.timestamp() os.utime(local_file, (item_modified_time, item_modified_time)) except Exception as e: diff --git a/tests/test_drive_redownload_loop.py b/tests/test_drive_redownload_loop.py index 7d94d4f36..6f1b95824 100644 --- a/tests/test_drive_redownload_loop.py +++ b/tests/test_drive_redownload_loop.py @@ -90,6 +90,18 @@ def test_false_when_no_item_or_path(self): self.assertFalse(drive_file_existence.package_bundle_unchanged(item=None, local_file="x")) self.assertFalse(drive_file_existence.package_bundle_unchanged(item=_bundle_item(), local_file=None)) + def test_true_for_timezone_aware_item(self): + # download_file stamps mtime via .timestamp() and the check reads via + # .timestamp(), so a timezone-aware date_modified round-trips correctly. + # Regression guard for the old time.mktime(timetuple()) write that + # misread tz-aware datetimes as local time and desynced the mtime. + item = _bundle_item( + modified=datetime.datetime(2021, 3, 7, 12, 0, 0, tzinfo=datetime.timezone.utc), + ) + path = os.path.join(self.tmp, "tz.key") + _write_bundle(path, item) + self.assertTrue(drive_file_existence.package_bundle_unchanged(item=item, local_file=path)) + class TestCollectSkipsUnchangedBundle(unittest.TestCase): """The active parallel-download path must skip an unchanged flat bundle.""" From 4c1a43b65fdb2437375d302ac9b4e79d2ea0415f Mon Sep 17 00:00:00 2001 From: Eric Pheterson Date: Fri, 5 Jun 2026 17:29:28 -0700 Subject: [PATCH 09/10] Revert "review: stamp mtime via .timestamp() to match the freshness-check read side" This reverts commit 32ef043d32353832f65614ea26ab5224793f9076. --- src/drive_file_download.py | 9 ++------- tests/test_drive_redownload_loop.py | 12 ------------ 2 files changed, 2 insertions(+), 19 deletions(-) diff --git a/src/drive_file_download.py b/src/drive_file_download.py index 5a5af7e7a..1656f5e15 100644 --- a/src/drive_file_download.py +++ b/src/drive_file_download.py @@ -7,6 +7,7 @@ __author__ = "Mandar Patil (mandarons@pm.me)" import os +import time from typing import Any from src import configure_icloudpy_logging, get_logger @@ -68,13 +69,7 @@ def download_file( local_file = processed_file # Set the file modification time to match the remote file - # Use .timestamp() (true epoch) to match the read side in - # file_exists() / package_bundle_unchanged(). time.mktime(timetuple()) - # misreads a timezone-aware date_modified as local time, desyncing the - # stamped mtime from the freshness check. For the naive datetimes - # iCloudPy returns today the two are numerically identical, so existing - # on-disk files are unaffected. - item_modified_time = item.date_modified.timestamp() + item_modified_time = time.mktime(item.date_modified.timetuple()) os.utime(local_file, (item_modified_time, item_modified_time)) except Exception as e: diff --git a/tests/test_drive_redownload_loop.py b/tests/test_drive_redownload_loop.py index 6f1b95824..7d94d4f36 100644 --- a/tests/test_drive_redownload_loop.py +++ b/tests/test_drive_redownload_loop.py @@ -90,18 +90,6 @@ def test_false_when_no_item_or_path(self): self.assertFalse(drive_file_existence.package_bundle_unchanged(item=None, local_file="x")) self.assertFalse(drive_file_existence.package_bundle_unchanged(item=_bundle_item(), local_file=None)) - def test_true_for_timezone_aware_item(self): - # download_file stamps mtime via .timestamp() and the check reads via - # .timestamp(), so a timezone-aware date_modified round-trips correctly. - # Regression guard for the old time.mktime(timetuple()) write that - # misread tz-aware datetimes as local time and desynced the mtime. - item = _bundle_item( - modified=datetime.datetime(2021, 3, 7, 12, 0, 0, tzinfo=datetime.timezone.utc), - ) - path = os.path.join(self.tmp, "tz.key") - _write_bundle(path, item) - self.assertTrue(drive_file_existence.package_bundle_unchanged(item=item, local_file=path)) - class TestCollectSkipsUnchangedBundle(unittest.TestCase): """The active parallel-download path must skip an unchanged flat bundle.""" From 61af5bb953b12b27812914c32ada4487c8ec0f42 Mon Sep 17 00:00:00 2001 From: Eric Pheterson Date: Sat, 6 Jun 2026 20:42:45 -0700 Subject: [PATCH 10/10] align package_bundle_unchanged to the merged #451 UTC mtime convention MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #451 (now merged to main) made file_exists/package_exists read mtimes as naive-UTC via .replace(tzinfo=timezone.utc).timestamp(). package_bundle_unchanged was the lone naive holdout — switch it to the same convention so flat-bundle dedup is TZ-invariant too (otherwise it'd break under a non-UTC TZ, the exact bug #451 fixes). Test stamps mtime the same way. Co-Authored-By: Claude --- src/drive_file_existence.py | 4 +++- tests/test_drive_redownload_loop.py | 2 +- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/src/drive_file_existence.py b/src/drive_file_existence.py index 58ef1ff92..0ad2a6d74 100644 --- a/src/drive_file_existence.py +++ b/src/drive_file_existence.py @@ -117,7 +117,9 @@ def package_bundle_unchanged(item: Any, local_file: str) -> bool: """ if not (item and local_file and os.path.isfile(local_file)): return False - return int(os.path.getmtime(local_file)) == int(item.date_modified.timestamp()) + # iCloudPy produces date_modified as naive UTC (strptime ...Z); replace(tzinfo=UTC) + # matches the convention file_exists/package_exists use so mtimes are TZ-invariant. + return int(os.path.getmtime(local_file)) == int(item.date_modified.replace(tzinfo=timezone.utc).timestamp()) def is_package(item: Any, timeout: int = DEFAULT_REQUEST_TIMEOUT_SEC) -> bool: diff --git a/tests/test_drive_redownload_loop.py b/tests/test_drive_redownload_loop.py index 7d94d4f36..af6a0bb38 100644 --- a/tests/test_drive_redownload_loop.py +++ b/tests/test_drive_redownload_loop.py @@ -42,7 +42,7 @@ def _write_bundle(path, item, on_disk_bytes=b"x" * 50): (as download_file does) and a byte count that differs from item.size.""" with open(path, "wb") as f: f.write(on_disk_bytes) - mtime = int(item.date_modified.timestamp()) + mtime = int(item.date_modified.replace(tzinfo=datetime.timezone.utc).timestamp()) os.utime(path, (mtime, mtime)) # Sanity: this is exactly the pathological condition — size differs. assert os.path.getsize(path) != item.size