Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/tests/objstorage_testing.py
| Show First 20 Lines • Show All 56 Lines • ▼ Show 20 Lines | def test_contains(self): | ||||
| content_p, obj_id_p = self.hash_content(b"contains_present") | content_p, obj_id_p = self.hash_content(b"contains_present") | ||||
| content_m, obj_id_m = self.hash_content(b"contains_missing") | content_m, obj_id_m = self.hash_content(b"contains_missing") | ||||
| self.storage.add(content_p, obj_id=obj_id_p) | self.storage.add(content_p, obj_id=obj_id_p) | ||||
| self.assertIn(obj_id_p, self.storage) | self.assertIn(obj_id_p, self.storage) | ||||
| self.assertNotIn(obj_id_m, self.storage) | self.assertNotIn(obj_id_m, self.storage) | ||||
| def test_add_get_w_id(self): | def test_add_get_w_id(self): | ||||
| content, obj_id = self.hash_content(b"add_get_w_id") | content, obj_id = self.hash_content(b"add_get_w_id") | ||||
| r = self.storage.add(content, obj_id=obj_id) | self.storage.add(content, obj_id=obj_id) | ||||
| self.assertEqual(obj_id, r) | |||||
| self.assertContentMatch(obj_id, content) | self.assertContentMatch(obj_id, content) | ||||
| def test_add_twice(self): | def test_add_twice(self): | ||||
| content, obj_id = self.hash_content(b"add_twice") | content, obj_id = self.hash_content(b"add_twice") | ||||
| r = self.storage.add(content, obj_id=obj_id) | self.storage.add(content, obj_id=obj_id) | ||||
| self.assertEqual(obj_id, r) | |||||
| self.assertContentMatch(obj_id, content) | self.assertContentMatch(obj_id, content) | ||||
| r = self.storage.add(content, obj_id=obj_id, check_presence=False) | self.storage.add(content, obj_id=obj_id, check_presence=False) | ||||
| self.assertEqual(obj_id, r) | |||||
| self.assertContentMatch(obj_id, content) | self.assertContentMatch(obj_id, content) | ||||
| def test_add_big(self): | def test_add_big(self): | ||||
| content, obj_id = self.hash_content(b"add_big" * 1024 * 1024) | content, obj_id = self.hash_content(b"add_big" * 1024 * 1024) | ||||
| r = self.storage.add(content, obj_id=obj_id) | self.storage.add(content, obj_id=obj_id) | ||||
| self.assertEqual(obj_id, r) | |||||
| self.assertContentMatch(obj_id, content) | self.assertContentMatch(obj_id, content) | ||||
| def test_add_get_batch(self): | def test_add_get_batch(self): | ||||
| content1, obj_id1 = self.hash_content(b"add_get_batch_1") | content1, obj_id1 = self.hash_content(b"add_get_batch_1") | ||||
| content2, obj_id2 = self.hash_content(b"add_get_batch_2") | content2, obj_id2 = self.hash_content(b"add_get_batch_2") | ||||
| self.storage.add(content1, obj_id1) | self.storage.add(content1, obj_id1) | ||||
| self.storage.add(content2, obj_id2) | self.storage.add(content2, obj_id2) | ||||
| cr1, cr2 = self.storage.get_batch([obj_id1, obj_id2]) | cr1, cr2 = self.storage.get_batch([obj_id1, obj_id2]) | ||||
| self.assertEqual(cr1, content1) | self.assertEqual(cr1, content1) | ||||
| self.assertEqual(cr2, content2) | self.assertEqual(cr2, content2) | ||||
| def test_get_batch_unexisting_content(self): | def test_get_batch_unexisting_content(self): | ||||
| content, obj_id = self.hash_content(b"get_batch_unexisting_content") | content, obj_id = self.hash_content(b"get_batch_unexisting_content") | ||||
| result = list(self.storage.get_batch([obj_id])) | result = list(self.storage.get_batch([obj_id])) | ||||
| self.assertTrue(len(result) == 1) | self.assertTrue(len(result) == 1) | ||||
| self.assertIsNone(result[0]) | self.assertIsNone(result[0]) | ||||
| def test_restore_content(self): | def test_restore_content(self): | ||||
| self.storage.allow_delete = True | self.storage.allow_delete = True | ||||
| valid_content, valid_obj_id = self.hash_content(b"restore_content") | valid_content, valid_obj_id = self.hash_content(b"restore_content") | ||||
| invalid_content = b"unexpected content" | invalid_content = b"unexpected content" | ||||
| id_adding = self.storage.add(invalid_content, valid_obj_id) | self.storage.add(invalid_content, valid_obj_id) | ||||
| self.assertEqual(id_adding, valid_obj_id) | |||||
| with self.assertRaises(exc.Error): | with self.assertRaises(exc.Error): | ||||
| self.storage.check(id_adding) | self.storage.check(valid_obj_id) | ||||
| id_restore = self.storage.restore(valid_content, valid_obj_id) | self.storage.restore(valid_content, valid_obj_id) | ||||
| self.assertEqual(id_restore, valid_obj_id) | |||||
| self.assertContentMatch(valid_obj_id, valid_content) | self.assertContentMatch(valid_obj_id, valid_content) | ||||
| def test_get_missing(self): | def test_get_missing(self): | ||||
| content, obj_id = self.hash_content(b"get_missing") | content, obj_id = self.hash_content(b"get_missing") | ||||
| with self.assertRaises(exc.ObjNotFoundError) as e: | with self.assertRaises(exc.ObjNotFoundError) as e: | ||||
| self.storage.get(obj_id) | self.storage.get(obj_id) | ||||
| self.assertIn(obj_id, e.exception.args) | self.assertIn(obj_id, e.exception.args) | ||||
| ▲ Show 20 Lines • Show All 102 Lines • Show Last 20 Lines | |||||