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_it(self):
for test_case in ENCODING_TEST_CASES:
string = test_case['string']
fully_encoded = test_case['fullyEncoded']
minimally_encoded = test_case['minimallyEncoded']
encoded = b2sdk.utils.b2_url_encode(string)
expected_encoded = (minimally_encoded, fully_encoded)
if encoded not in expected_encoded:
print(
'string: %s encoded: %s expected: %s' %
(repr(string), encoded, expected_encoded)
)
self.assertTrue(encoded in expected_encoded)
self.assertEqual(string, b2sdk.utils.b2_url_decode(fully_encoded))
self.assertEqual(string, b2sdk.utils.b2_url_decode(minimally_encoded))
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_simple_concatenate(self):
sources = [
CopySource(self.recommended_size),
UploadSource(self.recommended_size),
CopySource(self.recommended_size),
UploadSource(self.recommended_size),
]
self.verify_emerge_plan_for_write_intents(
WriteIntent.wrap_sources_iterator(sources),
[part(source) for source in sources],
)
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)
def _check_one(self, expected, x):
self.assertEqual(expected, b2sdk.utils.format_and_scale_number(x, 'B'))