Integration tests tools public api

.gitignore

@@ -13,4 +13,5 @@ venv
 .venv
 .vscode
 .pdm-build/
-.pdm-python
+.pdm-python
+.env

b2sdk/_internal/testing/__init__.py

@@ -0,0 +1,9 @@
+######################################################################
+#
+# File: b2sdk/_internal/testing/__init__.py
+#
+# Copyright 2021 Backblaze Inc. All Rights Reserved.
+#
+# License https://www.backblaze.com/using_b2_code.html
+#
+######################################################################

b2sdk/_internal/testing/fixtures/__init__.py

@@ -0,0 +1,9 @@
+######################################################################
+#
+# File: b2sdk/_internal/testing/fixtures/__init__.py
+#
+# Copyright 2021 Backblaze Inc. All Rights Reserved.
+#
+# License https://www.backblaze.com/using_b2_code.html
+#
+######################################################################

b2sdk/_internal/testing/fixtures/api.py

@@ -0,0 +1,48 @@
+######################################################################
+#
+# File: b2sdk/_internal/testing/fixtures/api.py
+#
+# Copyright 2021 Backblaze Inc. All Rights Reserved.
+#
+# License https://www.backblaze.com/using_b2_code.html
+#
+######################################################################
+from __future__ import annotations
+
+import http
+import http.client
+import os
+
+import pytest
+
+from b2sdk._internal.testing.helpers.api import authorize, get_b2_auth_data, get_realm
+
+
+@pytest.fixture(scope='session')
+def realm():
+    yield get_realm()
+
+
+@pytest.fixture(autouse=True, scope='session')
+def set_http_debug():
+    if os.environ.get('B2_DEBUG_HTTP'):
+        http.client.HTTPConnection.debuglevel = 1
+
+
+@pytest.fixture(scope='session')
+def b2_auth_data():
+    try:
+        return get_b2_auth_data()
+    except ValueError as ex:
+        pytest.fail(ex.args[0])
+
+
+@pytest.fixture(scope='session')
+def _b2_api(b2_auth_data, realm):
+    b2_api, _ = authorize(b2_auth_data, realm)
+    return b2_api
+
+
+@pytest.fixture(scope='session')
+def b2_api(_b2_api, bucket_manager):
+    return _b2_api

b2sdk/_internal/testing/fixtures/buckets.py

@@ -0,0 +1,71 @@
+######################################################################
+#
+# File: b2sdk/_internal/testing/fixtures/buckets.py
+#
+# Copyright 2021 Backblaze Inc. All Rights Reserved.
+#
+# License https://www.backblaze.com/using_b2_code.html
+#
+######################################################################
+from __future__ import annotations
+
+import secrets
+
+import pytest
+
+from b2sdk._internal.testing.helpers.bucket_manager import BucketManager
+from b2sdk._internal.testing.helpers.buckets import (
+    GENERAL_BUCKET_NAME_PREFIX,
+    get_bucket_name_prefix,
+)
+
+
+def pytest_addoption(parser):
+    """Add a flag for not cleaning up old buckets"""
+    parser.addoption(
+        '--dont-cleanup-old-buckets',
+        action='store_true',
+        default=False,
+    )
+
+
+@pytest.fixture(scope='session')
+def dont_cleanup_old_buckets(request):
+    return request.config.getoption('--dont-cleanup-old-buckets')
+
+
+@pytest.fixture(scope='session')
+def bucket_name_prefix():
+    return get_bucket_name_prefix(8)
+
+
+@pytest.fixture(scope='session')
+def general_bucket_name_prefix():
+    return GENERAL_BUCKET_NAME_PREFIX
+
+
+@pytest.fixture(scope='session')
+def bucket_manager(
+    bucket_name_prefix, general_bucket_name_prefix, dont_cleanup_old_buckets, _b2_api
+):
+    manager = BucketManager(
+        dont_cleanup_old_buckets,
+        _b2_api,
+        current_run_prefix=bucket_name_prefix,
+        general_prefix=general_bucket_name_prefix,
+    )
+    yield manager
+    manager.clean_buckets()
+
+
+@pytest.fixture
+def bucket(bucket_name_prefix, bucket_manager):
+    bucket = bucket_manager.create_bucket()
+    yield bucket
+    bucket_manager.clean_bucket(bucket)
+
+
+@pytest.fixture
+def b2_subfolder(bucket, request):
+    subfolder_name = f'{request.node.name}_{secrets.token_urlsafe(4)}'
+    return f'b2://{bucket.name}/{subfolder_name}'

b2sdk/_internal/testing/helpers/__init__.py

@@ -0,0 +1,9 @@
+######################################################################
+#
+# File: b2sdk/_internal/testing/helpers/__init__.py
+#
+# Copyright 2021 Backblaze Inc. All Rights Reserved.
+#
+# License https://www.backblaze.com/using_b2_code.html
+#
+######################################################################

b2sdk/_internal/testing/helpers/api.py

@@ -0,0 +1,38 @@
+######################################################################
+#
+# File: b2sdk/_internal/testing/helpers/api.py
+#
+# Copyright 2022 Backblaze Inc. All Rights Reserved.
+#
+# License https://www.backblaze.com/using_b2_code.html
+#
+######################################################################
+from __future__ import annotations
+
+import os
+
+from b2sdk._internal.account_info.in_memory import InMemoryAccountInfo
+from b2sdk._internal.api import B2Api
+from b2sdk._internal.api_config import DEFAULT_HTTP_API_CONFIG
+
+
+def get_realm() -> str:
+    return os.environ.get('B2_TEST_ENVIRONMENT', 'production')
+
+
+def get_b2_auth_data():
+    application_key_id = os.environ.get('B2_TEST_APPLICATION_KEY_ID')
+    if application_key_id is None:
+        raise ValueError('B2_TEST_APPLICATION_KEY_ID is not set.')
+
+    application_key = os.environ.get('B2_TEST_APPLICATION_KEY')
+    if application_key is None:
+        raise ValueError('B2_TEST_APPLICATION_KEY is not set.')
+    return application_key_id, application_key
+
+
+def authorize(b2_auth_data, realm, api_config=DEFAULT_HTTP_API_CONFIG):
+    info = InMemoryAccountInfo()
+    b2_api = B2Api(info, api_config=api_config)
+    b2_api.authorize_account(*b2_auth_data, realm=realm)
+    return b2_api, info

test/integration/base.py → b2sdk/_internal/testing/helpers/base.py

@@ -1,6 +1,6 @@
 ######################################################################
 #
-# File: test/integration/base.py
+# File: b2sdk/_internal/testing/helpers/base.py
 #
 # Copyright 2022 Backblaze Inc. All Rights Reserved.
 #
@@ -11,27 +11,23 @@
 
 import pytest
 
-from b2sdk.v2 import B2Api, current_time_millis
+from b2sdk._internal.testing.helpers.bucket_manager import BucketManager
+from b2sdk.v2 import B2Api
 from b2sdk.v2.exception import DuplicateBucketName
-from test.integration.bucket_cleaner import BucketCleaner
-from test.integration.helpers import (
-    BUCKET_CREATED_AT_MILLIS,
-    random_bucket_name,
-)
 
 
 @pytest.mark.usefixtures('cls_setup')
 class IntegrationTestBase:
     b2_api: B2Api
     this_run_bucket_name_prefix: str
-    bucket_cleaner: BucketCleaner
+    bucket_manager: BucketManager
 
     @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_manager):
         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.bucket_manager = bucket_manager
         cls.b2_api = b2_api
         cls.info = b2_api.account_info
 
@@ -40,10 +36,7 @@ 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)
+            self.bucket_manager.clean_bucket(bucket)
 
     def write_zeros(self, file, number):
         line = b'0' * 1000 + b'\n'
@@ -54,13 +47,9 @@ def write_zeros(self, file, number):
             written += line_len
 
     def create_bucket(self):
-        bucket_name = self.generate_bucket_name()
+        bucket_name = self.bucket_manager.new_bucket_name()
         try:
-            bucket = self.b2_api.create_bucket(
-                bucket_name,
-                'allPublic',
-                bucket_info={BUCKET_CREATED_AT_MILLIS: str(current_time_millis())},
-            )
+            bucket = self.bucket_manager.create_bucket(name=bucket_name)
         except DuplicateBucketName:
             self._duplicated_bucket_name_debug_info(bucket_name)
             raise