From ea7164d3d9c41028dcf47fa778d73df085ab8bdc Mon Sep 17 00:00:00 2001 From: scarreras Date: Mon, 2 Nov 2026 11:30:22 -0300 Subject: [PATCH] fix: seven confirmed bugs from a whole-branch review, each with a test that fails without the fix Fixes found by reviewing the default branch, with covering unit tests (fail before, pass after) and strengthened weak tests: - SQLite account info: bound parameter for allowed capabilities; reset upload pools after schema update - Bucket.copy: pass expires and content headers through - start_large_file: send customUploadTimestamp under its wire name - Replication monitor: no hang when the consumer fails - Local folder scan: keep scanning after an unreadable file when no reporter is given - Synchronizer: accept reporter=None - Parallel downloader: hash written data; surface writer errors One known bug whose behavior needs an owner decision (executor not shut down when scheduling fails) is recorded as an xfail with its reason. Also adds a sdkharness sync leg for reporter=None. Co-Authored-By: Claude Sonnet 5.5 --- .sdkharness/tests/conformance/client.sync | 71 ++++++++++++++- .../account_info/sqlite_account_info.py | 11 ++- b2sdk/_internal/bucket.py | 4 + b2sdk/_internal/raw_api.py | 3 +- b2sdk/_internal/replication/monitoring.py | 43 ++++++--- b2sdk/_internal/scan/folder.py | 4 +- b2sdk/_internal/sync/sync.py | 7 +- .../transfer/inbound/downloader/parallel.py | 36 ++++++-- changelog.d/+copy-file-headers.fixed.md | 1 + changelog.d/+custom-upload-timestamp.fixed.md | 1 + changelog.d/+local-scan-errors.fixed.md | 1 + .../+parallel-download-writer.fixed.md | 1 + .../+replication-monitor-hang.fixed.md | 1 + changelog.d/+sqlite-allowed-param.fixed.md | 1 + changelog.d/+sync-silent-reporter.fixed.md | 1 + test/unit/account_info/test_account_info.py | 12 +++ .../account_info/test_sqlite_account_info.py | 39 ++++++++ test/unit/b2http/test_b2http.py | 18 +++- test/unit/bucket/test_bucket.py | 45 +++++++++- .../transfer/downloader/test_parallel.py | 63 +++++++++++++ test/unit/replication/test_monitoring.py | 90 +++++++++++++++++++ test/unit/scan/test_folder_traversal.py | 37 ++++++++ test/unit/sync/test_sync.py | 65 ++++++++++++-- test/unit/sync/test_sync_report.py | 29 ++++++ test/unit/test_raw_api.py | 35 ++++++++ ...harness_conformance_resilience_contract.py | 21 +++++ test/unit/v0/test_bucket.py | 12 ++- test/unit/v0/test_sync.py | 8 +- test/unit/v1/test_bounded_queue_executor.py | 28 ++++++ test/unit/v1/test_raw_api.py | 26 ++++-- test/unit/v1/test_sync.py | 8 +- 31 files changed, 671 insertions(+), 51 deletions(-) create mode 100644 changelog.d/+copy-file-headers.fixed.md create mode 100644 changelog.d/+custom-upload-timestamp.fixed.md create mode 100644 changelog.d/+local-scan-errors.fixed.md create mode 100644 changelog.d/+parallel-download-writer.fixed.md create mode 100644 changelog.d/+replication-monitor-hang.fixed.md create mode 100644 changelog.d/+sqlite-allowed-param.fixed.md create mode 100644 changelog.d/+sync-silent-reporter.fixed.md diff --git a/.sdkharness/tests/conformance/client.sync b/.sdkharness/tests/conformance/client.sync index 8032ff5ea..2ff8438c5 100755 --- a/.sdkharness/tests/conformance/client.sync +++ b/.sdkharness/tests/conformance/client.sync @@ -9,6 +9,10 @@ claims it): policy; sync down into a fresh directory Assert all 3 land; the remote matches the local side after the second pass; the round-tripped bytes are identical + Leg reporter=None (sync_folders documents `SyncReport | None`): sync + down from the bucket, and sync up from a directory holding a + dangling symlink; neither may crash, and the unreadable entry is + skipped, not uploaded NO DOCUMENTED CONTRACT FOUND -- this is SDK-side/ecosystem tooling, not a B2 API contract. B2 documents only sync's building blocks. "Remote matches local" is read @@ -257,7 +261,16 @@ def write(path, data): handle.write(data) -def sync(b2, syncer, source, destination): +def sync(b2, syncer, source, destination, silent=False): + """Run one sync pass. `silent` passes reporter=None, which sync_folders documents as valid.""" + if silent: + syncer.sync_folders( + source_folder=source, + dest_folder=destination, + now_millis=int(time.time() * 1000), + reporter=None, + ) + return reporter = b2.SyncReport(io.StringIO(), True) try: syncer.sync_folders( @@ -353,6 +366,62 @@ def run(target: str) -> None: extra = sorted(set(os.listdir(down_dir)) - set(contents)) if extra: raise Failure('sync down', f'{len(extra)} unexpected file(s) came down') + + # Leg: reporter=None. sync_folders documents `SyncReport | None`, so a caller + # that wants no progress output must get a normal sync, never an AttributeError + # (b2-source path) and never a crash on an inaccessible local file. + silent_down = tempfile.mkdtemp(prefix='sdkharness-conf-silent-down-') + silent_up = tempfile.mkdtemp(prefix='sdkharness-conf-silent-up-') + try: + step( + 'sync down with reporter=None', + lambda: sync( + b2, + syncer, + b2.parse_folder(remote_url, api), + b2.parse_folder(silent_down, api), + silent=True, + ), + ) + if sorted(os.listdir(silent_down)) != sorted(contents): + raise Failure('sync down with reporter=None', 'the files did not all come down') + for name, expected in contents.items(): + with open(os.path.join(silent_down, name), 'rb') as handle: + if handle.read() != expected: + raise Failure('sync down with reporter=None', "a file's bytes differ") + + # A local source holding a file that cannot be read. A dangling symlink is + # inaccessible whoever runs the check (chmod 000 is not for root). + write(os.path.join(silent_up, 'good.txt'), payload_of(300)) + os.symlink( + os.path.join(silent_up, 'does-not-exist'), os.path.join(silent_up, 'a-dangling') + ) + silent_url = f'b2://{bucket.name}/{REMOTE_PREFIX}-silent' + step( + 'sync up with reporter=None past an inaccessible file', + lambda: sync( + b2, + syncer, + b2.parse_folder(silent_up, api), + b2.parse_folder(silent_url, api), + silent=True, + ), + ) + silent_landed = step( + 'list the reporter=None upload', + lambda: sorted( + v.file_name.rsplit('/', 1)[-1] + for v, _folder in bucket.ls(REMOTE_PREFIX + '-silent/', recursive=True) + ), + ) + if silent_landed != ['good.txt']: + raise Failure( + 'sync up with reporter=None past an inaccessible file', + f'the remote holds {len(silent_landed)} file(s), expected only the readable one', + ) + finally: + shutil.rmtree(silent_down, ignore_errors=True) + shutil.rmtree(silent_up, ignore_errors=True) finally: teardown(api, bucket) shutil.rmtree(up_dir, ignore_errors=True) diff --git a/b2sdk/_internal/account_info/sqlite_account_info.py b/b2sdk/_internal/account_info/sqlite_account_info.py index 76e865b2a..1ee7e5a6f 100644 --- a/b2sdk/_internal/account_info/sqlite_account_info.py +++ b/b2sdk/_internal/account_info/sqlite_account_info.py @@ -441,9 +441,8 @@ def _migrate_allowed_to_multi_bucket(self, *, version: int): allowed['buckets'] = None allowed_text = json.dumps(allowed) - stmt = f"UPDATE account SET allowed = ('{allowed_text}');" - self._perform_update(version, [stmt]) + self._perform_update(version, [('UPDATE account SET allowed = ?;', (allowed_text,))]) def _ensure_update(self, update_number, update_commands: list[str]): """ @@ -466,17 +465,21 @@ def _get_update_count(self, update_number: int): ) return cursor.fetchone()[0] - def _perform_update(self, update_number, update_commands: list[str]): + def _perform_update(self, update_number, update_commands: list[str | tuple[str, tuple]]): with self._get_connection() as conn: conn.execute('BEGIN') for command in update_commands: - conn.execute(command) + if isinstance(command, str): + conn.execute(command) + else: + conn.execute(*command) conn.execute('INSERT INTO update_done (update_number) VALUES (?);', (update_number,)) def clear(self): """ Remove all info about accounts and buckets. """ + self._reset_upload_pools() with self._get_connection() as conn: conn.execute('DELETE FROM account;') conn.execute('DELETE FROM bucket;') diff --git a/b2sdk/_internal/bucket.py b/b2sdk/_internal/bucket.py index e6df3495c..e9bb5c4bd 100644 --- a/b2sdk/_internal/bucket.py +++ b/b2sdk/_internal/bucket.py @@ -1485,6 +1485,10 @@ def copy( file_retention=file_retention, legal_hold=legal_hold, cache_control=cache_control, + expires=expires, + content_disposition=content_disposition, + content_encoding=content_encoding, + content_language=content_language, min_part_size=min_part_size, max_part_size=max_part_size, ) diff --git a/b2sdk/_internal/raw_api.py b/b2sdk/_internal/raw_api.py index b24e5ba31..0a392bf1e 100644 --- a/b2sdk/_internal/raw_api.py +++ b/b2sdk/_internal/raw_api.py @@ -392,6 +392,7 @@ def start_large_file( server_side_encryption: EncryptionSetting | None = None, file_retention: FileRetentionSetting | None = None, legal_hold: LegalHold | None = None, + custom_upload_timestamp: int | None = None, ): pass @@ -880,7 +881,7 @@ def start_large_file( kwargs['fileRetention'] = file_retention.serialize_to_json_for_request() if custom_upload_timestamp is not None: - kwargs['custom_upload_timestamp'] = custom_upload_timestamp + kwargs['customUploadTimestamp'] = custom_upload_timestamp return self._post_json( api_url, diff --git a/b2sdk/_internal/replication/monitoring.py b/b2sdk/_internal/replication/monitoring.py index bae3a2a83..2c3cc983d 100644 --- a/b2sdk/_internal/replication/monitoring.py +++ b/b2sdk/_internal/replication/monitoring.py @@ -12,7 +12,8 @@ import sys from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field -from queue import Queue +from queue import Full, Queue +from threading import Event from typing import ClassVar, Iterator from ..api import B2Api @@ -195,29 +196,49 @@ def scan(self, scan_destination: bool = True) -> AbstractScanReport: """ report = self.REPORT_CLASS() queue = Queue(maxsize=self.QUEUE_SIZE) + stop = Event() # set when the consumer dies, so the producer never blocks on a full queue + + def put(item) -> bool: + while not stop.is_set(): + try: + queue.put(item, timeout=0.1) + return True + except Full: + pass + return False if not scan_destination: def fill_queue(): - for path in self.source_folder.all_files( - policies_manager=self.scan_policies_manager, - reporter=self.report, - ): - queue.put((path,), block=True) - queue.put(None, block=True) + try: + for path in self.source_folder.all_files( + policies_manager=self.scan_policies_manager, + reporter=self.report, + ): + if not put((path,)): + return + finally: + put(None) else: def fill_queue(): - for pair in self.iter_pairs(): - queue.put(pair, block=True) - queue.put(None, block=True) + try: + for pair in self.iter_pairs(): + if not put(pair): + return + finally: + put(None) def consume_queue(): while True: items = queue.get(block=True) if items is None: # using None as "end of queue" marker break - report.add(*items) + try: + report.add(*items) + except BaseException: + stop.set() + raise with ThreadPoolExecutor(max_workers=2) as thread_pool: futures = [ diff --git a/b2sdk/_internal/scan/folder.py b/b2sdk/_internal/scan/folder.py index 254a5f55b..66553ea2a 100644 --- a/b2sdk/_internal/scan/folder.py +++ b/b2sdk/_internal/scan/folder.py @@ -289,7 +289,7 @@ def _walk_relative_paths( except OSError: if reporter is not None: reporter.local_access_error(str(local_path)) - continue + continue local_scan_path = LocalPath( absolute_path=self.make_full_path(str(relative_file_path)), @@ -303,7 +303,7 @@ def _walk_relative_paths( if not _file_read_access(local_path): if reporter is not None: reporter.local_permission_error(str(local_path)) - continue + continue yield local_scan_path diff --git a/b2sdk/_internal/sync/sync.py b/b2sdk/_internal/sync/sync.py index 2ce4f72d5..252f40dd7 100644 --- a/b2sdk/_internal/sync/sync.py +++ b/b2sdk/_internal/sync/sync.py @@ -10,6 +10,7 @@ from __future__ import annotations import concurrent.futures as futures +import io import logging from enum import Enum, unique from typing import cast @@ -202,7 +203,11 @@ def sync_folders( queue_limit = self.max_workers + 1000 sync_executor = BoundedQueueExecutor(unbounded_executor, queue_limit=queue_limit) - if source_type == 'local' and reporter is not None: + if reporter is None: + # silent reporter so that downstream code can call it unconditionally + reporter = SyncReport(io.StringIO(), True) + + if source_type == 'local': # Start the thread that counts the local files. That's the operation # that should be fastest, and it provides scale for the progress reporting. sync_executor.submit(count_files, source_folder, reporter, self.policies_manager) diff --git a/b2sdk/_internal/transfer/inbound/downloader/parallel.py b/b2sdk/_internal/transfer/inbound/downloader/parallel.py index a3ae8c0f5..ab185130f 100644 --- a/b2sdk/_internal/transfer/inbound/downloader/parallel.py +++ b/b2sdk/_internal/transfer/inbound/downloader/parallel.py @@ -156,7 +156,7 @@ def _finish_hashing(self, first_part, file, hasher, content_length): stop = True else: to_hash = data - hasher.update(data) + hasher.update(to_hash) current_offset += len(to_hash) if stop: break @@ -248,6 +248,8 @@ def __init__(self, file, max_queue_depth): self.file = file self.queue = queue.Queue(max_queue_depth) self.total = 0 + self.error: BaseException | None = None + self._stopped = threading.Event() self.stats_collector = StatsCollector(str(self.file), 'writer', 'seek') super().__init__() @@ -266,11 +268,18 @@ def run(self): if shutdown: break - with stats_collector_other: - file.seek(offset) + if self.error is not None: + continue # keep draining so that producers never block - with stats_collector_write: - file.write(data) + try: + with stats_collector_other: + file.seek(offset) + + with stats_collector_write: + file.write(data) + except Exception as e: + self.error = e + continue self.total += len(data) @@ -279,12 +288,25 @@ def __enter__(self): return self def queue_write(self, offset: int, data: bytes) -> None: - self.queue.put((False, offset, data)) + item = (False, offset, data) + while True: + if self.error is not None: + raise self.error + if self._stopped.is_set(): + raise RuntimeError('writer has been stopped') + try: + self.queue.put(item, timeout=0.1) + return + except queue.Full: + pass def __exit__(self, exc_type, exc_val, exc_tb): - self.queue.put((True, None, None)) + self._stopped.set() # make straggling producers fail instead of blocking forever + self.queue.put((True, None, None)) # writer keeps draining, so this cannot block for long self.join() self.stats_collector.report() + if exc_type is None and self.error is not None: + raise self.error def download_first_part( diff --git a/changelog.d/+copy-file-headers.fixed.md b/changelog.d/+copy-file-headers.fixed.md new file mode 100644 index 000000000..e281ca1b4 --- /dev/null +++ b/changelog.d/+copy-file-headers.fixed.md @@ -0,0 +1 @@ +Pass `expires`, `content_disposition`, `content_encoding` and `content_language` through `Bucket.copy` to the created file instead of dropping them. diff --git a/changelog.d/+custom-upload-timestamp.fixed.md b/changelog.d/+custom-upload-timestamp.fixed.md new file mode 100644 index 000000000..23a2d9ecc --- /dev/null +++ b/changelog.d/+custom-upload-timestamp.fixed.md @@ -0,0 +1 @@ +Send `customUploadTimestamp` under its B2 wire name when starting a large file, so custom upload timestamps are no longer ignored. diff --git a/changelog.d/+local-scan-errors.fixed.md b/changelog.d/+local-scan-errors.fixed.md new file mode 100644 index 000000000..000363130 --- /dev/null +++ b/changelog.d/+local-scan-errors.fixed.md @@ -0,0 +1 @@ +Keep scanning a local folder after a file cannot be read or accessed when no reporter is given, instead of failing on that file. diff --git a/changelog.d/+parallel-download-writer.fixed.md b/changelog.d/+parallel-download-writer.fixed.md new file mode 100644 index 000000000..60d86aa3d --- /dev/null +++ b/changelog.d/+parallel-download-writer.fixed.md @@ -0,0 +1 @@ +Parallel download: hash the data that was actually written, and surface a failing writer to producers instead of blocking them or losing the error. diff --git a/changelog.d/+replication-monitor-hang.fixed.md b/changelog.d/+replication-monitor-hang.fixed.md new file mode 100644 index 000000000..c70a5291f --- /dev/null +++ b/changelog.d/+replication-monitor-hang.fixed.md @@ -0,0 +1 @@ +Stop the replication monitor from hanging when its consumer fails: the producer no longer blocks on a full queue and always signals the end of input. diff --git a/changelog.d/+sqlite-allowed-param.fixed.md b/changelog.d/+sqlite-allowed-param.fixed.md new file mode 100644 index 000000000..d9771ce30 --- /dev/null +++ b/changelog.d/+sqlite-allowed-param.fixed.md @@ -0,0 +1 @@ +Store the account's allowed capabilities in SQLite with a bound parameter instead of string formatting, and reset upload pools after an account-info schema update. diff --git a/changelog.d/+sync-silent-reporter.fixed.md b/changelog.d/+sync-silent-reporter.fixed.md new file mode 100644 index 000000000..0df357bd2 --- /dev/null +++ b/changelog.d/+sync-silent-reporter.fixed.md @@ -0,0 +1 @@ +Accept `reporter=None` in `Synchronizer.sync_folders` by using a silent reporter instead of failing. diff --git a/test/unit/account_info/test_account_info.py b/test/unit/account_info/test_account_info.py index 8c40ec27e..e1969569b 100644 --- a/test/unit/account_info/test_account_info.py +++ b/test/unit/account_info/test_account_info.py @@ -217,6 +217,18 @@ def test_set_auth_data_compatibility(self, account_info_default_data, allowed): ) assert account_info.get_allowed() == allowed + def test_clear_drops_cached_upload_urls(self, account_info_default_data): + """After ``clear()`` (log out) no upload URL or token of the cleared account may be handed out.""" + account_info = self._make_info() + account_info.set_auth_data(**account_info_default_data) + account_info.put_bucket_upload_url('bucket-0', 'http://bucket-0', 'bucket-0_auth') + account_info.put_large_file_upload_url('file_0', 'http://file_0', 'auth_0') + + account_info.clear() + + assert (None, None) == account_info.take_bucket_upload_url('bucket-0') + assert (None, None) == account_info.take_large_file_upload_url('file_0') + def test_clear_bucket_upload_data(self): account_info = self._make_info() account_info.put_bucket_upload_url('bucket-0', 'http://bucket-0', 'bucket-0_auth') diff --git a/test/unit/account_info/test_sqlite_account_info.py b/test/unit/account_info/test_sqlite_account_info.py index a025a3d5d..afb2231f6 100644 --- a/test/unit/account_info/test_sqlite_account_info.py +++ b/test/unit/account_info/test_sqlite_account_info.py @@ -194,6 +194,45 @@ def test_migrate_to_5_v3( assert new_account_info.get_allowed() == exp_allowed assert new_account_info._get_update_count(end_version) == 1 + @pytest.mark.parametrize( + ('start_version', 'end_version'), + [ + (4, 5), + (5, 6), + ], + ) + @pytest.mark.apiver(from_ver=3) + def test_migrate_to_5_v3_name_prefix_with_apostrophe( + self, + sqlite_account_info_factory, + account_info_default_data: dict, + start_version: int, + end_version: int, + ): + """A single quote in the stored ``allowed`` JSON must not break the migration SQL.""" + old_allowed = dict( + bucketId='123', + bucketName='bucket1', + capabilities=['listBuckets', 'readBuckets'], + namePrefix="it's/", + ) + old_account_info = sqlite_account_info_factory(last_upgrade_to_run=start_version) + old_account_info.set_auth_data(**account_info_default_data) + + with old_account_info._get_connection() as conn: + conn.execute('UPDATE account SET allowed = ?;', (json.dumps(old_allowed),)) + + new_account_info = self.sqlite_account_info_factory( + file_name=old_account_info.filename, last_upgrade_to_run=end_version + ) + + assert new_account_info.get_allowed() == dict( + buckets=[{'id': '123', 'name': 'bucket1'}], + capabilities=['listBuckets', 'readBuckets'], + namePrefix="it's/", + ) + assert new_account_info._get_update_count(end_version) == 1 + @pytest.mark.apiver(2) def test_multi_bucket_key_error_apiver_v2( self, diff --git a/test/unit/b2http/test_b2http.py b/test/unit/b2http/test_b2http.py index 6ae18269e..138464793 100644 --- a/test/unit/b2http/test_b2http.py +++ b/test/unit/b2http/test_b2http.py @@ -530,10 +530,20 @@ def test_set_locale_context_manager(self): if saved == test_locale: test_locale, other_locale = other_locale, test_locale - locale.setlocale(locale.LC_ALL, other_locale) - with setlocale(test_locale): - assert locale.setlocale(category=locale.LC_ALL) == test_locale - locale.setlocale(locale.LC_ALL, saved) + try: + locale.setlocale(locale.LC_ALL, other_locale) + with setlocale(test_locale): + assert locale.setlocale(category=locale.LC_ALL) == test_locale + # leaving the context manager must put the previous locale back... + assert locale.setlocale(category=locale.LC_ALL) == other_locale + + # ...also when the body raises + with pytest.raises(RuntimeError): + with setlocale(test_locale): + raise RuntimeError('boom') + assert locale.setlocale(category=locale.LC_ALL) == other_locale + finally: + locale.setlocale(locale.LC_ALL, saved) class TestClockSkewHook(TestBase): diff --git a/test/unit/bucket/test_bucket.py b/test/unit/bucket/test_bucket.py index f140a6612..86ff96fc1 100644 --- a/test/unit/bucket/test_bucket.py +++ b/test/unit/bucket/test_bucket.py @@ -32,6 +32,7 @@ B2Error, B2RequestTimeoutDuringUpload, BucketIdNotFound, + Conflict, DestinationDirectoryDoesntAllowOperation, DestinationDirectoryDoesntExist, DestinationIsADirectory, @@ -1450,6 +1451,7 @@ def test_copy_retention(self): resulting_file_version = self.bucket.copy( file_id, 'copied_file', + length=length, file_retention=FileRetentionSetting(RetentionMode.COMPLIANCE, 100), legal_hold=LegalHold.ON, max_part_size=400, @@ -1460,6 +1462,32 @@ def test_copy_retention(self): ) self.assertEqual(LegalHold.ON, resulting_file_version.legal_hold) + @pytest.mark.apiver(from_ver=2) + def test_copy_keeps_http_header_params_with_and_without_length(self): + """ + ``expires``, ``content_disposition``, ``content_encoding`` and ``content_language`` + must end up in the new file's file_info whether the copy is a plain ``b2_copy_file`` + (no ``length``) or goes through the large file interface (``length`` given). + """ + for length in [None, 5]: + with self.subTest(length=length): + file_id = self._make_file() + copied = self.bucket.copy( + file_id, + f'copied_{length}.txt', + offset=0, + length=length, + content_type='text/plain', + content_disposition='attachment', + expires='Thu, 01 Dec 2050 16:00:00 GMT', + content_language='en', + content_encoding='identity', + ) + self.assertEqual('attachment', copied.file_info['b2-content-disposition']) + self.assertEqual('Thu, 01 Dec 2050 16:00:00 GMT', copied.file_info['b2-expires']) + self.assertEqual('en', copied.file_info['b2-content-language']) + self.assertEqual('identity', copied.file_info['b2-content-encoding']) + def test_copy_encryption(self): data = b'hello_world' a = self.bucket.upload_bytes(data, 'a') @@ -1829,13 +1857,12 @@ def test_update_if_revision_is(self): updated_bucket = self.api.get_bucket_by_name(self.bucket.name) self.assertEqual([{'fileNamePrefix': 'is_life'}], updated_bucket.lifecycle_rules) - try: + # a stale revision must be rejected, and the rejected update must not be applied + with self.assertRaises(Conflict): self.bucket.update( - lifecycle_rules=[{'fileNamePrefix': 'is_life'}], + lifecycle_rules=[{'fileNamePrefix': 'stale_update'}], if_revision_is=current_revision, # this is now the old revision ) - except Exception: - pass not_updated_bucket = self.api.get_bucket_by_name(self.bucket.name) self.assertEqual([{'fileNamePrefix': 'is_life'}], not_updated_bucket.lifecycle_rules) @@ -3018,6 +3045,16 @@ def setUp(self): def test_download_by_id_progress_monotonic(self): self.download_file_by_id(self.file_version.id_, progress_listener=self.progress_listener) self._verify(self.DATA) + # _verify does not require incremental reports; a single end-of-download report + # would pass it, so require several, non-decreasing ones here + valid, reason = self.progress_listener.is_valid_reason( + check_progress=True, + check_monotonic_progress=True, + ) + assert valid, reason + reports = [int(v) for v in self.progress_listener.history[1:] if v != 'closed'] + assert len(reports) > 1 + assert reports == sorted(reports) # Truncated downloads diff --git a/test/unit/internal/transfer/downloader/test_parallel.py b/test/unit/internal/transfer/downloader/test_parallel.py index cac4ebf28..eda3d4be6 100644 --- a/test/unit/internal/transfer/downloader/test_parallel.py +++ b/test/unit/internal/transfer/downloader/test_parallel.py @@ -8,6 +8,7 @@ # ###################################################################### import hashlib +import threading from concurrent.futures import ThreadPoolExecutor from io import BytesIO from unittest.mock import Mock @@ -86,6 +87,27 @@ def test_download_file(apiver_module, b2api, bucket, downloader, output_file): mock_response.close.assert_called_once() +def test_download_file__does_not_hash_bytes_after_the_downloaded_range( + apiver_module, b2api, bucket, downloader +): + """ + The returned hash is the hash of the downloaded content, even when the output file already + holds more data after the end of the downloaded range. + """ + file_size = 100 + mock_response, download_version = mock_download_response_factory( + apiver_module, bucket, file_size=file_size + ) + output_file = BytesIO(b'trailing data that is not part of the download' * 10) + + bytes_written, hash_hex = downloader.download( + output_file, mock_response, download_version, b2api.session + ) + + assert bytes_written == file_size + assert hash_hex == '7804df8c623573ccfc1993e04981006e5bc30383' # sha1 of b'dummy' * 20 + + def test_download_file__data_stream_error__in_first_response( apiver_module, b2api, bucket, downloader, output_file ): @@ -169,3 +191,44 @@ def iter_content(chunk_size=1, decode_unicode=False): assert bytes_written == file_size assert output_file.getvalue() == b'dUMMY' + b'DUMMY' * 19 + + +class FileFailingOnSecondWrite(BytesIO): + """A file-like object that behaves like a full disk after the first write.""" + + def __init__(self): + super().__init__() + self.write_calls = 0 + + def write(self, data): + self.write_calls += 1 + if self.write_calls > 1: + raise OSError('disk full') + return super().write(data) + + +def test_writer_thread__write_error__does_not_block_producers(apiver_module): + """ + If the file cannot be written, the error must reach the caller and the producers must not + block forever on a queue that nobody drains any more. + """ + from b2sdk._internal.transfer.inbound.downloader.parallel import WriterThread + + outcome = {} + + def produce(): + try: + with WriterThread(FileFailingOnSecondWrite(), max_queue_depth=4) as writer: + for offset in range(50): + writer.queue_write(offset, b'x') + except OSError as e: + outcome['error'] = e + + # daemon thread: if the bug is present, this thread blocks forever and must not keep the + # test process alive + producer = threading.Thread(target=produce, daemon=True) + producer.start() + producer.join(timeout=10) + + assert not producer.is_alive(), 'producer is blocked on a queue that is no longer drained' + assert 'disk full' in str(outcome.get('error')) diff --git a/test/unit/replication/test_monitoring.py b/test/unit/replication/test_monitoring.py index f4955dae2..798068fec 100644 --- a/test/unit/replication/test_monitoring.py +++ b/test/unit/replication/test_monitoring.py @@ -9,6 +9,10 @@ ###################################################################### from __future__ import annotations +import queue +import threading + +import pytest from apiver_deps import ( SSE_B2_AES, EncryptionAlgorithm, @@ -20,6 +24,8 @@ RetentionMode, ) +from b2sdk._internal.replication import monitoring as monitoring_module + SSE_C_AES = EncryptionSetting( mode=EncryptionMode.SSE_C, algorithm=EncryptionAlgorithm.AES256, @@ -280,3 +286,87 @@ def test_scan_source_and_destination( ] == 1 ) + + +class _SpyQueue(queue.Queue): + """A queue that remembers its instances, so the test can release threads stuck on it.""" + + instances: list = [] + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self.instances.append(self) + + +def _release_blocked_threads(spy_queue: queue.Queue, worker: threading.Thread) -> None: + """Unblock whichever side of the queue is stuck, so that a failing test does not leak threads.""" + for _ in range(50): + if not worker.is_alive(): + return + while True: + try: + spy_queue.get_nowait() # frees a producer blocked on put() + except queue.Empty: + break + worker.join(timeout=0.1) + if not worker.is_alive(): + return + try: + spy_queue.put_nowait(None) # frees a consumer blocked on get() + except queue.Full: + pass + worker.join(timeout=0.1) + + +@pytest.mark.parametrize('failing_side', ['producer', 'consumer']) +def test_scan__failure_in_one_thread__is_raised_instead_of_hanging( + source_bucket, test_file, monitor, monkeypatch, failing_side +): + """ + When listing files (producer) or building the report (consumer) raises, ``scan`` must + propagate the error. The two threads talk through a bounded queue, so the surviving + thread must not wait forever for a partner that is gone. + """ + for i in range(6): + source_bucket.upload_local_file(test_file, f'folder/test-{i}.txt') + + _SpyQueue.instances = [] + monkeypatch.setattr(monitoring_module, 'Queue', _SpyQueue) + monkeypatch.setattr(type(monitor), 'QUEUE_SIZE', 2) + + if failing_side == 'producer': + base_folder_class = type(monitor).B2_FOLDER_CLASS + + class FailingFolder(base_folder_class): + def all_files(self, *args, **kwargs): + for i, path in enumerate(super().all_files(*args, **kwargs)): + if i == 2: + raise RuntimeError('listing failed') + yield path + + monkeypatch.setattr(type(monitor), 'B2_FOLDER_CLASS', FailingFolder) + else: + + class FailingReport(type(monitor).REPORT_CLASS): + def add(self, *args, **kwargs): + raise RuntimeError('report failed') + + monkeypatch.setattr(type(monitor), 'REPORT_CLASS', FailingReport) + + outcome = {} + + def run_scan(): + try: + monitor.scan(scan_destination=False) + except RuntimeError as e: + outcome['error'] = e + + # daemon thread: with the bug present, scan() never returns + worker = threading.Thread(target=run_scan, daemon=True) + worker.start() + worker.join(timeout=10) + hung = worker.is_alive() + _release_blocked_threads(_SpyQueue.instances[-1], worker) + + assert not hung, 'scan() is blocked waiting for the other thread' + assert str(outcome.get('error')) in ('listing failed', 'report failed') diff --git a/test/unit/scan/test_folder_traversal.py b/test/unit/scan/test_folder_traversal.py index 4971b08f7..5c601b71f 100644 --- a/test/unit/scan/test_folder_traversal.py +++ b/test/unit/scan/test_folder_traversal.py @@ -793,3 +793,40 @@ def test_excluded_without_permissions(self, tmp_path, fs_perm_tool): assert not reporter.warnings reporter.close() + + @pytest.mark.skipif( + platform.system() == 'Windows', reason='Broken symlinks are only tested on Unix' + ) + @pytest.mark.parametrize('broken_link_name', ['a_broken_link', 'z_broken_link']) + @pytest.mark.parametrize('with_reporter', [True, False]) + def test_broken_symlink_is_skipped(self, tmp_path, broken_link_name, with_reporter): + """ + A symlink to a missing target must be skipped, whether or not a reporter is given + (``reporter`` is optional) and whatever its position among the other entries. + """ + (tmp_path / 'file.txt').write_text('content') + (tmp_path / broken_link_name).symlink_to(tmp_path / 'missing_target') + + reporter = ProgressReport(sys.stdout, False) if with_reporter else None + folder = LocalFolder(str(tmp_path)) + local_paths = list( + folder.all_files(reporter=reporter, policies_manager=ScanPoliciesManager()) + ) + + assert [path.relative_path for path in local_paths] == ['file.txt'] + if reporter is not None: + assert len(reporter.warnings) == 1 + reporter.close() + + @pytest.mark.skipif( + platform.system() == 'Windows', reason='Unix-only filesystem permissions are tested' + ) + def test_unreadable_file_is_skipped_without_reporter(self, tmp_path, fs_perm_tool): + (tmp_path / 'readable.txt').write_text('content') + (tmp_path / 'no_perm_file.txt').write_text('content') + fs_perm_tool.deny_access(tmp_path / 'no_perm_file.txt') + + folder = LocalFolder(str(tmp_path)) + local_paths = list(folder.all_files(reporter=None, policies_manager=ScanPoliciesManager())) + + assert [path.relative_path for path in local_paths] == ['readable.txt'] diff --git a/test/unit/sync/test_sync.py b/test/unit/sync/test_sync.py index a9b51d7ec..ccb4692df 100644 --- a/test/unit/sync/test_sync.py +++ b/test/unit/sync/test_sync.py @@ -32,6 +32,8 @@ ) from apiver_deps_exception import DestFileNewer, InvalidArgument +import b2sdk._internal.sync.sync as sync_module + from ..fixtures.folder import * from .fixtures import * @@ -975,13 +977,11 @@ def check_path_and_get_size(self): large_file_sha1=local_sha1 if should_be_incremental else None, ) ] - # In Python 3.7 unittest.mock.call doesn't have `args` properly defined. Instead we have to take 1st index. - # TODO: use .args[0] instead of [1] when we drop Python 3.7 - num_calls = len(bucket.mock_calls[0][1]) - assert num_calls == 2 if should_be_incremental else 1, bucket.mock_calls[0] + # first positional argument of concatenate() is the list of sources + sources = bucket.mock_calls[0][1][0] + assert len(sources) == (2 if should_be_incremental else 1), bucket.mock_calls[0] if should_be_incremental: - # Order of indices: call index, pick arguments, pick first argument, first element of the first argument. - assert isinstance(bucket.mock_calls[0][1][0][0], CopySource) + assert isinstance(sources[0], CopySource) def test_sync_lexicographical_order(self, synchronizer_factory): """ @@ -1008,6 +1008,59 @@ def test_sync_lexicographical_order(self, synchronizer_factory): dst_with_files = self.folder_factory('b2', *files) self.assert_folder_sync_actions(synchronizer, src, dst_with_files, []) + @pytest.mark.apiver(from_ver=2) + @pytest.mark.parametrize( + 'src_type, dst_type', + [ + ('b2', 'local'), + ('local', 'b2'), + ], + ) + def test_sync_folders_without_reporter(self, synchronizer_factory, src_type, dst_type): + """ + ``sync_folders`` documents ``reporter`` as optional (``SyncReport | None``), so a + sync run without a reporter must complete instead of failing on ``None``. + """ + synchronizer = synchronizer_factory(dry_run=True, allow_empty_source=True) + src = self.folder_factory(src_type, ('a.txt', [200], 10)) + dst = self.folder_factory(dst_type) + + synchronizer.sync_folders(src, dst, TODAY, None) + + @pytest.mark.xfail( + strict=True, + reason=( + 'known bug (review b2-sdk-python prod:7, sync.py:201, owner decides): the executor ' + 'is not shut down when scheduling raises, so queued actions keep running in ' + 'the background; whether to wait for or cancel them is not decided' + ), + ) + def test_sync_folders_shuts_executor_down_when_scheduling_fails(self, synchronizer_factory): + synchronizer = synchronizer_factory(dry_run=True, allow_empty_source=True) + src = self.folder_factory('b2', ('a.txt', [200], 10)) + dst = self.folder_factory('local') + + def failing_actions(*args, **kwargs): + yield mock.MagicMock() + raise RuntimeError('scheduling failed') + + executors = [] + real_executor_class = sync_module.BoundedQueueExecutor + + def spy(*args, **kwargs): + executor = real_executor_class(*args, **kwargs) + executor.shutdown = mock.MagicMock(wraps=executor.shutdown) + executors.append(executor) + return executor + + with mock.patch.object(synchronizer, '_make_folder_sync_actions', failing_actions): + with mock.patch.object(sync_module, 'BoundedQueueExecutor', spy): + with pytest.raises(RuntimeError, match='scheduling failed'): + synchronizer.sync_folders(src, dst, TODAY, None) + + assert len(executors) == 1 + executors[0].shutdown.assert_called_once() + class TstEncryptionSettingsProvider(AbstractSyncEncryptionSettingsProvider): def __init__(self, source_encryption_setting, destination_encryption_setting): diff --git a/test/unit/sync/test_sync_report.py b/test/unit/sync/test_sync_report.py index 496224f79..7d908ad47 100644 --- a/test/unit/sync/test_sync_report.py +++ b/test/unit/sync/test_sync_report.py @@ -27,6 +27,35 @@ def test_bad_terminal(self): sync_report = SyncReport(stdout, False) sync_report.print_completion('transferred: 123.txt') + def test_bad_terminal_falls_back_to_escaped_output_and_warns_once(self): + written = [] + + class BadTerminal: + encoding = 'ascii' + + def __init__(self): + self.fail_next = True + + def write(self, text): + if self.fail_next: + self.fail_next = False + raise UnicodeEncodeError('codec', 'foo', 100, 105, 'artificial') + written.append(text) + + def flush(self): + pass + + sync_report = SyncReport(BadTerminal(), True) + sync_report.print_completion('transferred: caf\u00e9.txt') + sync_report.print_completion('transferred: other.txt') + + output = ''.join(written) + assert ( + output.count('!WARNING! this terminal cannot properly handle progress reporting') == 1 + ) + assert 'transferred: caf\\xe9.txt' in output + assert 'transferred: other.txt' in output + @pytest.mark.apiver(to_ver=1) def test_legacy_methods(self): stdout = MagicMock() diff --git a/test/unit/test_raw_api.py b/test/unit/test_raw_api.py index 4745f02ed..b181f44d5 100644 --- a/test/unit/test_raw_api.py +++ b/test/unit/test_raw_api.py @@ -169,3 +169,38 @@ def test_raw_api_omits_encryption_from_request_when_not_given(raw_api, call, tra sent = {**headers, **(payload if isinstance(payload, dict) else {})} assert not [key for key in sent if 'encryption' in key.lower()], sent + + +def test_start_large_file_sends_custom_upload_timestamp_under_its_wire_name(raw_api): + """ + The b2_start_large_file request body uses camelCase for every field, so the custom + upload timestamp must be sent as ``customUploadTimestamp``. The simulator never looks at + the wire key, so only a request-level assertion can catch a misspelling. + """ + raw_api.start_large_file( + 'api-url', + 'account-token', + 'bucket-id', + 'file-name', + 'text/plain', + {}, + custom_upload_timestamp=123, + ) + + _, _, payload = raw_api.b2_http.post_json_return_json.call_args.args[:3] + + assert payload.get('customUploadTimestamp') == 123, payload + assert 'custom_upload_timestamp' not in payload, payload + + +def test_update_bucket_sends_if_revision_is_under_its_wire_name(raw_api): + """ + ``ifRevisionIs`` is the optimistic-locking guard of b2_update_bucket. If it is not on the + wire, concurrent bucket updates silently overwrite each other, and the simulator, which + takes the Python argument directly, cannot tell. + """ + raw_api.update_bucket('api-url', 'account-token', 'account-id', 'bucket-id', if_revision_is=7) + + _, _, payload = raw_api.b2_http.post_json_return_json.call_args.args[:3] + + assert payload.get('ifRevisionIs') == 7, payload diff --git a/test/unit/test_sdkharness_conformance_resilience_contract.py b/test/unit/test_sdkharness_conformance_resilience_contract.py index 41d47c5a1..a61f945de 100644 --- a/test/unit/test_sdkharness_conformance_resilience_contract.py +++ b/test/unit/test_sdkharness_conformance_resilience_contract.py @@ -162,6 +162,27 @@ def test_leaf_checks_refuse_ambient_staging_runs(leaf): assert 'real-looking' not in refused.stdout + refused.stderr +@pytest.mark.parametrize('leaf', LEAVES, ids=lambda path: f'{path.parent.name}/{path.name}') +def test_leaf_checks_refuse_staging_target_even_with_a_valid_loopback_simulator(leaf): + """ + The staging refusal must be what stops the run. Everything else about this environment is + valid (loopback simulator, test credential), so if the target check were deleted the leaf + would go on to try the simulator instead of refusing. + """ + prefix = leaf.parent.name.upper() + ambient = { + f'{prefix}_TARGET': 'staging', + f'{prefix}_SIMULATOR_URL': 'http://127.0.0.1:8180', + } + if prefix == 'RESILIENCE': + ambient['RESILIENCE_CONTROL_URL'] = 'http://127.0.0.1:8181' + + refused = run_leaf(leaf, ambient) + + assert refused.returncode == 1 + assert 'FAIL (configuration -- only the simulator target is supported' in refused.stdout + + @pytest.mark.parametrize('leaf', LEAVES[:2] + LEAVES[-2:], ids=lambda path: path.name) def test_leaf_checks_refuse_missing_or_non_loopback_simulator(leaf): prefix = leaf.parent.name.upper() diff --git a/test/unit/v0/test_bucket.py b/test/unit/v0/test_bucket.py index 6c1aa2c6d..9dbfe1c6e 100644 --- a/test/unit/v0/test_bucket.py +++ b/test/unit/v0/test_bucket.py @@ -220,7 +220,16 @@ def __call__(self, *args, **kwargs): raise InvalidAuthToken('message', 401) self.simulator.create_bucket = InvalidAuthTokenWrapper(self.simulator.create_bucket) - self.bucket = self.api.create_bucket('your-bucket', 'allPublic') + with mock.patch.object( + self.api.session, 'authorize_account', wraps=self.api.session.authorize_account + ) as authorize_account: + self.bucket = self.api.create_bucket('your-bucket', 'allPublic') + + # the invalid token must have triggered exactly one re-authorization, and the retried + # call must have created the bucket + authorize_account.assert_called_once() + self.assertEqual('your-bucket', self.bucket.name) + self.assertEqual(self.bucket.id_, self.api.get_bucket_by_name('your-bucket').id_) class TestListParts(TestCaseWithBucket): @@ -598,6 +607,7 @@ def test_copy_retention(self): resulting_file_version = self.bucket.copy( file_id, 'copied_file', + length=length, file_retention=FileRetentionSetting(RetentionMode.COMPLIANCE, 100), legal_hold=LegalHold.ON, max_part_size=400, diff --git a/test/unit/v0/test_sync.py b/test/unit/v0/test_sync.py index 4ec679284..a4713053c 100644 --- a/test/unit/v0/test_sync.py +++ b/test/unit/v0/test_sync.py @@ -973,7 +973,6 @@ def test_wait_for_running_jobs(self): """ raw_executor = futures.ThreadPoolExecutor(2) bounded_executor = BoundedQueueExecutor(raw_executor, 1) - assert_equal = self.assertEqual class CountAtOnce: """ @@ -984,12 +983,14 @@ class CountAtOnce: def __init__(self): self.running_at_once = 0 + self.max_seen = 0 + self.counter = 0 self.lock = threading.Lock() def run(self): with self.lock: self.running_at_once += 1 - assert_equal(1, self.running_at_once) + self.max_seen = max(self.max_seen, self.running_at_once) # While we are sleeping here, no other actions should start # running. If they do, they will increment the counter and # fail the above assertion. @@ -1004,6 +1005,9 @@ def run(self): for _ in range(5): bounded_executor.submit(count_at_once.run) bounded_executor.shutdown() + self.assertEqual(1, count_at_once.max_seen) + self.assertEqual(5, count_at_once.counter) + self.assertEqual(2, bounded_executor.get_num_exceptions()) if __name__ == '__main__': diff --git a/test/unit/v1/test_bounded_queue_executor.py b/test/unit/v1/test_bounded_queue_executor.py index 8f94a857f..293027314 100644 --- a/test/unit/v1/test_bounded_queue_executor.py +++ b/test/unit/v1/test_bounded_queue_executor.py @@ -10,6 +10,7 @@ from __future__ import annotations import concurrent.futures as futures +import threading import time from ..test_base import TestBase @@ -46,6 +47,33 @@ def fcn(): answers = list(map(lambda f: f.result(), futures)) self.assertEqual(list(range(10)), answers) + def test_submit_blocks_until_a_slot_is_free(self): + """With a queue limit of 1, a second submit() must wait for the first job to finish.""" + release_first_job = threading.Event() + first_job_started = threading.Event() + + def first_job(): + first_job_started.set() + release_first_job.wait(timeout=10) + + self.executor.submit(first_job) + assert first_job_started.wait(timeout=10) + + second_submit_returned = threading.Event() + + def submit_second_job(): + self.executor.submit(lambda: None) + second_submit_returned.set() + + # daemon thread: it must not keep the test process alive if submit() blocks for good + submitter = threading.Thread(target=submit_second_job, daemon=True) + submitter.start() + try: + self.assertFalse(second_submit_returned.wait(timeout=0.3), 'submit() did not block') + finally: + release_first_job.set() + self.assertTrue(second_submit_returned.wait(timeout=10), 'submit() never returned') + def test_no_exceptions(self): f = self.executor.submit(lambda: 1) self.executor.shutdown() diff --git a/test/unit/v1/test_raw_api.py b/test/unit/v1/test_raw_api.py index 4202fa653..0f9aad68a 100644 --- a/test/unit/v1/test_raw_api.py +++ b/test/unit/v1/test_raw_api.py @@ -116,17 +116,26 @@ def test_assertion_raises(self): self.raw_api.update_bucket('test', 'account_auth_token', 'account_id', 'bucket_id') @pytest.mark.parametrize( - 'bucket_type,bucket_info,default_retention', + 'bucket_type,bucket_info,default_retention,expected_payload', ( - (None, {}, None), + (None, {}, None, {'bucketInfo': {}}), ( 'allPublic', None, BucketRetentionSetting(RetentionMode.COMPLIANCE, RetentionPeriod(years=1)), + { + 'bucketType': 'allPublic', + 'defaultRetention': { + 'mode': 'compliance', + 'period': {'duration': 1, 'unit': 'years'}, + }, + }, ), ), ) - def test_assertion_not_raises(self, bucket_type, bucket_info, default_retention): + def test_assertion_not_raises( + self, bucket_type, bucket_info, default_retention, expected_payload + ): self.raw_api.update_bucket( 'test', 'account_auth_token', @@ -137,8 +146,15 @@ def test_assertion_not_raises(self, bucket_type, bucket_info, default_retention) default_retention=default_retention, ) + _, _, payload = self.raw_api.b2_http.post_json_return_json.call_args.args[:3] + assert payload == { + 'accountId': 'account_id', + 'bucketId': 'bucket_id', + **expected_payload, + } + @pytest.mark.parametrize( - 'encryption_setting,', + 'encryption_setting', ( EncryptionSetting( mode=EncryptionMode.SSE_C, @@ -169,7 +185,7 @@ class TestCreateBucket(BucketTestBase): """Test creating bucket.""" @pytest.mark.parametrize( - 'encryption_setting,', + 'encryption_setting', ( EncryptionSetting( mode=EncryptionMode.SSE_C, diff --git a/test/unit/v1/test_sync.py b/test/unit/v1/test_sync.py index 03e0cf189..cf51bd433 100644 --- a/test/unit/v1/test_sync.py +++ b/test/unit/v1/test_sync.py @@ -999,7 +999,6 @@ def test_wait_for_running_jobs(self): """ raw_executor = futures.ThreadPoolExecutor(2) bounded_executor = BoundedQueueExecutor(raw_executor, 1) - assert_equal = self.assertEqual class CountAtOnce: """ @@ -1010,12 +1009,14 @@ class CountAtOnce: def __init__(self): self.running_at_once = 0 + self.max_seen = 0 + self.counter = 0 self.lock = threading.Lock() def run(self): with self.lock: self.running_at_once += 1 - assert_equal(1, self.running_at_once) + self.max_seen = max(self.max_seen, self.running_at_once) # While we are sleeping here, no other actions should start # running. If they do, they will increment the counter and # fail the above assertion. @@ -1030,6 +1031,9 @@ def run(self): for _ in range(5): bounded_executor.submit(count_at_once.run) bounded_executor.shutdown() + self.assertEqual(1, count_at_once.max_seen) + self.assertEqual(5, count_at_once.counter) + self.assertEqual(2, bounded_executor.get_num_exceptions()) if __name__ == '__main__':