Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_single_multipart_upload(self):
self.assertGreater(self.recommended_size, 2 * self.min_size)
remainder = 2 * self.min_size
source = UploadSource(self.recommended_size * 5 + remainder)
expected_part_sizes = [self.recommended_size] * 5 + [remainder]
self.verify_emerge_plan_for_write_intents(
[WriteIntent(source)],
self.split_source_to_part_defs(source, expected_part_sizes),
)
def test_single_part_upload(self):
source = UploadSource(self.recommended_size)
self.verify_emerge_plan_for_write_intents(
[WriteIntent(source)],
[part(source)],
)
def test_local_small_copy_overlap(self):
self.assertGreater(self.recommended_size, self.min_size * 3 - 3)
source_upload = UploadSource(self.recommended_size)
small_size = self.min_size - 1
source_copy1 = CopySource(small_size)
source_copy2 = CopySource(small_size)
source_copy3 = CopySource(small_size)
write_intents = [
WriteIntent(source_upload),
WriteIntent(source_copy1),
WriteIntent(source_copy2, destination_offset=small_size),
WriteIntent(source_copy3, destination_offset=2 * small_size),
]
self.verify_emerge_plan_for_write_intents(
write_intents,
[part(source_upload)],
)
def test_local_remote_overlap_end(self):
source_upload = UploadSource(self.recommended_size * 2)
source_copy = CopySource(self.recommended_size)
write_intents = [
WriteIntent(source_upload),
WriteIntent(source_copy, destination_offset=self.recommended_size),
]
self.verify_emerge_plan_for_write_intents(
write_intents,
[
part(source_upload, 0, self.recommended_size),
part(source_copy),
],
def test_basic_local_overlap(self):
source1 = UploadSource(self.recommended_size * 2)
source2 = UploadSource(self.recommended_size * 2)
write_intents = [
WriteIntent(source1),
WriteIntent(source2, destination_offset=self.recommended_size),
]
self.verify_emerge_plan_for_write_intents(
write_intents,
[part(source1, 0, self.recommended_size)] +
self.split_source_to_part_defs(source2, [self.recommended_size] * 2),
)
def test_local_small_copy_overlap(self):
self.assertGreater(self.recommended_size, self.min_size * 3 - 3)
source_upload = UploadSource(self.recommended_size)
small_size = self.min_size - 1
source_copy1 = CopySource(small_size)
source_copy2 = CopySource(small_size)
source_copy3 = CopySource(small_size)
write_intents = [
WriteIntent(source_upload),
WriteIntent(source_copy1),
WriteIntent(source_copy2, destination_offset=small_size),
WriteIntent(source_copy3, destination_offset=2 * small_size),
]
self.verify_emerge_plan_for_write_intents(
write_intents,
[part(source_upload)],
)
def test_raise_on_hole(self):
source_upload1 = UploadSource(self.recommended_size)
source_upload2 = UploadSource(self.recommended_size)
source_copy1 = CopySource(self.recommended_size)
source_copy2 = CopySource(self.recommended_size)
write_intents = [
WriteIntent(source_upload1),
WriteIntent(source_upload2, destination_offset=self.recommended_size + 2 * MEGABYTE),
WriteIntent(source_copy1, destination_offset=MEGABYTE),
WriteIntent(source_copy2, destination_offset=self.recommended_size + 3 * MEGABYTE),
]
hole_msg = ('Cannot emerge file with holes. '
'Found hole range: ({}, {})'.format(
write_intents[2].destination_end_offset,
write_intents[1].destination_offset,
))
with self.assertRaises(ValueError, hole_msg):
self.planner.get_emerge_plan(write_intents)
open (and re-open) the file. The result of opening should be a binary
file whose read() method returns bytes.
:param b2sdk.v1.UploadSource upload_source: an object that opens the source of the upload
:param str file_name: the file name of the new B2 file
:param str,None content_type: the MIME type, or ``None`` to accept the default based on file extension of the B2 file name
:param dict,None file_infos: a file info to store with the file or ``None`` to not store anything
:param int,None min_part_size: the smallest part size to use or ``None`` to determine automatically
:param b2sdk.v1.AbstractProgressListener,None progress_listener: a progress listener object to use, or ``None`` to not report progress
The function `opener` should return a file-like object, and it
must be possible to call it more than once in case the upload
is retried.
"""
return self.create_file(
[WriteIntent(upload_source)],
file_name,
content_type=content_type,
file_info=file_info,
progress_listener=progress_listener,
# FIXME: Bucket.upload documents wrong logic
recommended_upload_part_size=min_part_size,
)