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_pgsql(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_pgsql(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,344 Lines • Show Last 20 Lines | |||||