Add ut for copy

Author: dt3310321

qcloud_cos/cos_comm.py

@@ -12,6 +12,7 @@
 from cos_exception import CosClientError
 from cos_exception import CosServiceError
 
+SINGLE_UPLOAD_LENGTH = 5*1024*1024*1024  # 单次上传文件最大为5G
 # kwargs中params到http headers的映射
 maplist = {
             'ContentLength': 'Content-Length',
@@ -165,7 +166,7 @@ def format_bucket(bucket, appid):
     if not isinstance(bucket, str):
         raise CosClientError("bucket is not str")
     # appid为空直接返回bucket
-    if appid == "":
+    if not appid:
         return bucket
     # appid不为空,检查是否以-appid结尾
     if bucket.endswith("-"+appid):
@@ -175,7 +176,7 @@ def format_bucket(bucket, appid):
 
 def get_copy_source_info(CopySource):
     """获取拷贝源的所有信息"""
-    appid = None
+    appid = ""
     if 'Appid' in CopySource.keys():
         appid = CopySource['Appid']
     if 'Bucket' in CopySource.keys():
@@ -190,8 +191,6 @@ def get_copy_source_info(CopySource):
         raise CosClientError('CopySource Need Parameter Region')
     if 'Key' in CopySource.keys():
         path = CopySource['Key']
-        if path and path[0] == '/':
-            path = path[1:]
     else:
         raise CosClientError('CopySource Need Parameter Key')
     return bucket, path, region
@@ -200,9 +199,20 @@ def get_copy_source_info(CopySource):
 def gen_copy_source_url(CopySource):
     """拼接拷贝源url"""
     bucket, path, region = get_copy_source_info(CopySource)
+    if path and path[0] == '/':
+        path = path[1:]
     url = "{bucket}.{region}.myqcloud.com/{path}".format(
             bucket=bucket,
             region=region,
             path=path
             )
     return url
+
+
+def gen_copy_source_range(begin_range, end_range):
+    """拼接bytes=begin-end形式的字符串"""
+    range = "bytes={first}-{end}".format(
+            first=begin_range,
+            end=end_range
+            )
+    return range

qcloud_cos/test.py

@@ -82,9 +82,11 @@ def test_put_get_delete_object_10MB():
         # get object
         get_response = client.get_object(
             Bucket=test_bucket,
-            Key=file_name
+            Key=file_name,
+            ResponseCacheControl='private'
         )
         assert etag == get_response['ETag']
+        assert 'private' == get_response['Cache-Control']
         download_fp = get_response['Body'].get_raw_stream()
         assert download_fp
         # delete object
@@ -494,7 +496,6 @@ def test_put_get_delete_replication():
     response = client.get_bucket_replication(
         Bucket=test_bucket
     )
-    print response
     assert response
     # delete lifecycle
     response = client.delete_bucket_replication(
@@ -559,11 +560,24 @@ def test_upload_file_multithreading():
         os.remove(file_name)
     print ed - st
 
+
+def test_copy_file_automatically():
+    """根据拷贝源文件的大小自动选择拷贝策略，小于5G直接copy_object，大于5G分块拷贝"""
+    copy_source = {'Appid': '1252448703', 'Bucket': 'testbucket', 'Key': '/thread_1MB', 'Region': 'ap-guangzhou'}
+    response = client.copy(
+        Bucket=test_bucket,
+        Key='copy_10G.txt',
+        CopySource=copy_source,
+        MAXThread=10
+    )
+
+
 if __name__ == "__main__":
     setUp()
     test_put_get_delete_object_10MB()
     test_put_get_versioning()
     test_put_get_delete_replication()
     test_upload_part_copy()
     test_upload_file_multithreading()
+    test_copy_file_automatically()
     tearDown()