add put/get bucket intelligenttiering

Author: jayzhenghan

qcloud_cos/cos_client.py

@@ -2847,6 +2847,7 @@ def get_bucket_referer(self, Bucket, **kwargs):
 
     def delete_bucket_referer(self, Bucket, **kwargs):
         """删除bucket防盗链规则
+
         :param Bucket(string): 存储桶名称.
         :param kwargs(dict): 设置请求headers.
         :return(dict): None.

ut/test.py

@@ -1356,6 +1356,35 @@ def test_aes_client():
     if os.path.exists('test_multi_upload_local'):
         os.remove('test_multi_upload_local')
 
+    client_for_rsa.delete_object(test_bucket, 'test_multi_upload')
+
+def test_rsa_client():
+    """测试rsa加密客户端的上传下载操作"""
+    content = '123456' * 1024 + '1'
+    client_for_rsa.delete_object(test_bucket, 'test_for_rsa')
+    client_for_rsa.put_object(test_bucket, content, 'test_for_rsa')
+    # 测试整个文件的md5
+    response = client_for_rsa.get_object(test_bucket, 'test_for_rsa')
+    response['Body'].get_stream_to_file('test_for_rsa_local')
+    local_file_md5 = None
+    content_md5 = None
+    with open('test_for_rsa_local', 'rb') as f:
+        local_file_md5 = get_raw_md5(f.read())
+    content_md5 = get_raw_md5(content.encode("utf-8"))
+    assert local_file_md5 and content_md5 and local_file_md5 == content_md5
+    if os.path.exists('test_for_rsa_local'):
+        os.remove('test_for_rsa_local')
+
+    # 测试读取部分数据的md5
+    response = client_for_rsa.get_object(test_bucket, 'test_for_rsa', Range='bytes=5-3000')
+    response['Body'].get_stream_to_file('test_for_rsa_local')
+    with open('test_for_rsa_local', 'rb') as f:
+        local_file_md5 = get_raw_md5(f.read())
+    content_md5 = get_raw_md5(content[5:3001].encode("utf-8"))
+    assert local_file_md5 and content_md5 and local_file_md5 == content_md5
+    if os.path.exists('test_for_rsa_local'):
+        os.remove('test_for_rsa_local')
+
     client_for_rsa.delete_object(test_bucket, 'test_for_rsa')
 
     content = '1' * 1024 * 1024
@@ -1375,6 +1404,7 @@ def test_aes_client():
     assert local_file_md5 and content_md5 and local_file_md5 == content_md5
     if os.path.exists('test_multi_upload_local'):
         os.remove('test_multi_upload_local')
+
     client_for_rsa.delete_object(test_bucket, 'test_multi_upload')