diff --git a/qcloud_cos/cos_client.py b/qcloud_cos/cos_client.py index d5379ad6..4a87dfcd 100644 --- a/qcloud_cos/cos_client.py +++ b/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) diff --git a/ut/test.py b/ut/test.py index a08ab275..74573bad 100644 --- a/ut/test.py +++ b/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()