Add bucket name check.

oss2/api.py

@@ -398,6 +398,8 @@ def __init__(self, auth, endpoint, bucket_name,
                                      app_name, enable_crc)
 
         self.bucket_name = bucket_name.strip()
+        if utils.is_valid_bucket_name(self.bucket_name) is not True:
+            raise ClientError("The bucket_name is invalid, please check it.")
 
     def sign_url(self, method, key, expires, headers=None, params=None, slash_safe=False):
         """生成签名URL。

tests/common.py

@@ -48,11 +48,19 @@ def clean_and_delete_bucket(bucket):
     try:
         result = bucket.get_bucket_info()
         if result.versioning_status in [oss2.BUCKET_VERSIONING_ENABLE, oss2.BUCKET_VERSIONING_SUSPEND]:
-            all_objects = bucket.list_object_versions()
-            for obj in all_objects.versions:
-                bucket.delete_object(obj.key, params={'versionId': obj.versionid})
-            for del_marker in all_objects.delete_marker:
-                bucket.delete_object(del_marker.key, params={'versionId': del_marker.versionid})
+            next_key_marker = None
+            next_versionid_marker = None
+            is_truncated = True
+            while is_truncated is True:
+                objects = bucket.list_object_versions(key_marker=next_key_marker, versionid_marker=next_versionid_marker)
+                for obj in objects.versions:
+                    bucket.delete_object(obj.key, params={'versionId': obj.versionid})
+                for del_marker in objects.delete_marker:
+                    bucket.delete_object(del_marker.key, params={'versionId': del_marker.versionid})
+                is_truncated = objects.is_truncated
+                if is_truncated:
+                    next_key_marker = objects.next_key_marker
+                    next_versionid_marker = objects.next_versionid_marker
     except:
         pass
 

tests/test_api_base.py

@@ -33,8 +33,7 @@ def test_ip(self):
 
     def test_invalid_bucket_name(self):
         bucket_name = random_string(64)
-        bucket = oss2.Bucket(oss2.AnonymousAuth(), OSS_ENDPOINT, bucket_name)
-        self.assertRaises(oss2.exceptions.NoSuchBucket, bucket.get_object, 'hello.txt')
+        self.assertRaises(oss2.exceptions.ClientError, oss2.Bucket, oss2.AnonymousAuth(), OSS_ENDPOINT, bucket_name)
 
     def test_whitespace(self):
         bucket = oss2.Bucket(oss2.Auth(OSS_ID, ' ' + OSS_SECRET + ' '), OSS_ENDPOINT, OSS_BUCKET)

tests/test_object.py

@@ -717,7 +717,7 @@ def test_object_exists(self):
         key = self.random_key()
 
         auth = oss2.Auth(OSS_ID, OSS_SECRET)
-        bucket_name = OSS_BUCKET + "-test-object_exists"
+        bucket_name = OSS_BUCKET + "-test-object-exists"
         bucket = oss2.Bucket(auth, OSS_ENDPOINT, bucket_name)
         self.assertRaises(NoSuchBucket, bucket.object_exists, key)
 

tests/test_object_versioning.py

@@ -1008,7 +1008,7 @@ def test_delete_object_versions_with_invalid_arguments(self):
         from oss2.models import BatchDeleteObjectVersionList
 
         auth = oss2.Auth(OSS_ID, OSS_SECRET)
-        bucket_name = OSS_BUCKET + "test-delete-object-versions"
+        bucket_name = OSS_BUCKET + "-test-delete-object-versions"
         bucket = oss2.Bucket(auth, self.endpoint, bucket_name)
         bucket.create_bucket(oss2.BUCKET_ACL_PRIVATE)
 

tests/test_utils.py

@@ -51,6 +51,17 @@ def test_is_valid_bucket_name(self):
         self.assertTrue(not oss2.is_valid_bucket_name('hello-'))
         self.assertTrue(not oss2.is_valid_bucket_name('-hello'))
 
+        access_key_id = "test_access_key_id"
+        access_key_secret = "test_access_key_secret"
+        endpoint = "oss-cn-shenzhen.aliyuncs.com"
+
+        bucket_name = "hello"
+        bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
+
+        bucket_name = "hello-"
+        self.assertRaises(oss2.exceptions.ClientError, oss2.Bucket, oss2.Auth(access_key_id, access_key_secret), 
+            endpoint, bucket_name)
+
     def test_compat(self):
         # from unicode
         u = u'中文'