Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/tests/objstorage_testing.py
| # Copyright (C) 2015-2020 The Software Heritage developers | # Copyright (C) 2015-2020 The Software Heritage developers | ||||
| # See the AUTHORS file at the top-level directory of this distribution | # See the AUTHORS file at the top-level directory of this distribution | ||||
| # License: GNU General Public License version 3, or any later version | # License: GNU General Public License version 3, or any later version | ||||
| # See top-level LICENSE file for more information | # See top-level LICENSE file for more information | ||||
| import inspect | import inspect | ||||
| from typing import Tuple | |||||
| from swh.objstorage import exc | from swh.objstorage import exc | ||||
| from swh.objstorage.interface import ObjStorageInterface | from swh.objstorage.interface import CompositeObjId, ObjStorageInterface | ||||
| from swh.objstorage.objstorage import compute_hash | from swh.objstorage.objstorage import compute_hash | ||||
| class ObjStorageTestFixture: | class ObjStorageTestFixture: | ||||
| def test_types(self): | def test_types(self): | ||||
| """Checks all methods of ObjStorageInterface are implemented by this | """Checks all methods of ObjStorageInterface are implemented by this | ||||
| backend, and that they have the same signature.""" | backend, and that they have the same signature.""" | ||||
| # Create an instance of the protocol (which cannot be instantiated | # Create an instance of the protocol (which cannot be instantiated | ||||
| Show All 25 Lines | def test_types(self): | ||||
| # And we could replace the assertions above by this one, but unlike | # And we could replace the assertions above by this one, but unlike | ||||
| # the assertions above, it doesn't explain what is missing. | # the assertions above, it doesn't explain what is missing. | ||||
| assert isinstance(self.storage, ObjStorageInterface) | assert isinstance(self.storage, ObjStorageInterface) | ||||
| def hash_content(self, content): | def hash_content(self, content): | ||||
| obj_id = compute_hash(content) | obj_id = compute_hash(content) | ||||
| return content, obj_id | return content, obj_id | ||||
| def compositehash_content(self, content) -> Tuple[bytes, CompositeObjId]: | |||||
| obj_id = compute_hash(content) | |||||
| return content, {"sha1": obj_id} | |||||
| def assertContentMatch(self, obj_id, expected_content): # noqa | def assertContentMatch(self, obj_id, expected_content): # noqa | ||||
| content = self.storage.get(obj_id) | content = self.storage.get(obj_id) | ||||
| self.assertEqual(content, expected_content) | self.assertEqual(content, expected_content) | ||||
| def test_check_config(self): | def test_check_config(self): | ||||
| self.assertTrue(self.storage.check_config(check_write=False)) | self.assertTrue(self.storage.check_config(check_write=False)) | ||||
| self.assertTrue(self.storage.check_config(check_write=True)) | self.assertTrue(self.storage.check_config(check_write=True)) | ||||
| def test_contains(self): | 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_contains_composite(self): | |||||
| content_p, obj_id_p = self.compositehash_content(b"contains_present") | |||||
| content_m, obj_id_m = self.compositehash_content(b"contains_missing") | |||||
| self.storage.add(content_p, obj_id=obj_id_p) | |||||
| self.assertIn(obj_id_p, 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") | ||||
| self.storage.add(content, obj_id=obj_id) | self.storage.add(content, obj_id=obj_id) | ||||
| self.assertContentMatch(obj_id, content) | self.assertContentMatch(obj_id, content) | ||||
| def test_add_get_w_composite_id(self): | |||||
| content, obj_id = self.compositehash_content(b"add_get_w_id") | |||||
| self.storage.add(content, obj_id=obj_id) | |||||
| 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") | ||||
| self.storage.add(content, obj_id=obj_id) | self.storage.add(content, obj_id=obj_id) | ||||
| self.assertContentMatch(obj_id, content) | self.assertContentMatch(obj_id, content) | ||||
| self.storage.add(content, obj_id=obj_id, check_presence=False) | self.storage.add(content, obj_id=obj_id, check_presence=False) | ||||
| 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) | ||||
| self.storage.add(content, obj_id=obj_id) | self.storage.add(content, obj_id=obj_id) | ||||
| 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_add_get_batch_composite(self): | |||||
| content1, obj_id1 = self.compositehash_content(b"add_get_batch_1") | |||||
| content2, obj_id2 = self.compositehash_content(b"add_get_batch_2") | |||||
| self.storage.add(content1, obj_id1) | |||||
| self.storage.add(content2, obj_id2) | |||||
| cr1, cr2 = self.storage.get_batch([obj_id1, obj_id2]) | |||||
| self.assertEqual(cr1, content1) | |||||
| 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" | ||||
| self.storage.add(invalid_content, valid_obj_id) | self.storage.add(invalid_content, valid_obj_id) | ||||
| with self.assertRaises(exc.Error): | with self.assertRaises(exc.Error): | ||||
| self.storage.check(valid_obj_id) | self.storage.check(valid_obj_id) | ||||
| self.storage.restore(valid_content, valid_obj_id) | self.storage.restore(valid_content, 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) | ||||
| def test_get_missing_composite(self): | |||||
| content, obj_id = self.compositehash_content(b"get_missing") | |||||
| with self.assertRaises(exc.ObjNotFoundError) as e: | |||||
| self.storage.get(obj_id) | |||||
| self.assertIn(obj_id, e.exception.args) | |||||
| def test_check_missing(self): | def test_check_missing(self): | ||||
| content, obj_id = self.hash_content(b"check_missing") | content, obj_id = self.hash_content(b"check_missing") | ||||
| with self.assertRaises(exc.Error): | with self.assertRaises(exc.Error): | ||||
| self.storage.check(obj_id) | self.storage.check(obj_id) | ||||
| def test_check_missing_composite(self): | |||||
| content, obj_id = self.compositehash_content(b"check_missing") | |||||
| with self.assertRaises(exc.Error): | |||||
| self.storage.check(obj_id) | |||||
| def test_check_present(self): | def test_check_present(self): | ||||
| content, obj_id = self.hash_content(b"check_present") | content, obj_id = self.hash_content(b"check_present") | ||||
| self.storage.add(content, obj_id) | self.storage.add(content, obj_id) | ||||
| try: | try: | ||||
| self.storage.check(obj_id) | self.storage.check(obj_id) | ||||
| except exc.Error: | except exc.Error: | ||||
| self.fail("Integrity check failed") | self.fail("Integrity check failed") | ||||
| def test_check_present_composite(self): | |||||
| content, obj_id = self.compositehash_content(b"check_present") | |||||
| self.storage.add(content, obj_id) | |||||
| try: | |||||
| self.storage.check(obj_id) | |||||
| except exc.Error: | |||||
| self.fail("Integrity check failed") | |||||
| def test_delete_missing(self): | def test_delete_missing(self): | ||||
| self.storage.allow_delete = True | self.storage.allow_delete = True | ||||
| content, obj_id = self.hash_content(b"missing_content_to_delete") | content, obj_id = self.hash_content(b"missing_content_to_delete") | ||||
| with self.assertRaises(exc.Error): | with self.assertRaises(exc.Error): | ||||
| self.storage.delete(obj_id) | self.storage.delete(obj_id) | ||||
| def test_delete_missing_composite(self): | |||||
| self.storage.allow_delete = True | |||||
| content, obj_id = self.compositehash_content(b"missing_content_to_delete") | |||||
| with self.assertRaises(exc.Error): | |||||
| self.storage.delete(obj_id) | |||||
| def test_delete_present(self): | def test_delete_present(self): | ||||
| self.storage.allow_delete = True | self.storage.allow_delete = True | ||||
| content, obj_id = self.hash_content(b"content_to_delete") | content, obj_id = self.hash_content(b"content_to_delete") | ||||
| self.storage.add(content, obj_id=obj_id) | self.storage.add(content, obj_id=obj_id) | ||||
| self.assertTrue(self.storage.delete(obj_id)) | self.assertTrue(self.storage.delete(obj_id)) | ||||
| with self.assertRaises(exc.Error): | with self.assertRaises(exc.Error): | ||||
| self.storage.get(obj_id) | self.storage.get(obj_id) | ||||
| def test_delete_present_composite(self): | |||||
| self.storage.allow_delete = True | |||||
| content, obj_id = self.compositehash_content(b"content_to_delete") | |||||
| self.storage.add(content, obj_id=obj_id) | |||||
| self.assertTrue(self.storage.delete(obj_id)) | |||||
| with self.assertRaises(exc.Error): | |||||
| self.storage.get(obj_id) | |||||
| def test_delete_not_allowed(self): | def test_delete_not_allowed(self): | ||||
| self.storage.allow_delete = False | self.storage.allow_delete = False | ||||
| content, obj_id = self.hash_content(b"content_to_delete") | content, obj_id = self.hash_content(b"content_to_delete") | ||||
| self.storage.add(content, obj_id=obj_id) | self.storage.add(content, obj_id=obj_id) | ||||
| with self.assertRaises(PermissionError): | with self.assertRaises(PermissionError): | ||||
| self.storage.delete(obj_id) | self.storage.delete(obj_id) | ||||
| def test_delete_not_allowed_by_default(self): | def test_delete_not_allowed_by_default(self): | ||||
| ▲ Show 20 Lines • Show All 66 Lines • Show Last 20 Lines | |||||