Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/tests/test_objstorage_seaweedfs.py
| # Copyright (C) 2019 The Software Heritage developers | # Copyright (C) 2019 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 unittest | import unittest | ||||
| from typing import Optional | |||||
| from swh.objstorage.objstorage import decompressors | from swh.objstorage.objstorage import decompressors | ||||
| from swh.objstorage.exc import Error | from swh.objstorage.exc import Error | ||||
| from swh.objstorage.backends.seaweed import WeedObjStorage, DEFAULT_LIMIT | from swh.objstorage.backends.seaweed import WeedObjStorage, DEFAULT_LIMIT | ||||
| from swh.objstorage.tests.objstorage_testing import ObjStorageTestFixture | from swh.objstorage.tests.objstorage_testing import ObjStorageTestFixture | ||||
| class MockWeedFiler: | class MockWeedFiler: | ||||
| Show All 19 Lines | def list(self, dir, last_file_name=None, limit=DEFAULT_LIMIT): | ||||
| if last_file_name is None: | if last_file_name is None: | ||||
| idx = 0 | idx = 0 | ||||
| else: | else: | ||||
| idx = keys.index(last_file_name) + 1 | idx = keys.index(last_file_name) + 1 | ||||
| return {'Entries': [{'FullPath': x} for x in keys[idx:idx+limit]]} | return {'Entries': [{'FullPath': x} for x in keys[idx:idx+limit]]} | ||||
| class TestWeedObjStorage(ObjStorageTestFixture, unittest.TestCase): | class TestWeedObjStorage(ObjStorageTestFixture, unittest.TestCase): | ||||
| compression = None # type: Optional[str] | compression = 'none' | ||||
| def setUp(self): | def setUp(self): | ||||
| super().setUp() | super().setUp() | ||||
| self.url = 'http://127.0.0.1/test' | self.url = 'http://127.0.0.1/test' | ||||
| self.storage = WeedObjStorage(url=self.url, | self.storage = WeedObjStorage(url=self.url, | ||||
| compression=self.compression) | compression=self.compression) | ||||
| self.storage.wf = MockWeedFiler(self.url) | self.storage.wf = MockWeedFiler(self.url) | ||||
| Show All 9 Lines | class TestWeedObjStorage(ObjStorageTestFixture, unittest.TestCase): | ||||
| def test_trailing_data_on_stored_blob(self): | def test_trailing_data_on_stored_blob(self): | ||||
| content, obj_id = self.hash_content(b'test content without garbage') | content, obj_id = self.hash_content(b'test content without garbage') | ||||
| self.storage.add(content, obj_id=obj_id) | self.storage.add(content, obj_id=obj_id) | ||||
| path = self.storage._path(obj_id) | path = self.storage._path(obj_id) | ||||
| self.storage.wf.content[path] += b'trailing garbage' | self.storage.wf.content[path] += b'trailing garbage' | ||||
| if self.compression is not None: | if self.compression == 'none': | ||||
| with self.assertRaises(Error) as e: | with self.assertRaises(Error) as e: | ||||
| self.storage.get(obj_id) | self.storage.check(obj_id) | ||||
| assert 'trailing data' in e.exception.args[0] | |||||
| else: | else: | ||||
| with self.assertRaises(Error) as e: | with self.assertRaises(Error) as e: | ||||
| self.storage.check(obj_id) | self.storage.get(obj_id) | ||||
| assert 'trailing data' in e.exception.args[0] | |||||
| class TestWeedObjStorageBz2(TestWeedObjStorage): | class TestWeedObjStorageBz2(TestWeedObjStorage): | ||||
| compression = 'bz2' | compression = 'bz2' | ||||
| class TestWeedObjStorageGzip(TestWeedObjStorage): | class TestWeedObjStorageGzip(TestWeedObjStorage): | ||||
| compression = 'gzip' | compression = 'gzip' | ||||
| class TestWeedObjStorageLzma(TestWeedObjStorage): | class TestWeedObjStorageLzma(TestWeedObjStorage): | ||||
| compression = 'lzma' | compression = 'lzma' | ||||
| class TestWeedObjStorageZlib(TestWeedObjStorage): | class TestWeedObjStorageZlib(TestWeedObjStorage): | ||||
| compression = 'zlib' | compression = 'zlib' | ||||