diff --git a/swh/objstorage/tests/objstorage_testing.py b/swh/objstorage/tests/objstorage_testing.py --- a/swh/objstorage/tests/objstorage_testing.py +++ b/swh/objstorage/tests/objstorage_testing.py @@ -4,15 +4,13 @@ # See top-level LICENSE file for more information import time +import collections from swh.objstorage import exc from swh.objstorage.objstorage import compute_hash -class ObjStorageTestFixture(): - - def setUp(self): - super().setUp() +class ObjStorageTestFixture: def hash_content(self, content): obj_id = compute_hash(content) @@ -116,7 +114,7 @@ content, obj_id = self.hash_content(b'content_to_delete') self.storage.add(content, obj_id=obj_id) with self.assertRaises(PermissionError): - self.assertTrue(self.storage.delete(obj_id)) + self.storage.delete(obj_id) def test_delete_not_allowed_by_default(self): content, obj_id = self.hash_content(b'content_to_delete') @@ -137,24 +135,29 @@ def gen_content(): yield b'chunk1' time.sleep(0.5) - yield b'chunk2' + yield b'chunk42' _, obj_id = self.hash_content(b'placeholder_id') try: self.storage.add_stream(gen_content(), obj_id=obj_id) except NotImplementedError: return - self.assertContentMatch(obj_id, b'chunk1chunk2') + self.assertContentMatch(obj_id, b'chunk1chunk42') def test_get_stream(self): - content_l = [b'1', b'2', b'3', b'4', b'5', b'6', b'7', b'8', b'9'] - content = b''.join(content_l) + content = b'123456789' _, obj_id = self.hash_content(content) self.storage.add(content, obj_id=obj_id) + r = self.storage.get(obj_id) + self.assertEqual(r, content) + try: - r = list(self.storage.get_stream(obj_id, chunk_size=1)) + r = self.storage.get_stream(obj_id, chunk_size=1) except NotImplementedError: return - self.assertEqual(r, content_l) + self.assertTrue(isinstance(r, collections.Iterator)) + r = list(r) + self.assertEqual(len(r), 9) + self.assertEqual(b''.join(r), content) def test_add_batch(self): contents = {} diff --git a/swh/objstorage/tests/test_objstorage_api.py b/swh/objstorage/tests/test_objstorage_api.py --- a/swh/objstorage/tests/test_objstorage_api.py +++ b/swh/objstorage/tests/test_objstorage_api.py @@ -41,3 +41,9 @@ def tearDown(self): super().tearDown() shutil.rmtree(self.tmpdir) + + def test_delete_not_allowed(self): + self.skipTest('makes no sense to test this for the remote api') + + def test_delete_not_allowed_by_default(self): + self.skipTest('makes no sense to test this for the remote api')