Differential D2051 Diff 6902 swh/objstorage/tests/test_objstorage_azure.py

Changeset View

Standalone View

swh/objstorage/tests/test_objstorage_azure.py

# Copyright (C) 2016-2018 The Software Heritage developers		# Copyright (C) 2016-2018 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 collections import defaultdict		from collections import defaultdict
from unittest.mock import patch		from unittest.mock import patch

from typing import Any, Dict, Optional		from typing import Any, Dict

from azure.common import AzureMissingResourceHttpError		from azure.common import AzureMissingResourceHttpError
from swh.model.hashutil import hash_to_hex		from swh.model.hashutil import hash_to_hex

from swh.objstorage import get_objstorage		from swh.objstorage import get_objstorage
from swh.objstorage.objstorage import decompressors		from swh.objstorage.objstorage import decompressors
from swh.objstorage.exc import Error		from swh.objstorage.exc import Error

▲ Show 20 Lines • Show All 45 Lines • ▼ Show 20 Lines	class MockBlockBlobService():

def list_blobs(self, container_name, marker=None, maxresults=None):		def list_blobs(self, container_name, marker=None, maxresults=None):
for blob_name, content in sorted(self._data[container_name].items()):		for blob_name, content in sorted(self._data[container_name].items()):
if marker is None or blob_name > marker:		if marker is None or blob_name > marker:
yield MockBlob(name=blob_name, content=content)		yield MockBlob(name=blob_name, content=content)


class TestAzureCloudObjStorage(ObjStorageTestFixture, unittest.TestCase):		class TestAzureCloudObjStorage(ObjStorageTestFixture, unittest.TestCase):
compression = None # type: Optional[str]		compression = 'none'

def setUp(self):		def setUp(self):
super().setUp()		super().setUp()
patcher = patch(		patcher = patch(
'swh.objstorage.backends.azure.BlockBlobService',		'swh.objstorage.backends.azure.BlockBlobService',
MockBlockBlobService,		MockBlockBlobService,
)		)
patcher.start()		patcher.start()
Show All 23 Lines	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)

blob_service, container = self.storage.get_blob_service(obj_id)		blob_service, container = self.storage.get_blob_service(obj_id)
internal_id = self.storage._internal_id(obj_id)		internal_id = self.storage._internal_id(obj_id)

blob_service._data[container][internal_id] += b'trailing garbage'		blob_service._data[container][internal_id] += b'trailing garbage'

if self.compression is not None:		if self.compression == 'none':
		douarddaUnsubmitted Done Inline Actions I tend to prefer the `if a == b` when both the if and else blocks are there. douardda: I tend to prefer the `if a == b` when both the if and else blocks are there.
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 TestAzureCloudObjStorageGzip(TestAzureCloudObjStorage):		class TestAzureCloudObjStorageGzip(TestAzureCloudObjStorage):
compression = 'gzip'		compression = 'gzip'


class TestAzureCloudObjStorageZlib(TestAzureCloudObjStorage):		class TestAzureCloudObjStorageZlib(TestAzureCloudObjStorage):
compression = 'zlib'		compression = 'zlib'
▲ Show 20 Lines • Show All 60 Lines • Show Last 20 Lines