Changeset View
Changeset View
Standalone View
Standalone View
swh/objstorage/tests/test_objstorage_cloud.py
# Copyright (C) 2016 The Software Heritage developers | # Copyright (C) 2016 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 libcloud.common.types import InvalidCredsError | from libcloud.common.types import InvalidCredsError | ||||
from libcloud.storage.types import (ContainerDoesNotExistError, | from libcloud.storage.types import (ContainerDoesNotExistError, | ||||
ObjectDoesNotExistError) | ObjectDoesNotExistError) | ||||
from typing import Optional | from typing import Optional | ||||
from swh.model import hashutil | from swh.model import hashutil | ||||
from swh.objstorage.objstorage import decompressors | from swh.objstorage.objstorage import decompressors | ||||
from swh.objstorage.exc import Error | |||||
from swh.objstorage.backends.libcloud import CloudObjStorage | from swh.objstorage.backends.libcloud import CloudObjStorage | ||||
from .objstorage_testing import ObjStorageTestFixture | from .objstorage_testing import ObjStorageTestFixture | ||||
API_KEY = 'API_KEY' | API_KEY = 'API_KEY' | ||||
API_SECRET_KEY = 'API SECRET KEY' | API_SECRET_KEY = 'API SECRET KEY' | ||||
CONTAINER_NAME = 'test_container' | CONTAINER_NAME = 'test_container' | ||||
▲ Show 20 Lines • Show All 86 Lines • ▼ Show 20 Lines | def test_compression(self): | ||||
data = self.storage.driver.containers[CONTAINER_NAME] | data = self.storage.driver.containers[CONTAINER_NAME] | ||||
obj_id = hashutil.hash_to_hex(obj_id) | obj_id = hashutil.hash_to_hex(obj_id) | ||||
raw_content = b''.join(data[obj_id].content) | raw_content = b''.join(data[obj_id].content) | ||||
d = decompressors[self.compression]() | d = decompressors[self.compression]() | ||||
assert d.decompress(raw_content) == content | assert d.decompress(raw_content) == content | ||||
assert d.unused_data == b'' | assert d.unused_data == b'' | ||||
def test_trailing_data_on_stored_blob(self): | |||||
content, obj_id = self.hash_content(b'test content without garbage') | |||||
self.storage.add(content, obj_id=obj_id) | |||||
data = self.storage.driver.containers[CONTAINER_NAME] | |||||
obj_id = hashutil.hash_to_hex(obj_id) | |||||
data[obj_id].content.append(b'trailing garbage') | |||||
if self.compression is not None: | |||||
with self.assertRaises(Error) as e: | |||||
self.storage.get(obj_id) | |||||
assert 'trailing data' in e.exception.args[0] | |||||
else: | |||||
with self.assertRaises(Error) as e: | |||||
self.storage.check(obj_id) | |||||
class TestCloudObjStorageBz2(TestCloudObjStorage): | class TestCloudObjStorageBz2(TestCloudObjStorage): | ||||
compression = 'bz2' | compression = 'bz2' | ||||
class TestCloudObjStorageGzip(TestCloudObjStorage): | class TestCloudObjStorageGzip(TestCloudObjStorage): | ||||
compression = 'gzip' | compression = 'gzip' | ||||
class TestCloudObjStorageLzma(TestCloudObjStorage): | class TestCloudObjStorageLzma(TestCloudObjStorage): | ||||
compression = 'lzma' | compression = 'lzma' | ||||
class TestCloudObjStorageZlib(TestCloudObjStorage): | class TestCloudObjStorageZlib(TestCloudObjStorage): | ||||
compression = 'zlib' | compression = 'zlib' |