Feat/new list buckets

qcloud_cos/cos_client.py

@@ -3345,7 +3345,7 @@ def get_bucket_intelligenttiering(self, Bucket, **kwargs):
         return data
 
     # service interface begin
-    def list_buckets(self, **kwargs):
+    def list_buckets(self, TagKey=None, TagValue=None, Region=None, CreateTime=None, Range=None, Marker="", MaxKeys=2000, **kwargs):
         """列出所有bucket
 
         :return(dict): 账号下bucket相关信息.
@@ -3366,11 +3366,26 @@ def list_buckets(self, **kwargs):
 
         if self._conf._service_domain is not None:
             url = '{scheme}://{domain}/'.format(scheme=self._conf._scheme, domain=self._conf._service_domain)
+
+        params = {
+            'marker': Marker,
+            'max-keys': MaxKeys,
+        }
+        if TagKey and TagValue:
+            params['tagkey'] = TagKey
+            params['tagvalue'] = TagValue
+        if Region:
+            params['region'] = Region
+        if CreateTime and Range:
+            params['create-time'] = CreateTime
+            params['range'] = Range
+
         rt = self.send_request(
             method='GET',
             url=url,
             bucket=None,
             headers=headers,
+            params=params,
             auth=CosS3Auth(self._conf),
         )
         data = xml_to_dict(rt.content)

ut/test.py

@@ -687,12 +687,50 @@ def test_get_bucket_location():
 
 
 def test_get_service():
-    return  # TODO: 测试账号的桶太多了导致列举超时，暂时屏蔽掉
-
     """列出账号下所有的bucket信息"""
     response = client.list_buckets()
     assert response
 
+    # 创建一个桶, 打tag
+    test_tagging_bucket = 'test-tagging-bucket-' + APPID
+    client.create_bucket(Bucket=test_tagging_bucket)
+    client.put_bucket_tagging(
+        Bucket=test_tagging_bucket,
+        Tagging={
+            'TagSet': {
+                'Tag': [
+                    {
+                        'Key': 'tagKey',
+                        'Value': 'tagValue'
+                    }
+                ]
+            }
+        }
+    )
+    response = client.list_buckets(Region=REGION, TagKey='tagKey', TagValue='tagValue')
+    for bucket in response['Buckets']['Bucket']:
+        tag = client.get_bucket_tagging(Bucket=bucket['Name'])
+        assert tag['TagSet']['Tag'][0]['Key'] == 'tagKey'
+        assert tag['TagSet']['Tag'][0]['Value'] == 'tagValue'
+
+    time.sleep(3)
+    client.delete_bucket(Bucket=test_tagging_bucket)
+
+    from datetime import datetime
+    marker = ""
+    list_over = False
+    while list_over is False:
+        create_time = 1514736000
+        response = client.list_buckets(Region='ap-beijing', CreateTime=create_time, Range='gt', Marker=marker)
+        for bucket in response['Buckets']['Bucket']:
+            ctime = int(time.mktime(datetime.strptime(bucket['CreationDate'], '%Y-%m-%dT%H:%M:%SZ').timetuple()))
+            assert ctime > create_time
+            assert bucket['Location'] == 'ap-beijing'
+
+        marker = response['Marker']
+        if response['IsTruncated'] == 'false':
+            list_over = True
+
 
 def test_put_get_delete_cors():
     """设置、获取、删除跨域配置"""
@@ -3526,7 +3564,6 @@ def test_check_multipart_upload():
 
 if __name__ == "__main__":
     setUp()
-    test_ci_delete_asr_template()
     """
     test_config_invalid_scheme()
     test_config_credential_inst()