Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/tests/test_storage.py
| Show First 20 Lines • Show All 566 Lines • ▼ Show 20 Lines | class CommonTestStorage(TestStorageData): | ||||
| def test_check_config(self): | def test_check_config(self): | ||||
| self.assertTrue(self.storage.check_config(check_write=True)) | self.assertTrue(self.storage.check_config(check_write=True)) | ||||
| 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 | ||||
| insertion_start_time = datetime.datetime.now(tz=datetime.timezone.utc) | |||||
| actual_result = self.storage.content_add([cont]) | actual_result = self.storage.content_add([cont]) | ||||
| insertion_end_time = datetime.datetime.now(tz=datetime.timezone.utc) | |||||
| self.assertEqual(actual_result, { | self.assertEqual(actual_result, { | ||||
| 'content:add': 1, | 'content:add': 1, | ||||
| 'content:add:bytes': 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: | self.assertLessEqual(insertion_start_time, obj['ctime']) | ||||
| self.assertLessEqual(obj['ctime'], insertion_end_time) | |||||
| del obj['ctime'] | del obj['ctime'] | ||||
| self.assertEqual(journal_objects, | self.assertEqual(journal_objects, | ||||
| [('content', expected_cont)]) | [('content', expected_cont)]) | ||||
| def test_content_get_missing(self): | def test_content_get_missing(self): | ||||
| cont = self.cont | cont = self.cont | ||||
| self.storage.content_add([cont]) | self.storage.content_add([cont]) | ||||
| ▲ Show 20 Lines • Show All 77 Lines • ▼ Show 20 Lines | def test_content_add_db(self): | ||||
| datum[3], datum[4]), | datum[3], datum[4]), | ||||
| (cont['sha1'], cont['sha1_git'], cont['sha256'], | (cont['sha1'], cont['sha1_git'], cont['sha256'], | ||||
| cont['length'], 'visible')) | cont['length'], 'visible')) | ||||
| 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: | |||||
| del obj['ctime'] | del obj['ctime'] | ||||
| self.assertEqual(journal_objects, | self.assertEqual(journal_objects, | ||||
| [('content', expected_cont)]) | [('content', expected_cont)]) | ||||
| def test_content_add_collision(self): | def test_content_add_collision(self): | ||||
| cont1 = self.cont | cont1 = self.cont | ||||
| # create (corrupted) content with same sha1{,_git} but != sha256 | # create (corrupted) content with same sha1{,_git} but != sha256 | ||||
| cont1b = cont1.copy() | cont1b = cont1.copy() | ||||
| ▲ Show 20 Lines • Show All 3,054 Lines • Show Last 20 Lines | |||||