From d7ebaae52c386d53f89079b3257234cfc3daa7f7 Mon Sep 17 00:00:00 2001 From: kris-konina-reef Date: Mon, 23 Sep 2024 12:27:17 +0000 Subject: [PATCH 1/7] Add persistent bucket helpers --- test/integration/persistent_bucket.py | 89 +++++++++++++++++++++++++++ 1 file changed, 89 insertions(+) create mode 100644 test/integration/persistent_bucket.py diff --git a/test/integration/persistent_bucket.py b/test/integration/persistent_bucket.py new file mode 100644 index 00000000..8a44e83d --- /dev/null +++ b/test/integration/persistent_bucket.py @@ -0,0 +1,89 @@ +###################################################################### +# +# File: test/integration/persistent_bucket.py +# +# Copyright 2024 Backblaze Inc. All Rights Reserved. +# +# License https://www.backblaze.com/using_b2_code.html +# +###################################################################### +import hashlib +import os +import uuid +from dataclasses import dataclass +from functools import cached_property +from test.integration.helpers import BUCKET_NAME_LENGTH + +from b2sdk._internal.bucket import Bucket +from b2sdk.v2 import B2Api +from b2sdk.v2.exception import NonExistentBucket + +PERSISTENT_BUCKET_NAME_PREFIX = "constst" + + +@dataclass +class PersistentBucketAggregate: + bucket: Bucket + + def __post_init__(self): + self.subfolder = self.new_subfolder() + + @property + def bucket_name(self) -> str: + return self.bucket.name + + def new_subfolder(self) -> str: + return f"test-{uuid.uuid4().hex[:8]}" + + @property + def bucket_id(self): + return self.bucket.id_ + + @cached_property + def b2_uri(self): + return f"b2://{self.bucket_name}/{self.subfolder}" + + +def hash_dict_sha256(d): + """ + Create a sha256 hash of the given dictionary. + """ + dict_repr = repr(sorted((k, repr(v)) for k, v in d.items())) + hash_obj = hashlib.sha256() + hash_obj.update(dict_repr.encode('utf-8')) + return hash_obj.hexdigest() + + +def get_persistent_bucket_name(b2_api: B2Api, create_options: dict) -> str: + """ + Create a hash of the `create_options` dictionary, include it in the bucket name + so that we can easily reuse buckets with the same options across (parallel) test runs. + """ + # Exclude sensitive options from the hash + unsafe_options = {"authorizationToken", "accountId", "default_server_side_encryption"} + create_options_hashable = {k: v for k, v in create_options.items() if k not in unsafe_options} + hashed_options = hash_dict_sha256(create_options_hashable) + bucket_owner = os.environ.get("GITHUB_REPOSITORY_ID", b2_api.get_account_id()) + bucket_base = f"{bucket_owner}:{hashed_options}" + bucket_hash = hashlib.sha256(bucket_base.encode()).hexdigest() + return f"{PERSISTENT_BUCKET_NAME_PREFIX}-{bucket_hash}" [:BUCKET_NAME_LENGTH] + + +def get_or_create_persistent_bucket(b2_api: B2Api, **create_options) -> Bucket: + bucket_name = get_persistent_bucket_name(b2_api, create_options.copy()) + try: + bucket = b2_api.get_bucket_by_name(bucket_name) + except NonExistentBucket: + bucket = b2_api.create_bucket( + bucket_name, + bucket_type="allPublic", + lifecycle_rules=[ + { + "daysFromHidingToDeleting": 1, + "daysFromUploadingToHiding": 1, + "fileNamePrefix": "", + } + ], + **create_options, + ) + return bucket From fa3158bffe019fab1bd7eb9c25487faf9cd0b672 Mon Sep 17 00:00:00 2001 From: kris-konina-reef Date: Mon, 23 Sep 2024 12:29:41 +0000 Subject: [PATCH 2/7] Phase out new bucket creation in favor of fetching/recreating persistent bucket --- test/integration/test_download.py | 37 +++++++++++-------- .../test_file_version_attributes.py | 8 ++-- test/integration/test_sync.py | 9 ++++- test/integration/test_upload.py | 16 ++++---- 4 files changed, 39 insertions(+), 31 deletions(-) diff --git a/test/integration/test_download.py b/test/integration/test_download.py index ef02ac80..cb5db69b 100644 --- a/test/integration/test_download.py +++ b/test/integration/test_download.py @@ -30,7 +30,8 @@ class TestDownload(IntegrationTestBase): def test_large_file(self): - bucket = self.create_bucket() + bucket = self.persistent_bucket.bucket + subfolder = self.persistent_bucket.subfolder with mock.patch.object( self.info, '_recommended_part_size', new=self.info.get_absolute_minimum_part_size() ): @@ -49,12 +50,12 @@ def test_large_file(self): ): # let's check that small file downloads do not fail with these settings - small_file_version = bucket.upload_bytes(b'0', 'a_single_char') + small_file_version = bucket.upload_bytes(b'0', f'{subfolder}/a_single_char') with io.BytesIO() as io_: - bucket.download_file_by_name('a_single_char').save(io_) + bucket.download_file_by_name(f'{subfolder}/a_single_char').save(io_) assert io_.getvalue() == b'0' - f, sha1 = self._file_helper(bucket) + f, sha1 = self._file_helper() if small_file_version._type() != 'large': # if we are here, that's not the production server! assert f.download_version.content_sha1_verified # large files don't have sha1, lets not check @@ -63,9 +64,11 @@ def test_large_file(self): assert LARGE_FILE_SHA1 in file_info assert file_info[LARGE_FILE_SHA1] == sha1 - def _file_helper(self, bucket, sha1_sum=None, + def _file_helper(self, sha1_sum=None, bytes_to_write: int | None = None) -> tuple[DownloadVersion, Sha1HexDigest]: bytes_to_write = bytes_to_write or int(self.info.get_absolute_minimum_part_size()) * 2 + 1 + + bucket = self.persistent_bucket.bucket with tempfile.TemporaryDirectory() as temp_dir: temp_dir = pathlib.Path(temp_dir) source_small_file = pathlib.Path(temp_dir) / 'source_small_file' @@ -73,12 +76,12 @@ def _file_helper(self, bucket, sha1_sum=None, self.write_zeros(small_file, bytes_to_write) bucket.upload_local_file( source_small_file, - 'small_file', + f'{self.persistent_bucket.subfolder}/small_file', sha1_sum=sha1_sum, ) target_small_file = pathlib.Path(temp_dir) / 'target_small_file' - f = bucket.download_file_by_name('small_file') + f = bucket.download_file_by_name(f'{self.persistent_bucket.subfolder}/small_file') f.save_to(target_small_file) source_sha1 = hex_sha1_of_file(source_small_file) @@ -86,20 +89,18 @@ def _file_helper(self, bucket, sha1_sum=None, return f, source_sha1 def test_small(self): - bucket = self.create_bucket() - f, _ = self._file_helper(bucket, bytes_to_write=1) + f, _ = self._file_helper(bytes_to_write=1) assert f.download_version.content_sha1_verified def test_small_unverified(self): - bucket = self.create_bucket() - f, _ = self._file_helper(bucket, sha1_sum='do_not_verify', bytes_to_write=1) + f, _ = self._file_helper(sha1_sum='do_not_verify', bytes_to_write=1) if f.download_version.content_sha1_verified: pprint(f.download_version._get_args_for_clone()) assert not f.download_version.content_sha1_verified @pytest.mark.parametrize("size_multiplier", [1, 100]) -def test_gzip(b2_auth_data, bucket, tmp_path, b2_api, size_multiplier): +def test_gzip(b2_auth_data, persistent_bucket, tmp_path, b2_api, size_multiplier): """Test downloading gzipped files of varius sizes with and without content-encoding.""" source_file = tmp_path / 'compressed_file.gz' downloaded_uncompressed_file = tmp_path / 'downloaded_uncompressed_file' @@ -107,8 +108,10 @@ def test_gzip(b2_auth_data, bucket, tmp_path, b2_api, size_multiplier): data_to_write = b"I'm about to be compressed and sent to the cloud, yay!\n" * size_multiplier source_file.write_bytes(gzip.compress(data_to_write)) - file_version = bucket.upload_local_file( - str(source_file), 'gzipped_file', file_info={'b2-content-encoding': 'gzip'} + file_version = persistent_bucket.bucket.upload_local_file( + str(source_file), + f'{persistent_bucket.subfolder}/gzipped_file', + file_info={'b2-content-encoding': 'gzip'} ) b2_api.download_file_by_id(file_id=file_version.id_).save_to(str(downloaded_compressed_file)) assert downloaded_compressed_file.read_bytes() == source_file.read_bytes() @@ -128,8 +131,10 @@ def source_file(tmp_path): @pytest.fixture -def uploaded_source_file_version(bucket, source_file): - file_version = bucket.upload_local_file(str(source_file), source_file.name) +def uploaded_source_file_version(persistent_bucket, source_file): + file_version = persistent_bucket.bucket.upload_local_file( + str(source_file), f'{persistent_bucket.subfolder}/{source_file.name}' + ) return file_version diff --git a/test/integration/test_file_version_attributes.py b/test/integration/test_file_version_attributes.py index 4f137eeb..cf95ae27 100644 --- a/test/integration/test_file_version_attributes.py +++ b/test/integration/test_file_version_attributes.py @@ -19,11 +19,11 @@ def _assert_object_has_attributes(self, object, kwargs): for key, value in kwargs.items(): assert getattr(object, key) == value - def test_file_info_b2_attributes(self): + def test_file_info_b2_attributes(self, persistent_bucket): # This test checks that attributes that are internally represented as file_info items with prefix `b2-` # are saved and retrieved correctly. - bucket = self.create_bucket() + bucket = persistent_bucket.bucket expected_attributes = { 'cache_control': 'max-age=3600', 'expires': 'Wed, 21 Oct 2105 07:28:00 GMT', @@ -36,7 +36,7 @@ def test_file_info_b2_attributes(self): dt.datetime(2105, 10, 21, 7, 28, tzinfo=dt.timezone.utc) } - file_version = bucket.upload_bytes(b'0', 'file', **kwargs) + file_version = bucket.upload_bytes(b'0', f'{persistent_bucket.subfolder}/file', **kwargs) self._assert_object_has_attributes(file_version, expected_attributes) file_version = bucket.get_file_info_by_id(file_version.id_) @@ -47,7 +47,7 @@ def test_file_info_b2_attributes(self): copied_version = bucket.copy( file_version.id_, - 'file_copy', + f'{persistent_bucket.subfolder}/file_copy', content_type='text/plain', **{ **kwargs, 'content_language': 'de' diff --git a/test/integration/test_sync.py b/test/integration/test_sync.py index 7366e4f5..c706fcdc 100644 --- a/test/integration/test_sync.py +++ b/test/integration/test_sync.py @@ -35,9 +35,14 @@ def local_folder_with_files(tmp_path): return folder -def test_sync_folder(b2_api, local_folder_with_files, b2_subfolder): +@pytest.fixture +def persistent_bucket(persistent_bucket_factory): + return persistent_bucket_factory() + + +def test_sync_folder(b2_api, local_folder_with_files, persistent_bucket): source_folder = parse_folder(str(local_folder_with_files), b2_api) - dest_folder = parse_folder(b2_subfolder, b2_api) + dest_folder = parse_folder(persistent_bucket.b2_uri, b2_api) synchronizer = Synchronizer( max_workers=10, diff --git a/test/integration/test_upload.py b/test/integration/test_upload.py index 1dba4986..0973bc86 100644 --- a/test/integration/test_upload.py +++ b/test/integration/test_upload.py @@ -17,14 +17,13 @@ from b2sdk.v2 import B2RawHTTPApi from .base import IntegrationTestBase -from .test_raw_api import authorize_raw_api class TestUnboundStreamUpload(IntegrationTestBase): def assert_data_uploaded_via_stream(self, data: bytes, part_size: int | None = None): - bucket = self.create_bucket() + bucket = self.persistent_bucket.bucket stream = io.BytesIO(data) - file_name = 'unbound_stream' + file_name = f'{self.persistent_bucket.subfolder}/unbound_stream' bucket.upload_unbound_stream(stream, file_name, recommended_upload_part_size=part_size) @@ -46,7 +45,7 @@ def test_streamed_large_buffer_small_part_size(self): class TestUploadLargeFile(IntegrationTestBase): - def test_ssec_key_id(self): + def test_ssec_key_id(self, auth_info): sse_c = EncryptionSetting( mode=EncryptionMode.SSE_C, algorithm=EncryptionAlgorithm.AES256, @@ -55,16 +54,15 @@ def test_ssec_key_id(self): raw_api = B2RawHTTPApi(B2Http()) - auth_dict = authorize_raw_api(raw_api) - account_auth_token = auth_dict['authorizationToken'] - api_url = auth_dict['apiUrl'] - bucket = self.create_bucket() + account_auth_token = auth_info['authorizationToken'] + api_url = auth_info['apiUrl'] + bucket = self.persistent_bucket.bucket large_info = raw_api.start_large_file( api_url, account_auth_token, bucket.id_, - 'test_largefile_sse_c.txt', + f'{self.persistent_bucket.subfolder}/test_largefile_sse_c.txt', 'text/plain', None, server_side_encryption=sse_c, From 184580e9e71977c5e264408d41427ffc272fdfc1 Mon Sep 17 00:00:00 2001 From: kris-konina-reef Date: Mon, 23 Sep 2024 12:30:14 +0000 Subject: [PATCH 3/7] Add persistent bucket fixtures --- test/integration/conftest.py | 45 ++++++++++++++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/test/integration/conftest.py b/test/integration/conftest.py index aacb348c..0dcdfdb8 100644 --- a/test/integration/conftest.py +++ b/test/integration/conftest.py @@ -21,10 +21,18 @@ get_bucket_name_prefix, random_bucket_name, ) +from test.integration.persistent_bucket import ( + PersistentBucketAggregate, + get_or_create_persistent_bucket, +) +from typing import Callable import pytest +from b2sdk._internal.b2http import B2Http +from b2sdk._internal.raw_api import REALM_URLS from b2sdk._internal.utils import current_time_millis +from b2sdk.v2.raw_api import B2RawHTTPApi def pytest_addoption(parser): @@ -100,3 +108,40 @@ def bucket(b2_api, bucket_name_prefix, bucket_cleaner): def b2_subfolder(bucket, request): subfolder_name = f"{request.node.name}_{secrets.token_urlsafe(4)}" return f"b2://{bucket.name}/{subfolder_name}" + + +@pytest.fixture(scope="class") +def raw_api(): + return B2RawHTTPApi(B2Http()) + + +@pytest.fixture(scope="class") +def auth_info(raw_api): + application_key_id = os.environ.get('B2_TEST_APPLICATION_KEY_ID') + application_key = os.environ.get('B2_TEST_APPLICATION_KEY') + if application_key_id is None or application_key is None: + pytest.fail('B2_TEST_APPLICATION_KEY_ID or B2_TEST_APPLICATION_KEY is not set.') + + realm = os.environ.get('B2_TEST_ENVIRONMENT', 'production') + realm_url = REALM_URLS.get(realm, realm) + return raw_api.authorize_account(realm_url, application_key_id, application_key) + + +# -- Persistent bucket fixtures -- +@pytest.fixture(scope="session") +def persistent_bucket_factory(b2_api) -> Callable[[], PersistentBucketAggregate]: + """ + Since all consumers of the `bucket_name` fixture expect a new bucket to be created, + we need to mirror this behavior by appending a unique subfolder to the persistent bucket name. + """ + + def _persistent_bucket(**bucket_create_options): + persistent_bucket = get_or_create_persistent_bucket(b2_api, **bucket_create_options) + return PersistentBucketAggregate(persistent_bucket) + + yield _persistent_bucket + + +@pytest.fixture(scope="class") +def persistent_bucket(persistent_bucket_factory): + return persistent_bucket_factory() From cdaa5278092afa6c2815a8ab87ed2e1bca9ecbca Mon Sep 17 00:00:00 2001 From: kris-konina-reef Date: Mon, 23 Sep 2024 12:31:50 +0000 Subject: [PATCH 4/7] Refactor B2RawHTTPApi integration tests --- test/integration/base.py | 57 +- test/integration/cleanup_buckets.py | 2 +- test/integration/helpers.py | 90 +++ test/integration/test_raw_api.py | 1023 ++++++++++++++------------- 4 files changed, 619 insertions(+), 553 deletions(-) diff --git a/test/integration/base.py b/test/integration/base.py index 99e7a755..3a238795 100644 --- a/test/integration/base.py +++ b/test/integration/base.py @@ -10,15 +10,11 @@ from __future__ import annotations from test.integration.bucket_cleaner import BucketCleaner -from test.integration.helpers import ( - BUCKET_CREATED_AT_MILLIS, - random_bucket_name, -) +from test.integration.persistent_bucket import PersistentBucketAggregate import pytest -from b2sdk.v2 import B2Api, current_time_millis -from b2sdk.v2.exception import DuplicateBucketName +from b2sdk.v2 import B2Api @pytest.mark.usefixtures("cls_setup") @@ -26,25 +22,19 @@ class IntegrationTestBase: b2_api: B2Api this_run_bucket_name_prefix: str bucket_cleaner: BucketCleaner + persistent_bucket: PersistentBucketAggregate @pytest.fixture(autouse=True, scope="class") - def cls_setup(self, request, b2_api, b2_auth_data, bucket_name_prefix, bucket_cleaner): + def cls_setup( + self, request, b2_api, b2_auth_data, bucket_name_prefix, bucket_cleaner, persistent_bucket + ): cls = request.cls cls.b2_auth_data = b2_auth_data cls.this_run_bucket_name_prefix = bucket_name_prefix cls.bucket_cleaner = bucket_cleaner cls.b2_api = b2_api cls.info = b2_api.account_info - - @pytest.fixture(autouse=True) - def setup_method(self): - self.buckets_created = [] - yield - for bucket in self.buckets_created: - self.bucket_cleaner.cleanup_bucket(bucket) - - def generate_bucket_name(self): - return random_bucket_name(self.this_run_bucket_name_prefix) + cls.persistent_bucket = persistent_bucket def write_zeros(self, file, number): line = b'0' * 1000 + b'\n' @@ -53,36 +43,3 @@ def write_zeros(self, file, number): while written <= number: file.write(line) written += line_len - - def create_bucket(self): - bucket_name = self.generate_bucket_name() - try: - bucket = self.b2_api.create_bucket( - bucket_name, - 'allPublic', - bucket_info={BUCKET_CREATED_AT_MILLIS: str(current_time_millis())} - ) - except DuplicateBucketName: - self._duplicated_bucket_name_debug_info(bucket_name) - raise - self.buckets_created.append(bucket) - return bucket - - def _duplicated_bucket_name_debug_info(self, bucket_name: str) -> None: - # Trying to obtain as much information as possible about this bucket. - print(' DUPLICATED BUCKET DEBUG START '.center(60, '=')) - bucket = self.b2_api.get_bucket_by_name(bucket_name) - - print('Bucket metadata:') - bucket_dict = bucket.as_dict() - for info_key, info in bucket_dict.items(): - print(f'\t{info_key}: "{info}"') - - print('All files (and their versions) inside the bucket:') - ls_generator = bucket.ls(recursive=True, latest_only=False) - for file_version, _directory in ls_generator: - # as_dict() is bound to have more info than we can use, - # but maybe some of it will cast some light on the issue. - print(f'\t{file_version.file_name} ({file_version.as_dict()})') - - print(' DUPLICATED BUCKET DEBUG END '.center(60, '=')) diff --git a/test/integration/cleanup_buckets.py b/test/integration/cleanup_buckets.py index 220aaf57..c751af8a 100755 --- a/test/integration/cleanup_buckets.py +++ b/test/integration/cleanup_buckets.py @@ -13,7 +13,7 @@ from . import get_b2_auth_data from .bucket_cleaner import BucketCleaner -from .test_raw_api import cleanup_old_buckets +from .helpers import cleanup_old_buckets if __name__ == '__main__': cleanup_old_buckets() diff --git a/test/integration/helpers.py b/test/integration/helpers.py index 14147476..16b8dbd7 100644 --- a/test/integration/helpers.py +++ b/test/integration/helpers.py @@ -10,8 +10,14 @@ from __future__ import annotations import os +import re import secrets +import sys +import time +from b2sdk._internal.b2http import B2Http +from b2sdk._internal.file_lock import NO_RETENTION_FILE_SETTING +from b2sdk._internal.raw_api import REALM_URLS, B2RawHTTPApi from b2sdk.v2 import ( BUCKET_NAME_CHARS_UNIQ, BUCKET_NAME_LENGTH_RANGE, @@ -45,3 +51,87 @@ def authorize(b2_auth_data, api_config=DEFAULT_HTTP_API_CONFIG): realm = os.environ.get('B2_TEST_ENVIRONMENT', 'production') b2_api.authorize_account(realm, *b2_auth_data) return b2_api, info + + +def authorize_raw_api(raw_api): + application_key_id = os.environ.get('B2_TEST_APPLICATION_KEY_ID') + if application_key_id is None: + print('B2_TEST_APPLICATION_KEY_ID is not set.', file=sys.stderr) + sys.exit(1) + + application_key = os.environ.get('B2_TEST_APPLICATION_KEY') + if application_key is None: + print('B2_TEST_APPLICATION_KEY is not set.', file=sys.stderr) + sys.exit(1) + + realm = os.environ.get('B2_TEST_ENVIRONMENT', 'production') + realm_url = REALM_URLS.get(realm, realm) + auth_dict = raw_api.authorize_account(realm_url, application_key_id, application_key) + return auth_dict + + +def cleanup_old_buckets(): + raw_api = B2RawHTTPApi(B2Http()) + auth_dict = authorize_raw_api(raw_api) + bucket_list_dict = raw_api.list_buckets( + auth_dict['apiUrl'], auth_dict['authorizationToken'], auth_dict['accountId'] + ) + _cleanup_old_buckets(raw_api, auth_dict, bucket_list_dict) + + +def _cleanup_old_buckets(raw_api, auth_dict, bucket_list_dict): + for bucket_dict in bucket_list_dict['buckets']: + bucket_id = bucket_dict['bucketId'] + bucket_name = bucket_dict['bucketName'] + if _should_delete_bucket(bucket_name): + print('cleaning up old bucket: ' + bucket_name) + _clean_and_delete_bucket( + raw_api, + auth_dict['apiUrl'], + auth_dict['authorizationToken'], + auth_dict['accountId'], + bucket_id, + ) + + +def _clean_and_delete_bucket(raw_api, api_url, account_auth_token, account_id, bucket_id): + # Delete the files. This test never creates more than a few files, + # so one call to list_file_versions should get them all. + versions_dict = raw_api.list_file_versions(api_url, account_auth_token, bucket_id) + for version_dict in versions_dict['files']: + file_id = version_dict['fileId'] + file_name = version_dict['fileName'] + action = version_dict['action'] + if action in ['hide', 'upload']: + print('b2_delete_file', file_name, action) + if action == 'upload' and version_dict[ + 'fileRetention'] and version_dict['fileRetention']['value']['mode'] is not None: + raw_api.update_file_retention( + api_url, + account_auth_token, + file_id, + file_name, + NO_RETENTION_FILE_SETTING, + bypass_governance=True + ) + raw_api.delete_file_version(api_url, account_auth_token, file_id, file_name) + else: + print('b2_cancel_large_file', file_name) + raw_api.cancel_large_file(api_url, account_auth_token, file_id) + + # Delete the bucket + print('b2_delete_bucket', bucket_id) + raw_api.delete_bucket(api_url, account_auth_token, account_id, bucket_id) + + +def _should_delete_bucket(bucket_name): + # Bucket names for this test look like: c7b22d0b0ad7-1460060364-5670 + # Other buckets should not be deleted. + match = re.match(r'^test-raw-api-[a-f0-9]+-([0-9]+)-([0-9]+)', bucket_name) + if match is None: + return False + + # Is it more than an hour old? + bucket_time = int(match.group(1)) + now = time.time() + return bucket_time + 3600 <= now \ No newline at end of file diff --git a/test/integration/test_raw_api.py b/test/integration/test_raw_api.py index 07d36d8b..c394b9fa 100644 --- a/test/integration/test_raw_api.py +++ b/test/integration/test_raw_api.py @@ -12,11 +12,10 @@ import io import os import random -import re -import sys import time -import traceback from test.helpers import type_validator_factory +from test.integration.helpers import _clean_and_delete_bucket +from test.integration.persistent_bucket import PersistentBucketAggregate from typing import List import pytest @@ -29,7 +28,6 @@ ) from b2sdk._internal.exception import DisablingFileLockNotSupported, Unauthorized from b2sdk._internal.file_lock import ( - NO_RETENTION_FILE_SETTING, BucketRetentionSetting, FileRetentionSetting, RetentionMode, @@ -46,170 +44,175 @@ from b2sdk._internal.utils import hex_sha1_of_stream -# TODO: rewrite to separate test cases after introduction of reusable bucket -def test_raw_api(dont_cleanup_old_buckets): - """ - Exercise the code in B2RawHTTPApi by making each call once, just - to make sure the parameters are passed in, and the result is - passed back. +@pytest.fixture(scope="class") +def raw_api(): + return B2RawHTTPApi(B2Http()) - The goal is to be a complete test of B2RawHTTPApi, so the tests for - the rest of the code can use the simulator. - Prints to stdout if things go wrong. - - :return: 0 on success, non-zero on failure - """ - application_key_id = os.environ.get('B2_TEST_APPLICATION_KEY_ID') - if application_key_id is None: - pytest.fail('B2_TEST_APPLICATION_KEY_ID is not set.') - - application_key = os.environ.get('B2_TEST_APPLICATION_KEY') - if application_key is None: - pytest.fail('B2_TEST_APPLICATION_KEY is not set.') - - print() - - try: - raw_api = B2RawHTTPApi(B2Http()) - raw_api_test_helper(raw_api, not dont_cleanup_old_buckets) - except Exception: - traceback.print_exc(file=sys.stdout) - pytest.fail('test_raw_api failed') - - -def authorize_raw_api(raw_api): +@pytest.fixture(scope="class") +def auth_info(raw_api): application_key_id = os.environ.get('B2_TEST_APPLICATION_KEY_ID') - if application_key_id is None: - print('B2_TEST_APPLICATION_KEY_ID is not set.', file=sys.stderr) - sys.exit(1) - application_key = os.environ.get('B2_TEST_APPLICATION_KEY') - if application_key is None: - print('B2_TEST_APPLICATION_KEY is not set.', file=sys.stderr) - sys.exit(1) + if application_key_id is None or application_key is None: + pytest.fail('B2_TEST_APPLICATION_KEY_ID or B2_TEST_APPLICATION_KEY is not set.') realm = os.environ.get('B2_TEST_ENVIRONMENT', 'production') realm_url = REALM_URLS.get(realm, realm) - auth_dict = raw_api.authorize_account(realm_url, application_key_id, application_key) - return auth_dict - - -def raw_api_test_helper(raw_api, should_cleanup_old_buckets): - """ - Try each of the calls to the raw api. Raise an - exception if anything goes wrong. - - This uses a Backblaze account that is just for this test. - The account uses the free level of service, which should - be enough to run this test a reasonable number of times - each day. If somebody abuses the account for other things, - this test will break and we'll have to do something about - it. - """ - # b2_authorize_account - print('b2_authorize_account') - auth_dict = authorize_raw_api(raw_api) - - preview_feature_caps = { - 'readBucketNotifications', - 'writeBucketNotifications', - } - missing_capabilities = ( - set(ALL_CAPABILITIES) - {'readBuckets', 'listAllBucketNames'} - preview_feature_caps - - set(auth_dict['allowed']['capabilities']) - ) - assert not missing_capabilities, 'it appears that the raw_api integration test is being run with a non-full key. Missing capabilities: {}'.format( - missing_capabilities, - ) + return raw_api.authorize_account(realm_url, application_key_id, application_key) - account_id = auth_dict['accountId'] - account_auth_token = auth_dict['authorizationToken'] - api_url = auth_dict['apiUrl'] - download_url = auth_dict['downloadUrl'] - - # b2_create_key - print('b2_create_key') - key_dict = raw_api.create_key( - api_url, - account_auth_token, - account_id, - ['readFiles'], - 'testKey', - None, - None, - None, - ) - # b2_list_keys - print('b2_list_keys') - raw_api.list_keys(api_url, account_auth_token, account_id, 10) - - # b2_delete_key - print('b2_delete_key') - raw_api.delete_key(api_url, account_auth_token, key_dict['applicationKeyId']) - - # b2_create_bucket, with a unique bucket name - # Include the account ID in the bucket name to be - # sure it doesn't collide with bucket names from - # other accounts. - print('b2_create_bucket') - bucket_name = 'test-raw-api-%s-%d-%d' % ( - account_id, int(time.time()), random.randint(1000, 9999) +@pytest.fixture(scope="session") +def sse_b2_aes(): + return EncryptionSetting( + mode=EncryptionMode.SSE_B2, + algorithm=EncryptionAlgorithm.AES256, ) - # very verbose http debug - #import http.client; http.client.HTTPConnection.debuglevel = 1 +@pytest.fixture(scope="class") +def test_bucket(raw_api, auth_info): + bucket_name = f'test-raw-api-{auth_info["accountId"]}-{int(time.time())}-{random.randint(1000, 9999)}' bucket_dict = raw_api.create_bucket( - api_url, - account_auth_token, - account_id, + auth_info['apiUrl'], + auth_info['authorizationToken'], + auth_info['accountId'], bucket_name, 'allPublic', is_file_lock_enabled=True, + lifecycle_rules=[ + { + "fileNamePrefix": "", + "daysFromHidingToDeleting": 1, + "daysFromUploadingToHiding": 1 + } + ] ) - bucket_id = bucket_dict['bucketId'] - first_bucket_revision = bucket_dict['revision'] - - ################################# - print('b2 / replication') - - # 1) create source key (read permissions) - replication_source_key_dict = raw_api.create_key( - api_url, - account_auth_token, - account_id, - [ - 'listBuckets', - 'listFiles', - 'readFiles', - 'writeFiles', # Pawel @ 2022-06-21: adding this to make tests pass with a weird server validator - ], - 'testReplicationSourceKey', - None, - None, - None, + return bucket_dict + + +@pytest.fixture(scope="class") +def lock_enabled_bucket(persistent_bucket_factory): + return persistent_bucket_factory(is_file_lock_enabled=True) + + +@pytest.fixture(scope="class") +def upload_url_dict(raw_api, auth_info, lock_enabled_bucket): + upload_url_dict = raw_api.get_upload_url( + auth_info['apiUrl'], + auth_info['authorizationToken'], + lock_enabled_bucket.bucket_id, ) - replication_source_key = replication_source_key_dict['applicationKeyId'] + return upload_url_dict - # 2) create source bucket with replication to destination - existing bucket - try: - # in order to test replication, we need to create a second bucket + +@pytest.fixture(scope="class") +def part_contents_dict(): + part_contents = b'hello part' + yield { + 'part_contents': part_contents, + 'part_sha1': hex_sha1_of_stream(io.BytesIO(part_contents), len(part_contents)) + } + + +@pytest.fixture(scope="class") +def uploaded_file_dict(raw_api, lock_enabled_bucket, sse_b2_aes, upload_url_dict): + upload_url = upload_url_dict['uploadUrl'] + upload_auth_token = upload_url_dict['authorizationToken'] + # file_name = 'test.txt' + file_name = f'{lock_enabled_bucket.subfolder}/test.txt' + file_contents = b'hello world' + file_sha1 = hex_sha1_of_stream(io.BytesIO(file_contents), len(file_contents)) + uploaded_file_dict = raw_api.upload_file( + upload_url, + upload_auth_token, + file_name, + len(file_contents), + 'text/plain', + file_sha1, { + 'color': 'blue', + 'b2-cache-control': 'private, max-age=2222' + }, + io.BytesIO(file_contents), + server_side_encryption=sse_b2_aes, + file_retention=FileRetentionSetting( + RetentionMode.GOVERNANCE, + int(time.time() + 100) * 1000, + ) + ) + uploaded_file_dict['file_contents'] = file_contents + return uploaded_file_dict + + +@pytest.fixture(scope="class") +def download_auth_token(raw_api, auth_info, lock_enabled_bucket, uploaded_file_dict): + download_auth = raw_api.get_download_authorization( + auth_info['apiUrl'], auth_info['authorizationToken'], lock_enabled_bucket.bucket_id, + uploaded_file_dict['fileName'][:-2], 12345 + ) + yield download_auth['authorizationToken'] + + +class TestRawAPIBucketOps: + + raw_api: B2RawHTTPApi + auth_info: dict + test_bucket: dict + + @pytest.fixture(autouse=True, scope="class") + def setup(self, request, raw_api, auth_info, test_bucket): + cls = request.cls + cls.raw_api = raw_api + cls.auth_info = auth_info + cls.test_bucket = test_bucket + + @pytest.fixture(scope='class') + def replication_source_key(self): + replication_source_key_dict = self.raw_api.create_key( + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.auth_info['accountId'], + [ + 'listBuckets', + 'listFiles', + 'readFiles', + 'writeFiles', # Pawel @ 2022-06-21: adding this to make tests pass with a weird server validator + ], + 'testReplicationSourceKey', + None, + None, + None, + ) + assert 'applicationKeyId' in replication_source_key_dict + replication_source_key = replication_source_key_dict['applicationKeyId'] + yield replication_source_key + + self.raw_api.delete_key( + self.auth_info['apiUrl'], self.auth_info['authorizationToken'], replication_source_key + ) + + @pytest.fixture(scope='class') + def replication_source_bucket_dict(self, replication_source_key): replication_source_bucket_name = 'test-raw-api-%s-%d-%d' % ( - account_id, int(time.time()), random.randint(1000, 9999) + self.auth_info['accountId'], int(time.time()), random.randint(1000, 9999) ) - replication_source_bucket_dict = raw_api.create_bucket( - api_url, - account_auth_token, - account_id, + replication_source_bucket_dict = self.raw_api.create_bucket( + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.auth_info['accountId'], replication_source_bucket_name, 'allPublic', is_file_lock_enabled=True, + lifecycle_rules=[ + { + "fileNamePrefix": "", + "daysFromHidingToDeleting": 1, + "daysFromUploadingToHiding": 1 + } + ], replication=ReplicationConfiguration( rules=[ ReplicationRule( - destination_bucket_id=bucket_id, + destination_bucket_id=self.test_bucket['bucketId'], include_existing_files=True, name='test-rule', ), @@ -217,6 +220,103 @@ def raw_api_test_helper(raw_api, should_cleanup_old_buckets): source_key_id=replication_source_key, ), ) + yield replication_source_bucket_dict + + _clean_and_delete_bucket( + self.raw_api, + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.auth_info['accountId'], + replication_source_bucket_dict['bucketId'], + ) + + @pytest.fixture(scope='class') + def replication_destination_key(self): + replication_destination_key_dict = self.raw_api.create_key( + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.auth_info['accountId'], + ['listBuckets', 'listFiles', 'writeFiles'], + 'testReplicationDestinationKey', + None, + None, + None, + ) + + replication_destination_key = replication_destination_key_dict['applicationKeyId'] + yield replication_destination_key + + self.raw_api.delete_key( + self.auth_info['apiUrl'], self.auth_info['authorizationToken'], + replication_destination_key + ) + + def test_update_bucket_with_encryption(self, sse_b2_aes): + sse_none = EncryptionSetting(mode=EncryptionMode.NONE) + test_cases = [ + ( + sse_none, + BucketRetentionSetting( + mode=RetentionMode.GOVERNANCE, period=RetentionPeriod(days=1) + ) + ), + (sse_b2_aes, BucketRetentionSetting(RetentionMode.NONE)), + (sse_b2_aes, BucketRetentionSetting(RetentionMode.NONE)), + ] + + for encryption_setting, default_retention in test_cases: + self.raw_api.update_bucket( + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.auth_info['accountId'], + self.test_bucket['bucketId'], + 'allPublic', + default_server_side_encryption=encryption_setting, + default_retention=default_retention, + ) + + def test_disable_file_lock(self): + with pytest.raises(DisablingFileLockNotSupported): + self.raw_api.update_bucket( + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.auth_info['accountId'], + self.test_bucket['bucketId'], + 'allPrivate', + is_file_lock_enabled=False, + ) + + def test_authorize_account(self): + preview_feature_caps = { + 'readBucketNotifications', + 'writeBucketNotifications', + } + missing_capabilities = ( + set(ALL_CAPABILITIES) - {'readBuckets', 'listAllBucketNames'} - preview_feature_caps - + set(self.auth_info['allowed']['capabilities']) + ) + assert not missing_capabilities, f'it appears that the raw_api integration test is being run with a non-full key. Missing capabilities: {missing_capabilities}' + + def test_create_list_delete_key(self): + account_id = self.auth_info['accountId'] + account_auth_token = self.auth_info['authorizationToken'] + api_url = self.auth_info['apiUrl'] + key_dict = self.raw_api.create_key( + api_url, + account_auth_token, + account_id, + ['readFiles'], + 'testKey', + None, + None, + None, + ) + self.raw_api.list_keys(api_url, account_auth_token, account_id, 10) + self.raw_api.delete_key(api_url, account_auth_token, key_dict['applicationKeyId']) + + def test_create_bucket_with_replication( + self, replication_source_key, replication_source_bucket_dict + ): assert 'replicationConfiguration' in replication_source_bucket_dict assert replication_source_bucket_dict['replicationConfiguration'] == { 'isClientAuthorizedToRead': True, @@ -227,7 +327,7 @@ def raw_api_test_helper(raw_api, should_cleanup_old_buckets): "replicationRules": [ { - "destinationBucketId": bucket_id, + "destinationBucketId": self.test_bucket['bucketId'], "fileNamePrefix": "", "includeExistingFiles": True, "isEnabled": True, @@ -241,14 +341,13 @@ def raw_api_test_helper(raw_api, should_cleanup_old_buckets): }, } - # 3) upload test file and check replication status - upload_url_dict = raw_api.get_upload_url( - api_url, - account_auth_token, + upload_url_dict = self.raw_api.get_upload_url( + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], replication_source_bucket_dict['bucketId'], ) file_contents = b'hello world' - file_dict = raw_api.upload_file( + file_dict = self.raw_api.upload_file( upload_url_dict['uploadUrl'], upload_url_dict['authorizationToken'], 'test.txt', @@ -258,33 +357,17 @@ def raw_api_test_helper(raw_api, should_cleanup_old_buckets): {'color': 'blue'}, io.BytesIO(file_contents), ) - assert ReplicationStatus[file_dict['replicationStatus'].upper() ] == ReplicationStatus.PENDING - finally: - raw_api.delete_key(api_url, account_auth_token, replication_source_key) - - # 4) create destination key (write permissions) - replication_destination_key_dict = raw_api.create_key( - api_url, - account_auth_token, - account_id, - ['listBuckets', 'listFiles', 'writeFiles'], - 'testReplicationDestinationKey', - None, - None, - None, - ) - replication_destination_key = replication_destination_key_dict['applicationKeyId'] - - # 5) update destination bucket to receive updates - try: - bucket_dict = raw_api.update_bucket( - api_url, - account_auth_token, - account_id, - bucket_id, + def test_update_bucket_to_receive_updates( + self, replication_destination_key, replication_source_key + ): + bucket_dict = self.raw_api.update_bucket( + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.auth_info['accountId'], + self.test_bucket['bucketId'], 'allPublic', replication=ReplicationConfiguration( source_to_destination_key_mapping={ @@ -306,300 +389,296 @@ def raw_api_test_helper(raw_api, should_cleanup_old_buckets): 'asReplicationSource': None, }, } - finally: - raw_api.delete_key( - api_url, - account_auth_token, - replication_destination_key_dict['applicationKeyId'], + + def test_disable_replication(self): + bucket_dict = self.raw_api.update_bucket( + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.auth_info['accountId'], + self.test_bucket['bucketId'], + 'allPublic', + replication=ReplicationConfiguration(), ) + assert bucket_dict['replicationConfiguration'] == { + 'isClientAuthorizedToRead': True, + 'value': None, + } - # 6) cleanup: disable replication for destination and remove source - bucket_dict = raw_api.update_bucket( - api_url, - account_auth_token, - account_id, - bucket_id, - 'allPublic', - replication=ReplicationConfiguration(), - ) - assert bucket_dict['replicationConfiguration'] == { - 'isClientAuthorizedToRead': True, - 'value': None, - } - _clean_and_delete_bucket( - raw_api, - api_url, - account_auth_token, - account_id, - replication_source_bucket_dict['bucketId'], - ) +class TestRawAPIFileOps: + + raw_api: B2RawHTTPApi + auth_info: dict + lock_enabled_bucket: PersistentBucketAggregate + sse_b2_aes: EncryptionSetting + upload_url_dict: dict + + uploaded_file_dict: dict + download_auth_token: str + + @pytest.fixture(autouse=True, scope="class") + def setup( + self, request, raw_api, auth_info, lock_enabled_bucket, sse_b2_aes, uploaded_file_dict, + download_auth_token + ): + cls = request.cls + cls.raw_api = raw_api + cls.auth_info = auth_info + cls.lock_enabled_bucket = lock_enabled_bucket + cls.sse_b2_aes = sse_b2_aes + cls.uploaded_file_dict = uploaded_file_dict + cls.download_auth_token = download_auth_token + + @pytest.fixture(scope="class") + def large_file(self): + unique_subfolder = self.lock_enabled_bucket.new_subfolder() + file_info = {'color': 'red'} + large_info = self.raw_api.start_large_file( + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.lock_enabled_bucket.bucket_id, + f'{unique_subfolder}/large_file.txt', + 'text/plain', + file_info, + ) + return large_info - ################# - print('b2_update_bucket') - sse_b2_aes = EncryptionSetting( - mode=EncryptionMode.SSE_B2, - algorithm=EncryptionAlgorithm.AES256, - ) - sse_none = EncryptionSetting(mode=EncryptionMode.NONE) - for encryption_setting, default_retention in [ - ( - sse_none, - BucketRetentionSetting(mode=RetentionMode.GOVERNANCE, period=RetentionPeriod(days=1)) - ), - (sse_b2_aes, None), - (sse_b2_aes, BucketRetentionSetting(RetentionMode.NONE)), - ]: - bucket_dict = raw_api.update_bucket( - api_url, - account_auth_token, - account_id, - bucket_id, - 'allPublic', - default_server_side_encryption=encryption_setting, - default_retention=default_retention, + @pytest.fixture(scope="class") + def upload_part_dict(self, large_file): + yield self.raw_api.get_upload_part_url( + self.auth_info['apiUrl'], self.auth_info['authorizationToken'], large_file['fileId'] ) - # b2_list_buckets - print('b2_list_buckets') - bucket_list_dict = raw_api.list_buckets(api_url, account_auth_token, account_id) - #print(bucket_list_dict) + def test_list_bucket(self): + self.raw_api.list_buckets( + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.auth_info['accountId'], + self.lock_enabled_bucket.bucket_id, + ) - # b2_get_upload_url - print('b2_get_upload_url') - upload_url_dict = raw_api.get_upload_url(api_url, account_auth_token, bucket_id) - upload_url = upload_url_dict['uploadUrl'] - upload_auth_token = upload_url_dict['authorizationToken'] + def test_list_file_versions(self): + file_name = self.uploaded_file_dict['fileName'] + list_versions_dict = self.raw_api.list_file_versions( + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.lock_enabled_bucket.bucket_id, + ) + assert [file_name] == [f_dict['fileName'] for f_dict in list_versions_dict['files']] + assert ['private, max-age=2222'] == [ + f_dict['fileInfo']['b2-cache-control'] for f_dict in list_versions_dict['files'] + ] + + def test_download_file_by_id_auth(self): + url = self.raw_api.get_download_url_by_id( + self.auth_info['downloadUrl'], self.uploaded_file_dict['fileId'] + ) + account_auth_token = self.auth_info['authorizationToken'] + file_contents = self.uploaded_file_dict['file_contents'] + with self.raw_api.download_file_from_url(account_auth_token, url) as response: + data = next(response.iter_content(chunk_size=len(file_contents))) + assert data == file_contents, data + + def test_download_file_by_id_no_auth(self): + url = self.raw_api.get_download_url_by_name( + self.auth_info['downloadUrl'], self.lock_enabled_bucket.bucket_name, + self.uploaded_file_dict['fileName'] + ) + with self.raw_api.download_file_from_url( + self.auth_info['authorizationToken'], url + ) as response: + data = next( + response.iter_content(chunk_size=len(self.uploaded_file_dict['file_contents'])) + ) + assert data == self.uploaded_file_dict['file_contents'], data - # b2_upload_file - print('b2_upload_file') - file_name = 'test.txt' - file_contents = b'hello world' - file_sha1 = hex_sha1_of_stream(io.BytesIO(file_contents), len(file_contents)) - file_dict = raw_api.upload_file( - upload_url, - upload_auth_token, - file_name, - len(file_contents), - 'text/plain', - file_sha1, - { - 'color': 'blue', - 'b2-cache-control': 'private, max-age=2222' - }, - io.BytesIO(file_contents), - server_side_encryption=sse_b2_aes, - #custom_upload_timestamp=12345, - file_retention=FileRetentionSetting( - RetentionMode.GOVERNANCE, - int(time.time() + 100) * 1000, + def test_download_file_by_name_name(self): + url = self.raw_api.get_download_url_by_name( + self.auth_info['downloadUrl'], self.lock_enabled_bucket.bucket_name, + self.uploaded_file_dict['fileName'] ) - ) + with self.raw_api.download_file_from_url(None, url) as response: + data = next( + response.iter_content(chunk_size=len(self.uploaded_file_dict['file_contents'])) + ) + assert data == self.uploaded_file_dict['file_contents'], data - file_id = file_dict['fileId'] + def test_download_file_by_name_auth(self): + url = self.raw_api.get_download_url_by_name( + self.auth_info['downloadUrl'], self.lock_enabled_bucket.bucket_name, + self.uploaded_file_dict['fileName'] + ) + with self.raw_api.download_file_from_url(self.download_auth_token, url) as response: + data = next( + response.iter_content(chunk_size=len(self.uploaded_file_dict['file_contents'])) + ) + assert data == self.uploaded_file_dict['file_contents'], data - # b2_list_file_versions - print('b2_list_file_versions') - list_versions_dict = raw_api.list_file_versions(api_url, account_auth_token, bucket_id) - assert [file_name] == [f_dict['fileName'] for f_dict in list_versions_dict['files']] - assert ['private, max-age=2222'] == [ - f_dict['fileInfo']['b2-cache-control'] for f_dict in list_versions_dict['files'] - ] + def test_list_file_names(self): + url = self.raw_api.get_download_url_by_name( + self.auth_info['downloadUrl'], self.lock_enabled_bucket.bucket_name, + self.uploaded_file_dict['fileName'] + ) + with self.raw_api.download_file_from_url(self.download_auth_token, url) as response: + data = next( + response.iter_content(chunk_size=len(self.uploaded_file_dict['file_contents'])) + ) + assert data == self.uploaded_file_dict['file_contents'], data + + def test_list_file_names_start_count(self): + list_names_dict = self.raw_api.list_file_names( + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.lock_enabled_bucket.bucket_id, + start_file_name=self.uploaded_file_dict['fileName'], + max_file_count=5 + ) + assert [self.uploaded_file_dict['fileName']] == [ + f_dict['fileName'] for f_dict in list_names_dict['files'] + ] + + def test_copy_file(self): + copy_file_name = f'{self.lock_enabled_bucket.subfolder}/test_copy.txt' + self.raw_api.copy_file( + self.auth_info['apiUrl'], self.auth_info['authorizationToken'], + self.uploaded_file_dict['fileId'], copy_file_name + ) - # b2_download_file_by_id with auth - print('b2_download_file_by_id (auth)') - url = raw_api.get_download_url_by_id(download_url, file_id) - with raw_api.download_file_from_url(account_auth_token, url) as response: - data = next(response.iter_content(chunk_size=len(file_contents))) - assert data == file_contents, data - - # b2_download_file_by_id no auth - print('b2_download_file_by_id (no auth)') - url = raw_api.get_download_url_by_id(download_url, file_id) - with raw_api.download_file_from_url(None, url) as response: - data = next(response.iter_content(chunk_size=len(file_contents))) - assert data == file_contents, data - - # b2_download_file_by_name with auth - print('b2_download_file_by_name (auth)') - url = raw_api.get_download_url_by_name(download_url, bucket_name, file_name) - with raw_api.download_file_from_url(account_auth_token, url) as response: - data = next(response.iter_content(chunk_size=len(file_contents))) - assert data == file_contents, data - - # b2_download_file_by_name no auth - print('b2_download_file_by_name (no auth)') - url = raw_api.get_download_url_by_name(download_url, bucket_name, file_name) - with raw_api.download_file_from_url(None, url) as response: - data = next(response.iter_content(chunk_size=len(file_contents))) - assert data == file_contents, data - - # b2_get_download_authorization - print('b2_get_download_authorization') - download_auth = raw_api.get_download_authorization( - api_url, account_auth_token, bucket_id, file_name[:-2], 12345 - ) - download_auth_token = download_auth['authorizationToken'] - - # b2_download_file_by_name with download auth - print('b2_download_file_by_name (download auth)') - url = raw_api.get_download_url_by_name(download_url, bucket_name, file_name) - with raw_api.download_file_from_url(download_auth_token, url) as response: - data = next(response.iter_content(chunk_size=len(file_contents))) - assert data == file_contents, data - - # b2_list_file_names - print('b2_list_file_names') - list_names_dict = raw_api.list_file_names(api_url, account_auth_token, bucket_id) - assert [file_name] == [f_dict['fileName'] for f_dict in list_names_dict['files']] - - # b2_list_file_names (start, count) - print('b2_list_file_names (start, count)') - list_names_dict = raw_api.list_file_names( - api_url, account_auth_token, bucket_id, start_file_name=file_name, max_file_count=5 - ) - assert [file_name] == [f_dict['fileName'] for f_dict in list_names_dict['files']] - - # b2_copy_file - print('b2_copy_file') - copy_file_name = 'test_copy.txt' - raw_api.copy_file(api_url, account_auth_token, file_id, copy_file_name) - - # b2_get_file_info_by_id - print('b2_get_file_info_by_id') - file_info_dict = raw_api.get_file_info_by_id(api_url, account_auth_token, file_id) - assert file_info_dict['fileName'] == file_name - - # b2_get_file_info_by_name - print('b2_get_file_info_by_name (no auth)') - info_headers = raw_api.get_file_info_by_name(download_url, None, bucket_name, file_name) - assert info_headers['x-bz-file-id'] == file_id - - # b2_get_file_info_by_name - print('b2_get_file_info_by_name (auth)') - info_headers = raw_api.get_file_info_by_name( - download_url, account_auth_token, bucket_name, file_name - ) - assert info_headers['x-bz-file-id'] == file_id + def test_get_file_info_by_id(self): + file_info_dict = self.raw_api.get_file_info_by_id( + self.auth_info['apiUrl'], self.auth_info['authorizationToken'], + self.uploaded_file_dict['fileId'] + ) + assert file_info_dict['fileName'] == self.uploaded_file_dict['fileName'] - # b2_get_file_info_by_name - print('b2_get_file_info_by_name (download auth)') - info_headers = raw_api.get_file_info_by_name( - download_url, download_auth_token, bucket_name, file_name - ) - assert info_headers['x-bz-file-id'] == file_id - - # b2_hide_file - print('b2_hide_file') - raw_api.hide_file(api_url, account_auth_token, bucket_id, file_name) - - # b2_start_large_file - print('b2_start_large_file') - file_info = {'color': 'red'} - large_info = raw_api.start_large_file( - api_url, - account_auth_token, - bucket_id, - file_name, - 'text/plain', - file_info, - server_side_encryption=sse_b2_aes, - ) - large_file_id = large_info['fileId'] + def test_get_file_info_by_name_no_auth(self): + file_info_dict = self.raw_api.get_file_info_by_name( + self.auth_info['downloadUrl'], None, self.lock_enabled_bucket.bucket_name, + self.uploaded_file_dict['fileName'] + ) + assert file_info_dict['x-bz-file-id'] == self.uploaded_file_dict['fileId'] - # b2_get_upload_part_url - print('b2_get_upload_part_url') - upload_part_dict = raw_api.get_upload_part_url(api_url, account_auth_token, large_file_id) - upload_part_url = upload_part_dict['uploadUrl'] - upload_path_auth = upload_part_dict['authorizationToken'] + def test_get_file_info_by_name_auth(self): + file_info_dict = self.raw_api.get_file_info_by_name( + self.auth_info['downloadUrl'], self.auth_info['authorizationToken'], + self.lock_enabled_bucket.bucket_name, self.uploaded_file_dict['fileName'] + ) + assert file_info_dict['x-bz-file-id'] == self.uploaded_file_dict['fileId'] - # b2_upload_part - print('b2_upload_part') - part_contents = b'hello part' - part_sha1 = hex_sha1_of_stream(io.BytesIO(part_contents), len(part_contents)) - raw_api.upload_part( - upload_part_url, upload_path_auth, 1, len(part_contents), part_sha1, - io.BytesIO(part_contents) - ) + def test_get_file_info_by_name_download_auth(self): + file_info_dict = self.raw_api.get_file_info_by_name( + self.auth_info['downloadUrl'], self.download_auth_token, + self.lock_enabled_bucket.bucket_name, self.uploaded_file_dict['fileName'] + ) + assert file_info_dict['x-bz-file-id'] == self.uploaded_file_dict['fileId'] - # b2_copy_part - print('b2_copy_part') - raw_api.copy_part(api_url, account_auth_token, file_id, large_file_id, 2, (0, 5)) + def test_hide_file(self): + self.raw_api.hide_file( + self.auth_info['apiUrl'], self.auth_info['authorizationToken'], + self.lock_enabled_bucket.bucket_id, self.uploaded_file_dict['fileName'] + ) - # b2_list_parts - print('b2_list_parts') - parts_response = raw_api.list_parts(api_url, account_auth_token, large_file_id, 1, 100) - assert [1, 2] == [part['partNumber'] for part in parts_response['parts']] + def test_upload_part(self, upload_part_dict, part_contents_dict): + upload_part_url = upload_part_dict['uploadUrl'] + upload_path_auth = upload_part_dict['authorizationToken'] + part_contents = part_contents_dict['part_contents'] + part_sha1 = part_contents_dict['part_sha1'] + self.raw_api.upload_part( + upload_part_url, upload_path_auth, 1, len(part_contents), part_sha1, + io.BytesIO(part_contents) + ) - # b2_list_unfinished_large_files - unfinished_list = raw_api.list_unfinished_large_files(api_url, account_auth_token, bucket_id) - assert [file_name] == [f_dict['fileName'] for f_dict in unfinished_list['files']] - assert file_info == unfinished_list['files'][0]['fileInfo'] + def test_copy_part(self, large_file): + self.raw_api.copy_part( + self.auth_info['apiUrl'], self.auth_info['authorizationToken'], + self.uploaded_file_dict['fileId'], large_file['fileId'], 2, (0, 5) + ) - # b2_finish_large_file - print('b2_finish_large_file') - try: - raw_api.finish_large_file(api_url, account_auth_token, large_file_id, [part_sha1]) - raise Exception('finish should have failed') - except Exception as e: - assert 'large files must have at least 2 parts' in str(e) - # TODO: make another attempt to finish but this time successfully - - # b2_update_bucket - print('b2_update_bucket') - updated_bucket = raw_api.update_bucket( - api_url, - account_auth_token, - account_id, - bucket_id, - 'allPrivate', - bucket_info={'color': 'blue'}, - default_retention=BucketRetentionSetting( - mode=RetentionMode.GOVERNANCE, period=RetentionPeriod(days=1) - ), - is_file_lock_enabled=True, - ) - assert first_bucket_revision < updated_bucket['revision'] - - # NOTE: this update_bucket call is only here to be able to find out the error code returned by - # the server if an attempt is made to disable file lock. It has to be done here since the CLI - # by design does not allow disabling file lock at all (i.e. there is no --fileLockEnabled=false - # option or anything equivalent to that). - with pytest.raises(DisablingFileLockNotSupported): - raw_api.update_bucket( - api_url, - account_auth_token, - account_id, - bucket_id, - 'allPrivate', - is_file_lock_enabled=False, + def test_list_parts(self, large_file): + parts_response = self.raw_api.list_parts( + self.auth_info['apiUrl'], self.auth_info['authorizationToken'], large_file['fileId'], 1, + 100 ) + assert [1, 2] == [part['partNumber'] for part in parts_response['parts']] - # b2_delete_file_version - print('b2_delete_file_version') + def test_list_unfinished_large_files(self, large_file): + unfinished_list = self.raw_api.list_unfinished_large_files( + self.auth_info['apiUrl'], self.auth_info['authorizationToken'], + self.lock_enabled_bucket.bucket_id + ) + assert [large_file['fileName']] == [ + f_dict['fileName'] for f_dict in unfinished_list['files'] + ] + assert large_file['fileInfo'] == unfinished_list['files'][0]['fileInfo'] + + def test_finish_large_file_too_few_parts(self, large_file, part_contents_dict): + try: + self.raw_api.finish_large_file( + self.auth_info['apiUrl'], self.auth_info['authorizationToken'], + large_file['fileId'], [part_contents_dict['part_sha1']] + ) + pytest.fail('finish should have failed') + except Exception as e: + assert 'large files must have at least 2 parts' in str(e) + + def test_finish_large_file_success(self, large_file, upload_part_dict): + upload_part_url = upload_part_dict['uploadUrl'] + upload_path_auth = upload_part_dict['authorizationToken'] + + # Create two parts, each at least 5 MB in size + part_size = 5 * 1024 * 1024 # 5 MB + part1_contents = b'0' * part_size + part2_contents = b'1' * part_size + + part1_sha1 = hex_sha1_of_stream(io.BytesIO(part1_contents), len(part1_contents)) + part2_sha1 = hex_sha1_of_stream(io.BytesIO(part2_contents), len(part2_contents)) + + # Upload the first part + self.raw_api.upload_part( + upload_part_url, upload_path_auth, 1, len(part1_contents), part1_sha1, + io.BytesIO(part1_contents) + ) - with pytest.raises(Unauthorized): - raw_api.delete_file_version(api_url, account_auth_token, file_id, file_name) - raw_api.delete_file_version(api_url, account_auth_token, file_id, file_name, True) + # Upload the second part + self.raw_api.upload_part( + upload_part_url, upload_path_auth, 2, len(part2_contents), part2_sha1, + io.BytesIO(part2_contents) + ) - print('b2_get_bucket_notification_rules & b2_set_bucket_notification_rules') - try: - _subtest_bucket_notification_rules( - raw_api, auth_dict, api_url, account_auth_token, bucket_id + self.raw_api.finish_large_file( + self.auth_info['apiUrl'], self.auth_info['authorizationToken'], large_file['fileId'], + [part1_sha1, part2_sha1] ) - except pytest.skip.Exception as e: - print(e) - # Clean up this test. - _clean_and_delete_bucket(raw_api, api_url, account_auth_token, account_id, bucket_id) + def test_list_finished_large_files(self, large_file): + finished_file = self.raw_api.list_file_names( + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.lock_enabled_bucket.bucket_id, + prefix=large_file['fileName'][:6] + ) + assert [large_file['fileName']] == [f_dict['fileName'] for f_dict in finished_file['files']] + assert large_file['fileInfo'] == finished_file['files'][0]['fileInfo'] + + def test_unauthorized_delete_file_version(self): + with pytest.raises(Unauthorized): + self.raw_api.delete_file_version( + self.auth_info['apiUrl'], self.auth_info['authorizationToken'], + self.uploaded_file_dict['fileId'], self.uploaded_file_dict['fileName'] + ) + + def test_delete_file_version_with_auth(self): + self.raw_api.delete_file_version( + self.auth_info['apiUrl'], self.auth_info['authorizationToken'], + self.uploaded_file_dict['fileId'], self.uploaded_file_dict['fileName'], True + ) - if should_cleanup_old_buckets: - # Clean up from old tests. Empty and delete any buckets more than an hour old. - _cleanup_old_buckets(raw_api, auth_dict, bucket_list_dict) +def _subtest_bucket_notification_rules(raw_api, auth_info, bucket_id): + account_auth_token = auth_info['authorizationToken'] + api_url = auth_info['apiUrl'] -def _subtest_bucket_notification_rules(raw_api, auth_dict, api_url, account_auth_token, bucket_id): - if 'writeBucketNotifications' not in auth_dict['allowed']['capabilities']: + if 'writeBucketNotifications' not in auth_info['allowed']['capabilities']: pytest.skip('Test account does not have writeBucketNotifications capability') notification_rule = { @@ -640,68 +719,8 @@ def _subtest_bucket_notification_rules(raw_api, auth_dict, api_url, account_auth assert raw_api.get_bucket_notification_rules(api_url, account_auth_token, bucket_id) == [] -def cleanup_old_buckets(): - raw_api = B2RawHTTPApi(B2Http()) - auth_dict = authorize_raw_api(raw_api) - bucket_list_dict = raw_api.list_buckets( - auth_dict['apiUrl'], auth_dict['authorizationToken'], auth_dict['accountId'] - ) - _cleanup_old_buckets(raw_api, auth_dict, bucket_list_dict) - - -def _cleanup_old_buckets(raw_api, auth_dict, bucket_list_dict): - for bucket_dict in bucket_list_dict['buckets']: - bucket_id = bucket_dict['bucketId'] - bucket_name = bucket_dict['bucketName'] - if _should_delete_bucket(bucket_name): - print('cleaning up old bucket: ' + bucket_name) - _clean_and_delete_bucket( - raw_api, - auth_dict['apiUrl'], - auth_dict['authorizationToken'], - auth_dict['accountId'], - bucket_id, - ) - - -def _clean_and_delete_bucket(raw_api, api_url, account_auth_token, account_id, bucket_id): - # Delete the files. This test never creates more than a few files, - # so one call to list_file_versions should get them all. - versions_dict = raw_api.list_file_versions(api_url, account_auth_token, bucket_id) - for version_dict in versions_dict['files']: - file_id = version_dict['fileId'] - file_name = version_dict['fileName'] - action = version_dict['action'] - if action in ['hide', 'upload']: - print('b2_delete_file', file_name, action) - if action == 'upload' and version_dict[ - 'fileRetention'] and version_dict['fileRetention']['value']['mode'] is not None: - raw_api.update_file_retention( - api_url, - account_auth_token, - file_id, - file_name, - NO_RETENTION_FILE_SETTING, - bypass_governance=True - ) - raw_api.delete_file_version(api_url, account_auth_token, file_id, file_name) - else: - print('b2_cancel_large_file', file_name) - raw_api.cancel_large_file(api_url, account_auth_token, file_id) - - # Delete the bucket - print('b2_delete_bucket', bucket_id) - raw_api.delete_bucket(api_url, account_auth_token, account_id, bucket_id) - - -def _should_delete_bucket(bucket_name): - # Bucket names for this test look like: c7b22d0b0ad7-1460060364-5670 - # Other buckets should not be deleted. - match = re.match(r'^test-raw-api-[a-f0-9]+-([0-9]+)-([0-9]+)', bucket_name) - if match is None: - return False - - # Is it more than an hour old? - bucket_time = int(match.group(1)) - now = time.time() - return bucket_time + 3600 <= now +def test_get_and_set_bucket_notification_rules(raw_api, auth_info, test_bucket): + try: + _subtest_bucket_notification_rules(raw_api, auth_info, test_bucket['bucketId']) + except pytest.skip.Exception as e: + print(e) From 560a58a9402917e557f966c26eeeea55b9632876 Mon Sep 17 00:00:00 2001 From: kris-konina-reef Date: Mon, 23 Sep 2024 12:31:59 +0000 Subject: [PATCH 5/7] Add changelog --- changelog.d/+test_with_persistent_bucket.infrastructure.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog.d/+test_with_persistent_bucket.infrastructure.md diff --git a/changelog.d/+test_with_persistent_bucket.infrastructure.md b/changelog.d/+test_with_persistent_bucket.infrastructure.md new file mode 100644 index 00000000..39419d1b --- /dev/null +++ b/changelog.d/+test_with_persistent_bucket.infrastructure.md @@ -0,0 +1 @@ +Improve internal testing infrastructure by updating integration tests to use persistent buckets. \ No newline at end of file From 749aa47f10c04c452c7481717638597701b35d30 Mon Sep 17 00:00:00 2001 From: kris-konina-reef Date: Tue, 24 Sep 2024 17:56:03 +0000 Subject: [PATCH 6/7] Improve fixture deps; further isolate persistent bucket bound tests --- test/integration/conftest.py | 8 +-- test/integration/test_raw_api.py | 100 ++++++++++++++++--------------- 2 files changed, 55 insertions(+), 53 deletions(-) diff --git a/test/integration/conftest.py b/test/integration/conftest.py index 0dcdfdb8..5bc99974 100644 --- a/test/integration/conftest.py +++ b/test/integration/conftest.py @@ -116,12 +116,8 @@ def raw_api(): @pytest.fixture(scope="class") -def auth_info(raw_api): - application_key_id = os.environ.get('B2_TEST_APPLICATION_KEY_ID') - application_key = os.environ.get('B2_TEST_APPLICATION_KEY') - if application_key_id is None or application_key is None: - pytest.fail('B2_TEST_APPLICATION_KEY_ID or B2_TEST_APPLICATION_KEY is not set.') - +def auth_info(b2_auth_data, raw_api): + application_key_id, application_key = b2_auth_data realm = os.environ.get('B2_TEST_ENVIRONMENT', 'production') realm_url = REALM_URLS.get(realm, realm) return raw_api.authorize_account(realm_url, application_key_id, application_key) diff --git a/test/integration/test_raw_api.py b/test/integration/test_raw_api.py index c394b9fa..97501585 100644 --- a/test/integration/test_raw_api.py +++ b/test/integration/test_raw_api.py @@ -10,7 +10,6 @@ from __future__ import annotations import io -import os import random import time from test.helpers import type_validator_factory @@ -35,7 +34,6 @@ ) from b2sdk._internal.raw_api import ( ALL_CAPABILITIES, - REALM_URLS, B2RawHTTPApi, NotificationRuleResponse, ) @@ -49,18 +47,6 @@ def raw_api(): return B2RawHTTPApi(B2Http()) -@pytest.fixture(scope="class") -def auth_info(raw_api): - application_key_id = os.environ.get('B2_TEST_APPLICATION_KEY_ID') - application_key = os.environ.get('B2_TEST_APPLICATION_KEY') - if application_key_id is None or application_key is None: - pytest.fail('B2_TEST_APPLICATION_KEY_ID or B2_TEST_APPLICATION_KEY is not set.') - - realm = os.environ.get('B2_TEST_ENVIRONMENT', 'production') - realm_url = REALM_URLS.get(realm, realm) - return raw_api.authorize_account(realm_url, application_key_id, application_key) - - @pytest.fixture(scope="session") def sse_b2_aes(): return EncryptionSetting( @@ -115,32 +101,41 @@ def part_contents_dict(): @pytest.fixture(scope="class") -def uploaded_file_dict(raw_api, lock_enabled_bucket, sse_b2_aes, upload_url_dict): - upload_url = upload_url_dict['uploadUrl'] - upload_auth_token = upload_url_dict['authorizationToken'] - # file_name = 'test.txt' - file_name = f'{lock_enabled_bucket.subfolder}/test.txt' - file_contents = b'hello world' - file_sha1 = hex_sha1_of_stream(io.BytesIO(file_contents), len(file_contents)) - uploaded_file_dict = raw_api.upload_file( - upload_url, - upload_auth_token, - file_name, - len(file_contents), - 'text/plain', - file_sha1, { - 'color': 'blue', - 'b2-cache-control': 'private, max-age=2222' - }, - io.BytesIO(file_contents), - server_side_encryption=sse_b2_aes, - file_retention=FileRetentionSetting( - RetentionMode.GOVERNANCE, - int(time.time() + 100) * 1000, +def upload_file_dict_factory(raw_api, lock_enabled_bucket, sse_b2_aes, upload_url_dict): + def _upload_file_dict_factory(): + upload_url = upload_url_dict['uploadUrl'] + upload_auth_token = upload_url_dict['authorizationToken'] + subfolder = lock_enabled_bucket.new_subfolder() + file_name = f'{subfolder}/test.txt' + file_contents = b'hello world' + file_sha1 = hex_sha1_of_stream(io.BytesIO(file_contents), len(file_contents)) + uploaded_file_dict = raw_api.upload_file( + upload_url, + upload_auth_token, + file_name, + len(file_contents), + 'text/plain', + file_sha1, { + 'color': 'blue', + 'b2-cache-control': 'private, max-age=2222' + }, + io.BytesIO(file_contents), + server_side_encryption=sse_b2_aes, + file_retention=FileRetentionSetting( + RetentionMode.GOVERNANCE, + int(time.time() + 100) * 1000, + ) ) - ) - uploaded_file_dict['file_contents'] = file_contents - return uploaded_file_dict + uploaded_file_dict['file_contents'] = file_contents + uploaded_file_dict['subfolder'] = subfolder + return uploaded_file_dict + + return _upload_file_dict_factory + + +@pytest.fixture(scope="class") +def uploaded_file_dict(upload_file_dict_factory): + return upload_file_dict_factory() @pytest.fixture(scope="class") @@ -419,7 +414,7 @@ class TestRawAPIFileOps: @pytest.fixture(autouse=True, scope="class") def setup( self, request, raw_api, auth_info, lock_enabled_bucket, sse_b2_aes, uploaded_file_dict, - download_auth_token + upload_file_dict_factory, download_auth_token ): cls = request.cls cls.raw_api = raw_api @@ -427,6 +422,7 @@ def setup( cls.lock_enabled_bucket = lock_enabled_bucket cls.sse_b2_aes = sse_b2_aes cls.uploaded_file_dict = uploaded_file_dict + cls.single_file_dict = upload_file_dict_factory() cls.download_auth_token = download_auth_token @pytest.fixture(scope="class") @@ -441,6 +437,7 @@ def large_file(self): 'text/plain', file_info, ) + large_info['subfolder'] = unique_subfolder return large_info @pytest.fixture(scope="class") @@ -459,10 +456,12 @@ def test_list_bucket(self): def test_list_file_versions(self): file_name = self.uploaded_file_dict['fileName'] + subfolder = self.uploaded_file_dict['subfolder'] list_versions_dict = self.raw_api.list_file_versions( self.auth_info['apiUrl'], self.auth_info['authorizationToken'], self.lock_enabled_bucket.bucket_id, + prefix=subfolder ) assert [file_name] == [f_dict['fileName'] for f_dict in list_versions_dict['files']] assert ['private, max-age=2222'] == [ @@ -530,6 +529,7 @@ def test_list_file_names_start_count(self): self.auth_info['apiUrl'], self.auth_info['authorizationToken'], self.lock_enabled_bucket.bucket_id, + prefix=self.uploaded_file_dict['subfolder'], start_file_name=self.uploaded_file_dict['fileName'], max_file_count=5 ) @@ -603,8 +603,10 @@ def test_list_parts(self, large_file): def test_list_unfinished_large_files(self, large_file): unfinished_list = self.raw_api.list_unfinished_large_files( - self.auth_info['apiUrl'], self.auth_info['authorizationToken'], - self.lock_enabled_bucket.bucket_id + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + self.lock_enabled_bucket.bucket_id, + prefix=large_file['subfolder'] ) assert [large_file['fileName']] == [ f_dict['fileName'] for f_dict in unfinished_list['files'] @@ -614,8 +616,10 @@ def test_list_unfinished_large_files(self, large_file): def test_finish_large_file_too_few_parts(self, large_file, part_contents_dict): try: self.raw_api.finish_large_file( - self.auth_info['apiUrl'], self.auth_info['authorizationToken'], - large_file['fileId'], [part_contents_dict['part_sha1']] + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + large_file['fileId'], + [part_contents_dict['part_sha1']], ) pytest.fail('finish should have failed') except Exception as e: @@ -646,8 +650,10 @@ def test_finish_large_file_success(self, large_file, upload_part_dict): ) self.raw_api.finish_large_file( - self.auth_info['apiUrl'], self.auth_info['authorizationToken'], large_file['fileId'], - [part1_sha1, part2_sha1] + self.auth_info['apiUrl'], + self.auth_info['authorizationToken'], + large_file['fileId'], + [part1_sha1, part2_sha1], ) def test_list_finished_large_files(self, large_file): @@ -655,7 +661,7 @@ def test_list_finished_large_files(self, large_file): self.auth_info['apiUrl'], self.auth_info['authorizationToken'], self.lock_enabled_bucket.bucket_id, - prefix=large_file['fileName'][:6] + prefix=large_file['subfolder'] ) assert [large_file['fileName']] == [f_dict['fileName'] for f_dict in finished_file['files']] assert large_file['fileInfo'] == finished_file['files'][0]['fileInfo'] From 419318573fc04f2eb60755e96dc47e38d868a94b Mon Sep 17 00:00:00 2001 From: kris-konina-reef Date: Mon, 30 Sep 2024 16:58:54 +0200 Subject: [PATCH 7/7] Fix docstring --- test/integration/helpers.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/test/integration/helpers.py b/test/integration/helpers.py index 16b8dbd7..b8291db2 100644 --- a/test/integration/helpers.py +++ b/test/integration/helpers.py @@ -95,8 +95,11 @@ def _cleanup_old_buckets(raw_api, auth_dict, bucket_list_dict): def _clean_and_delete_bucket(raw_api, api_url, account_auth_token, account_id, bucket_id): - # Delete the files. This test never creates more than a few files, - # so one call to list_file_versions should get them all. + """ + Clean up and delete a bucket, including all its contents. + List and delete all file versions, handle retention settings, + and remove both regular and large files before deleting the bucket. + """ versions_dict = raw_api.list_file_versions(api_url, account_auth_token, bucket_id) for version_dict in versions_dict['files']: file_id = version_dict['fileId']