Implement persistent bucket fixtures for integration tests

changelog.d/+test_with_persistent_bucket.infrastructure.md

@@ -0,0 +1 @@
+Improve internal testing infrastructure by updating integration tests to use persistent buckets.

test/integration/conftest.py

@@ -16,6 +16,7 @@
 import subprocess
 import sys
 import tempfile
+import uuid
 from os import environ, path
 from tempfile import TemporaryDirectory
 
@@ -31,6 +32,10 @@
 
 from ..helpers import b2_uri_args_v3, b2_uri_args_v4
 from .helpers import NODE_DESCRIPTION, RNG_SEED, Api, CommandLine, bucket_name_part, random_token
+from .persistent_bucket import (
+    PersistentBucketAggregate,
+    get_or_create_persistent_bucket,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -402,3 +407,20 @@ def b2_uri_args(apiver_int):
         return b2_uri_args_v4
     else:
         return b2_uri_args_v3
+
+
+# -- Persistent bucket fixtures --
+@pytest.fixture
+def unique_subfolder():
+    subfolder = f"test-{uuid.uuid4().hex[:8]}"
+    yield subfolder
+
+
+@pytest.fixture
+def persistent_bucket(unique_subfolder, b2_api) -> 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.
+    """
+    persistent_bucket = get_or_create_persistent_bucket(b2_api)
+    yield PersistentBucketAggregate(persistent_bucket.name, unique_subfolder)

test/integration/helpers.py

@@ -188,7 +188,6 @@ def _should_remove_bucket(self, bucket: Bucket) -> tuple[bool, str]:
     def clean_buckets(self, quick=False):
         # even with use_cache=True, if cache is empty API call will be made
         buckets = self.api.list_buckets(use_cache=quick)
-        print('Total bucket count:', len(buckets))
         remaining_buckets = []
         for bucket in buckets:
             should_remove, why = self._should_remove_bucket(bucket)
@@ -539,9 +538,9 @@ def reauthorize(self, check_key_capabilities=False):
             } - private_preview_caps - set(auth_dict['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 list_file_versions(self, bucket_name):
+    def list_file_versions(self, bucket_name, path=''):
         return self.should_succeed_json(
-            ['ls', '--json', '--recursive', '--versions', *self.b2_uri_args(bucket_name)]
+            ['ls', '--json', '--recursive', '--versions', *self.b2_uri_args(bucket_name, path)]
         )
 
     def cleanup_buckets(self, buckets: dict[str, dict | None]) -> None:

test/integration/persistent_bucket.py

@@ -0,0 +1,63 @@
+######################################################################
+#
+# 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
+from dataclasses import dataclass
+from functools import cached_property
+from test.integration.helpers import BUCKET_NAME_LENGTH, Api
+
+import backoff
+from b2sdk.v2 import Bucket
+from b2sdk.v2.exception import DuplicateBucketName, NonExistentBucket
+
+PERSISTENT_BUCKET_NAME_PREFIX = "constst"
+
+
+@dataclass
+class PersistentBucketAggregate:
+    bucket_name: str
+    subfolder: str
+
+    @cached_property
+    def virtual_bucket_name(self):
+        return f"{self.bucket_name}/{self.subfolder}"
+
+
+def get_persistent_bucket_name(b2_api: Api) -> str:
+    bucket_base = os.environ.get("GITHUB_REPOSITORY_ID", b2_api.api.get_account_id())
+    bucket_hash = hashlib.sha256(bucket_base.encode()).hexdigest()
+    return f"{PERSISTENT_BUCKET_NAME_PREFIX}-{bucket_hash}" [:BUCKET_NAME_LENGTH]
+
+
+@backoff.on_exception(
+    backoff.expo,
+    DuplicateBucketName,
+    max_tries=3,
+    jitter=backoff.full_jitter,
+)
+def get_or_create_persistent_bucket(b2_api: Api) -> Bucket:
+    bucket_name = get_persistent_bucket_name(b2_api)
+    try:
+        bucket = b2_api.api.get_bucket_by_name(bucket_name)
+    except NonExistentBucket:
+        bucket = b2_api.api.create_bucket(
+            bucket_name,
+            bucket_type="allPublic",
+            lifecycle_rules=[
+                {
+                    "daysFromHidingToDeleting": 1,
+                    "daysFromUploadingToHiding": 1,
+                    "fileNamePrefix": "",
+                }
+            ],
+        )
+    # add the new bucket name to the list of bucket names
+    b2_api.bucket_name_log.append(bucket_name)
+    return bucket