Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 552 Lines • ▼ Show 20 Lines | def test_check_config(self): | ||||
| self.assertTrue(self.storage.check_config(check_write=False)) | self.assertTrue(self.storage.check_config(check_write=False)) | ||||
| def test_content_add(self): | def test_content_add(self): | ||||
| cont = self.cont | cont = self.cont | ||||
| actual_result = self.storage.content_add([cont]) | actual_result = self.storage.content_add([cont]) | ||||
| self.assertEqual(actual_result, { | self.assertEqual(actual_result, { | ||||
| 'content:add': 1, | 'content:add': 1, | ||||
| 'content:bytes:add': cont['length'], | 'content:add:bytes': cont['length'], | ||||
| 'skipped_content:add': 0 | 'skipped_content:add': 0 | ||||
| }) | }) | ||||
| self.assertEqual(list(self.storage.content_get([cont['sha1']])), | self.assertEqual(list(self.storage.content_get([cont['sha1']])), | ||||
| [{'sha1': cont['sha1'], 'data': cont['data']}]) | [{'sha1': cont['sha1'], 'data': cont['data']}]) | ||||
| expected_cont = cont.copy() | expected_cont = cont.copy() | ||||
| del expected_cont['data'] | del expected_cont['data'] | ||||
| journal_objects = list(self.journal_writer.objects) | journal_objects = list(self.journal_writer.objects) | ||||
| for (obj_type, obj) in journal_objects: | for (obj_type, obj) in journal_objects: | ||||
| if 'ctime' in obj: | if 'ctime' in obj: | ||||
| del obj['ctime'] | del obj['ctime'] | ||||
| self.assertEqual(journal_objects, | self.assertEqual(journal_objects, | ||||
| [('content', expected_cont)]) | [('content', expected_cont)]) | ||||
| def test_content_add_same_input(self): | def test_content_add_same_input(self): | ||||
| cont = self.cont | cont = self.cont | ||||
| actual_result = self.storage.content_add([cont, cont]) | actual_result = self.storage.content_add([cont, cont]) | ||||
| self.assertEqual(actual_result, { | self.assertEqual(actual_result, { | ||||
| 'content:add': 1, | 'content:add': 1, | ||||
| 'content:bytes:add': cont['length'], | 'content:add:bytes': cont['length'], | ||||
| 'skipped_content:add': 0 | 'skipped_content:add': 0 | ||||
| }) | }) | ||||
| def test_content_add_different_input(self): | def test_content_add_different_input(self): | ||||
| cont = self.cont | cont = self.cont | ||||
| cont2 = self.cont2 | cont2 = self.cont2 | ||||
| actual_result = self.storage.content_add([cont, cont2]) | actual_result = self.storage.content_add([cont, cont2]) | ||||
| self.assertEqual(actual_result, { | self.assertEqual(actual_result, { | ||||
| 'content:add': 2, | 'content:add': 2, | ||||
| 'content:bytes:add': cont['length'] + cont2['length'], | 'content:add:bytes': cont['length'] + cont2['length'], | ||||
| 'skipped_content:add': 0 | 'skipped_content:add': 0 | ||||
| }) | }) | ||||
| def test_content_add_db(self): | def test_content_add_db(self): | ||||
| cont = self.cont | cont = self.cont | ||||
| actual_result = self.storage.content_add([cont]) | actual_result = self.storage.content_add([cont]) | ||||
| self.assertEqual(actual_result, { | self.assertEqual(actual_result, { | ||||
| 'content:add': 1, | 'content:add': 1, | ||||
| 'content:bytes:add': cont['length'], | 'content:add:bytes': cont['length'], | ||||
| 'skipped_content:add': 0 | 'skipped_content:add': 0 | ||||
| }) | }) | ||||
| if hasattr(self.storage, 'objstorage'): | if hasattr(self.storage, 'objstorage'): | ||||
| self.assertIn(cont['sha1'], self.storage.objstorage) | self.assertIn(cont['sha1'], self.storage.objstorage) | ||||
| self.cursor.execute('SELECT sha1, sha1_git, sha256, length, status' | self.cursor.execute('SELECT sha1, sha1_git, sha256, length, status' | ||||
| ' FROM content WHERE sha1 = %s', | ' FROM content WHERE sha1 = %s', | ||||
| (cont['sha1'],)) | (cont['sha1'],)) | ||||
| ▲ Show 20 Lines • Show All 119 Lines • ▼ Show 20 Lines | 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, | ||||
| 'content:bytes:add': 0, | 'content:add:bytes': 0, | ||||
| 'skipped_content:add': 2, | 'skipped_content:add': 2, | ||||
| }) | }) | ||||
| self.cursor.execute('SELECT sha1, sha1_git, sha256, blake2s256, ' | self.cursor.execute('SELECT sha1, sha1_git, sha256, blake2s256, ' | ||||
| 'length, status, reason ' | 'length, status, reason ' | ||||
| 'FROM skipped_content ORDER BY sha1_git') | 'FROM skipped_content ORDER BY sha1_git') | ||||
| datums = self.cursor.fetchall() | datums = self.cursor.fetchall() | ||||
| Show All 25 Lines | def test_skipped_content_add(self): | ||||
| missing = list(self.storage.skipped_content_missing([cont, cont2])) | missing = list(self.storage.skipped_content_missing([cont, cont2])) | ||||
| self.assertEqual(len(missing), 2, missing) | self.assertEqual(len(missing), 2, missing) | ||||
| 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, | ||||
| 'content:bytes:add': 0, | 'content:add:bytes': 0, | ||||
| 'skipped_content:add': 2, | 'skipped_content:add': 2, | ||||
| }) | }) | ||||
| missing = list(self.storage.skipped_content_missing([cont, cont2])) | missing = list(self.storage.skipped_content_missing([cont, cont2])) | ||||
| self.assertEqual(missing, []) | self.assertEqual(missing, []) | ||||
| @pytest.mark.property_based | @pytest.mark.property_based | ||||
| ▲ Show 20 Lines • Show All 2,421 Lines • Show Last 20 Lines | |||||