fix some bugs and modify ut

Author: dt3310321

qcloud_cos/cos_auth.py

@@ -54,12 +54,13 @@ def __call__(self, r):
         path = self._path
         uri_params = self._params
         headers = filter_headers(r.headers)
+        uri_params = dict([(k.lower(), v.lower()) for k, v in uri_params.items()])
         # reserved keywords in headers urlencode are -_.~, notice that / should be encoded and space should not be encoded to plus sign(+)
         headers = dict([(k.lower(), quote(v, '-_.~')) for k, v in headers.items()])  # headers中的key转换为小写，value进行encode
         format_str = "{method}\n{host}\n{params}\n{headers}\n".format(
             method=r.method.lower(),
             host=path,
-            params=urllib.urlencode(sorted(uri_params.items())),
+            params=urllib.urlencode(sorted(uri_params.items())).replace('+', '%20'),
             headers='&'.join(map(lambda (x, y): "%s=%s" % (x, y), sorted(headers.items())))
         )
         logger.debug("format str: " + format_str)

qcloud_cos/cos_client.py

@@ -1829,11 +1829,12 @@ def _check_all_upload_parts(self, bucket, key, uploadid, local_path, parts_num,
                 UploadId=uploadid,
                 PartNumberMarker=part_number_marker
             )
-            parts_info.extend(response['Part'])
+            if 'Part' in response:
+                parts_info.extend(response['Part'])
             if response['IsTruncated'] == 'false':
                 list_over_status = True
             else:
-                part_number_marker = int(response['NextMarker'])
+                part_number_marker = int(response['NextPartNumberMarker'])
         for part in parts_info:
             part_num = int(part['PartNumber'])
             # 如果分块数量大于本地计算出的最大数量,校验失败

ut/test.py

@@ -12,7 +12,8 @@
 
 SECRET_ID = os.environ["SECRET_ID"]
 SECRET_KEY = os.environ["SECRET_KEY"]
-test_bucket = "test01-1252448703"
+TRAVIS_FLAG = os.environ["TRAVIS_FLAG"]
+test_bucket = 'cos-python-v5-test-' + str(sys.version_info[0]) + '-' + str(sys.version_info[1]) + '-' + '1252448703'
 test_object = "test.txt"
 special_file_name = "中文" + "→↓←→↖↗↙↘! \"#$%&'()*+,-./0123456789:;<=>@ABCDEFGHIJKLMNOPQRSTUVWXYZ[\\]^_`abcdefghijklmnopqrstuvwxyz{|}~"
 conf = CosConfig(
@@ -23,6 +24,19 @@
 client = CosS3Client(conf)
 
 
+def _create_test_bucket(test_bucket):
+    try:
+        response = client.create_bucket(
+            Bucket=test_bucket,
+        )
+    except Exception as e:
+        if e.get_error_code() == 'BucketAlreadyOwnedByYou':
+            print('BucketAlreadyOwnedByYou')
+        else:
+            raise e
+    return None
+
+
 def get_raw_md5(data):
     m2 = hashlib.md5(data)
     etag = '"' + str(m2.hexdigest()) + '"'
@@ -37,22 +51,24 @@ def gen_file(path, size):
 
 
 def print_error_msg(e):
-    print e.get_origin_msg()
-    print e.get_digest_msg()
-    print e.get_status_code()
-    print e.get_error_code()
-    print e.get_error_msg()
-    print e.get_resource_location()
-    print e.get_trace_id()
-    print e.get_request_id()
+    print (e.get_origin_msg())
+    print (e.get_digest_msg())
+    print (e.get_status_code())
+    print (e.get_error_code())
+    print (e.get_error_msg())
+    print (e.get_resource_location())
+    print (e.get_trace_id())
+    print (e.get_request_id())
 
 
 def setUp():
-    print "start test..."
+    print ("start test...")
+    print ("start create bucket " + test_bucket)
+    _create_test_bucket(test_bucket)
 
 
 def tearDown():
-    print "function teardown"
+    print ("function teardown")
 
 
 def test_put_get_delete_object_10MB():
@@ -146,6 +162,11 @@ def test_put_object_non_exist_bucket():
 
 def test_put_object_acl():
     """设置object acl"""
+    response = client.put_object(
+        Bucket=test_bucket,
+        Key=test_object,
+        Body='test acl'
+    )
     response = client.put_object_acl(
         Bucket=test_bucket,
         Key=test_object,
@@ -160,6 +181,10 @@ def test_get_object_acl():
         Key=test_object
     )
     assert response
+    response = client.delete_object(
+        Bucket=test_bucket,
+        Key=test_object
+    )
 
 
 def test_copy_object_diff_bucket():
@@ -352,7 +377,7 @@ def test_get_bucket_acl_normal():
 def test_list_objects():
     """列出bucket下的objects"""
     response = client.list_objects(
-        Bucket=test_bucket,
+        Bucket='test01-1252448703',
         MaxKeys=100,
         Prefix='中文',
         Delimiter='/'
@@ -376,7 +401,7 @@ def test_get_presigned_url():
         Key='中文.txt'
     )
     assert url
-    print url
+    print (url)
 
 
 def test_get_bucket_location():
@@ -496,7 +521,7 @@ def test_put_get_delete_replication():
             {
                 'ID': '123',
                 'Status': 'Enabled',
-                'Prefix': 'replication',
+                'Prefix': '中文',
                 'Destination': {
                     'Bucket': 'qcs:id/0:cos:cn-south:appid/1252448703:replicationsouth'
                 }
@@ -516,7 +541,7 @@ def test_put_get_delete_replication():
         Bucket=test_bucket
     )
     assert response
-    # delete lifecycle
+    # delete replication
     response = client.delete_bucket_replication(
         Bucket=test_bucket
     )
@@ -564,7 +589,10 @@ def test_list_multipart_uploads():
 def test_upload_file_multithreading():
     """根据文件大小自动选择分块大小,多线程并发上传提高上传速度"""
     file_name = "thread_1GB"
-    gen_file(file_name, 5)  # set 5MB beacuse travis too slow
+    file_size = 1024
+    if TRAVIS_FLAG == 'true':
+        file_size = 5  # set 5MB beacuse travis too slow
+    gen_file(file_name, file_size)
     st = time.time()  # 记录开始时间
     response = client.upload_file(
         Bucket=test_bucket,
@@ -577,7 +605,7 @@ def test_upload_file_multithreading():
     ed = time.time()  # 记录结束时间
     if os.path.exists(file_name):
         os.remove(file_name)
-    print ed - st
+    print (ed - st)
 
 
 def test_copy_file_automatically():
@@ -625,7 +653,8 @@ def test_use_get_auth():
         Key='test.txt',
         Params={'acl': '', 'unsed': '123'}
     )
-    response = requests.get('http://test01-1252448703.cos.ap-beijing-1.myqcloud.com/test.txt?acl&unsed=123', headers={'Authorization': auth})
+    url = 'http://' + test_bucket + '.cos.ap-beijing-1.myqcloud.com/test.txt?acl&unsed=123'
+    response = requests.get(url, headers={'Authorization': auth})
     assert response.status_code == 200
 
 
@@ -663,16 +692,15 @@ def test_put_get_bucket_logging():
     response = logging_client.get_bucket_logging(
         Bucket=logging_bucket
     )
-    print response
+    print (response)
     assert response['LoggingEnabled']['TargetBucket'] == logging_bucket
     assert response['LoggingEnabled']['TargetPrefix'] == 'test'
 
 
 def test_put_object_enable_md5():
     """上传文件,SDK计算content-md5头部"""
-    file_size = 10
     file_name = 'test_object_sdk_caculate_md5.file'
-    gen_file(file_name, 10)
+    gen_file(file_name, 1)
     with open(file_name, 'rb') as f:
         etag = get_raw_md5(f.read())
     with open(file_name, 'rb') as fp:
@@ -703,6 +731,10 @@ def test_put_object_from_local_file():
         Key=file_name
     )
     assert put_response['ETag'] == etag
+    response = client.delete_object(
+        Bucket=test_bucket,
+        Key=file_name
+    )
     if os.path.exists(file_name):
         os.remove(file_name)