Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def do_multipart_internal(self, do_md5):
key = self.random_key()
content = random_bytes(128 * 1024)
parts = []
upload_id = self.bucket.init_multipart_upload(key).upload_id
if do_md5:
headers = {'Content-Md5': oss2.utils.content_md5(content)}
else:
headers = None
result = self.bucket.upload_part(key, upload_id, 1, content, headers=headers)
parts.append(oss2.models.PartInfo(1, result.etag, size=len(content), part_crc=result.crc))
self.assertTrue(result.crc is not None)
complete_result = self.bucket.complete_multipart_upload(key, upload_id, parts)
object_crc = calc_obj_crc_from_parts(parts)
self.assertTrue(complete_result.crc is not None)
self.assertEqual(object_crc, result.crc)
result = self.bucket.get_object(key)
self.assertEqual(content, result.read())
# Upload part without payer setting, should be failed.
with open(filename, 'rb') as fileobj:
part_number = 1
num_to_upload = part_size
self.assertRaises(oss2.exceptions.AccessDenied, self.payer_bucket.upload_part, key, upload_id, part_number,
SizedFileAdapter(fileobj, num_to_upload))
# Upload part with payer setting, should be successful.
with open(filename, 'rb') as fileobj:
part_number = 1
offset = 0
while offset < total_size:
num_to_upload = min(part_size, total_size - offset)
result = self.payer_bucket.upload_part(key, upload_id, part_number,
SizedFileAdapter(fileobj, num_to_upload), headers=headers)
parts.append(PartInfo(part_number, result.etag))
offset += num_to_upload
part_number += 1
# Complete multipart upload without payer setting, should be failed.
self.assertRaises(oss2.exceptions.AccessDenied, self.payer_bucket.complete_multipart_upload, key, upload_id, parts)
# Complete multipart upload with payer setting, should be successful.
result = self.payer_bucket.complete_multipart_upload(key, upload_id, parts, headers=headers)
self.assertEqual(result.status, 200)
self.bucket.delete_object(key)
os.remove(filename)
def test_upload_part_copy(self):
src_object = self.random_key()
dst_object = self.random_key()
content = random_bytes(200 * 1024)
# 上传源文件
self.bucket.put_object(src_object, content)
# part copy到目标文件
parts = []
upload_id = self.bucket.init_multipart_upload(dst_object).upload_id
result = self.bucket.upload_part_copy(self.bucket.bucket_name, src_object,
(0, 100 * 1024 - 1), dst_object, upload_id, 1)
parts.append(oss2.models.PartInfo(1, result.etag))
result = self.bucket.upload_part_copy(self.bucket.bucket_name, src_object,
(100*1024, None), dst_object, upload_id, 2)
parts.append(oss2.models.PartInfo(2, result.etag))
self.bucket.complete_multipart_upload(dst_object, upload_id, parts)
# 验证
content_got = self.bucket.get_object(dst_object).read()
self.assertEqual(len(content_got), len(content))
self.assertEqual(content_got, content)
def test_part_iterator(self):
for key in [random_string(16), '中文+_)(*&^%$#@!前缀', u'中文+_)(*&^%$#@!前缀']:
upload_id = self.bucket.init_multipart_upload(key).upload_id
# 准备分片
part_list = []
for part_number in [1, 3, 6, 7, 9, 10]:
content = random_string(128 * 1024)
etag = hashlib.md5(oss2.to_bytes(content)).hexdigest().upper()
part_list.append(oss2.models.PartInfo(part_number, etag, len(content)))
self.bucket.upload_part(key, upload_id, part_number, content)
# 验证
parts_got = []
for part_info in oss2.PartIterator(self.bucket, key, upload_id):
parts_got.append(part_info)
self.assertEqual(len(part_list), len(parts_got))
for i in range(len(part_list)):
self.assertEqual(part_list[i].part_number, parts_got[i].part_number)
self.assertEqual(part_list[i].etag, parts_got[i].etag)
self.assertEqual(part_list[i].size, parts_got[i].size)
self.bucket.abort_multipart_upload(key, upload_id)
def test_complete(self, do_request):
from oss2.models import PartInfo
parts = list()
parts.append(PartInfo(1, '4DE8075FB607DF4D13FBC480EA488EFA'))
parts.append(PartInfo(2, 'AF947EC157726CEA88ED83B3C989063B'))
request_text = '''POST /pasncdoyuvuvuiyewfsobdwn?uploadId=65484B78EF3846298B8E2DC1643F8F37 HTTP/1.1
Host: ming-oss-share.oss-cn-hangzhou.aliyuncs.com
Accept-Encoding: identity
Connection: keep-alive
Content-Length: 223
date: Sat, 12 Dec 2015 00:36:26 GMT
User-Agent: aliyun-sdk-python/2.0.2(Windows/7/;3.3.3)
Accept: */*
authorization: OSS ZCDmm7TPZKHtx77j:TgjWAumJAl8dDr0yqWHOyqqwrd0=
1"4DE8075FB607DF4D13FBC480EA488EFA"2"AF947EC157726CEA88ED83B3C989063B"'''
response_text = '''HTTP/1.1 200 OK
Server: AliyunOSS
def test_part_iterator(self):
for key in [random_string(16), '中文+_)(*&^%$#@!前缀', u'中文+_)(*&^%$#@!前缀']:
upload_id = self.bucket.init_multipart_upload(key).upload_id
# 准备分片
part_list = []
for part_number in [1, 3, 6, 7, 9, 10]:
content = random_string(128 * 1024)
etag = hashlib.md5(oss2.to_bytes(content)).hexdigest().upper()
part_list.append(oss2.models.PartInfo(part_number, etag, len(content)))
self.bucket.upload_part(key, upload_id, part_number, content)
# 验证
parts_got = []
for part_info in oss2.PartIterator(self.bucket, key, upload_id):
parts_got.append(part_info)
self.assertEqual(len(part_list), len(parts_got))
for i in range(len(part_list)):
self.assertEqual(part_list[i].part_number, parts_got[i].part_number)
self.assertEqual(part_list[i].etag, parts_got[i].etag)
self.assertEqual(part_list[i].size, parts_got[i].size)
self.bucket.abort_multipart_upload(key, upload_id)
def send_buf(self):
upload_id = self.get_upload_id()
assert upload_id is not None
if not self.buf:
return
self.part_num += 1
res = self.upload_part()
self.buf = ''
self.part_list.append(oss2.models.PartInfo(self.part_num, res.etag))