Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
Show First 20 Lines • Show All 723 Lines • ▼ Show 20 Lines | def test_content_add_metadata_collision(self): | ||||
sha256_array[0] += 1 | sha256_array[0] += 1 | ||||
cont1b['sha256'] = bytes(sha256_array) | cont1b['sha256'] = bytes(sha256_array) | ||||
with self.assertRaises(HashCollision) as cm: | with self.assertRaises(HashCollision) as cm: | ||||
self.storage.content_add_metadata([cont1, cont1b]) | self.storage.content_add_metadata([cont1, cont1b]) | ||||
self.assertIn(cm.exception.args[0], ['sha1', 'sha1_git', 'blake2s256']) | self.assertIn(cm.exception.args[0], ['sha1', 'sha1_git', 'blake2s256']) | ||||
def test_skipped_content_add(self): | def test_skipped_content_add_db(self): | ||||
cont = self.skipped_cont.copy() | cont = self.skipped_cont.copy() | ||||
cont2 = self.skipped_cont2.copy() | cont2 = self.skipped_cont2.copy() | ||||
cont2['blake2s256'] = None | cont2['blake2s256'] = None | ||||
actual_result = self.storage.content_add([cont, cont, cont2]) | actual_result = self.storage.content_add([cont, cont, cont2]) | ||||
self.assertEqual(actual_result, { | self.assertEqual(actual_result, { | ||||
'content:add': 0, | 'content:add': 0, | ||||
Show All 21 Lines | def test_skipped_content_add_db(self): | ||||
self.assertEqual( | self.assertEqual( | ||||
(datum2[0].tobytes(), datum2[1].tobytes(), datum2[2].tobytes(), | (datum2[0].tobytes(), datum2[1].tobytes(), datum2[2].tobytes(), | ||||
datum2[3], datum2[4], datum2[5], datum2[6]), | datum2[3], datum2[4], datum2[5], datum2[6]), | ||||
(cont2['sha1'], cont2['sha1_git'], cont2['sha256'], | (cont2['sha1'], cont2['sha1_git'], cont2['sha256'], | ||||
cont2['blake2s256'], cont2['length'], 'absent', | cont2['blake2s256'], cont2['length'], 'absent', | ||||
'Content too long') | 'Content too long') | ||||
) | ) | ||||
def test_skipped_content_add(self): | |||||
cont = self.skipped_cont.copy() | |||||
cont2 = self.skipped_cont2.copy() | |||||
cont2['blake2s256'] = None | |||||
missing = list(self.storage.skipped_content_missing([cont, cont2])) | |||||
self.assertEqual(len(missing), 2, missing) | |||||
actual_result = self.storage.content_add([cont, cont, cont2]) | |||||
self.assertEqual(actual_result, { | |||||
'content:add': 0, | |||||
'content:bytes:add': 0, | |||||
'skipped_content:add': 2, | |||||
}) | |||||
missing = list(self.storage.skipped_content_missing([cont, cont2])) | |||||
self.assertEqual(missing, []) | |||||
@pytest.mark.property_based | @pytest.mark.property_based | ||||
@given(strategies.sets( | @given(strategies.sets( | ||||
elements=strategies.sampled_from( | elements=strategies.sampled_from( | ||||
['sha256', 'sha1_git', 'blake2s256']), | ['sha256', 'sha1_git', 'blake2s256']), | ||||
min_size=0)) | min_size=0)) | ||||
def test_content_missing(self, algos): | def test_content_missing(self, algos): | ||||
algos |= {'sha1'} | algos |= {'sha1'} | ||||
cont2 = self.cont2 | cont2 = self.cont2 | ||||
▲ Show 20 Lines • Show All 2,358 Lines • Show Last 20 Lines |