Improve debugging in integration tests BucketManager

b2sdk/_internal/testing/helpers/base.py

@@ -13,7 +13,6 @@
 
 from b2sdk._internal.testing.helpers.bucket_manager import BucketManager
 from b2sdk.v2 import B2Api
-from b2sdk.v2.exception import DuplicateBucketName
 
 
 @pytest.mark.usefixtures('cls_setup')
@@ -47,30 +46,6 @@ def write_zeros(self, file, number):
             written += line_len
 
     def create_bucket(self):
-        bucket_name = self.bucket_manager.new_bucket_name()
-        try:
-            bucket = self.bucket_manager.create_bucket(name=bucket_name)
-        except DuplicateBucketName:
-            self._duplicated_bucket_name_debug_info(bucket_name)
-            raise
+        bucket = self.bucket_manager.create_bucket()
         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, '='))

b2sdk/_internal/testing/helpers/bucket_manager.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import logging
+import os
 from collections.abc import Iterable
 from datetime import datetime, timedelta
 from itertools import chain
@@ -19,7 +20,13 @@
 
 from b2sdk._internal.api import B2Api
 from b2sdk._internal.bucket import Bucket
-from b2sdk._internal.exception import BadRequest, BucketIdNotFound, FileNotPresent, TooManyRequests
+from b2sdk._internal.exception import (
+    BadRequest,
+    BucketIdNotFound,
+    DuplicateBucketName,
+    FileNotPresent,
+    TooManyRequests,
+)
 from b2sdk._internal.file_lock import NO_RETENTION_FILE_SETTING, LegalHold, RetentionMode
 from b2sdk._internal.testing.helpers.buckets import (
     BUCKET_CREATED_AT_MILLIS,
@@ -48,14 +55,26 @@ def __init__(
         self.general_prefix = general_prefix
         self.dont_cleanup_old_buckets = dont_cleanup_old_buckets
         self.b2_api = b2_api
-        self.bucket_name_log: list[str] = []
+        self.bucket_name_mapping: dict[str, str] = {}
 
     def new_bucket_name(self) -> str:
-        bucket_name = self.current_run_prefix + bucket_name_part(
-            BUCKET_NAME_LENGTH - len(self.current_run_prefix)
+        attempts = 5
+
+        for _ in range(attempts):
+            bucket_name = self.current_run_prefix + bucket_name_part(
+                BUCKET_NAME_LENGTH - len(self.current_run_prefix)
+            )
+            if bucket_name not in self.bucket_name_mapping:
+                return bucket_name
+
+        raise RuntimeError(
+            'Failed to generate a unique bucket name after %d attempts. Last tried name: %s. Existing buckets:\n%s'
+            % (
+                attempts,
+                bucket_name,
+                self.bucket_name_mapping,
+            )
         )
-        self.bucket_name_log.append(bucket_name)
-        return bucket_name
 
     def new_bucket_info(self) -> dict:
         return {
@@ -65,12 +84,25 @@ def new_bucket_info(self) -> dict:
 
     def create_bucket(self, bucket_type: str = 'allPublic', **kwargs) -> Bucket:
         bucket_name = kwargs.pop('name', self.new_bucket_name())
-        return self.b2_api.create_bucket(
-            bucket_name,
-            bucket_type=bucket_type,
-            bucket_info=self.new_bucket_info(),
-            **kwargs,
-        )
+
+        try:
+            bucket = self.b2_api.create_bucket(
+                bucket_name,
+                bucket_type=bucket_type,
+                bucket_info=self.new_bucket_info(),
+                **kwargs,
+            )
+        except DuplicateBucketName:
+            logger.error(
+                'Bucket %s already exists. Currently used buckets:\n%s',
+                bucket_name,
+                self.bucket_name_mapping,
+            )
+            raise
+
+        self.bucket_name_mapping[bucket_name] = os.getenv('PYTEST_CURRENT_TEST', 'unknown test')
+
+        return bucket
 
     def _should_remove_bucket(self, bucket: Bucket) -> tuple[bool, str]:
         if self.current_run_prefix and bucket.name.startswith(self.current_run_prefix):

b2sdk/_internal/testing/helpers/buckets.py

@@ -39,7 +39,7 @@ def get_seed() -> str:
             os.getenv('WORKFLOW_ID', secrets.token_hex(8)),
             NODE_DESCRIPTION,
             str(time.time_ns()),
-            os.getenv('PYTEST_XDIST_WORKER', 'gw0'),
+            os.getenv('PYTEST_XDIST_WORKER', 'master'),
         )
     )
     return sha256(seed.encode()).hexdigest()[:16]

changelog.d/+bucket-manager-debugging.infrastructure.md

@@ -0,0 +1 @@
+Improve debugging in integration tests bucket manager.