"Fossies" - the Fresh Open Source Software Archive  

Source code changes of the file "glance/tests/unit/common/test_utils.py" between
glance-20.0.0.tar.gz and glance-20.0.1.tar.gz

About: OpenStack Glance (Core Service: Image Service) stores and retrieves virtual machine disk images. OpenStack Compute makes use of this during instance provisioning.
The "Ussuri" series (latest release).

test_utils.py  (glance-20.0.0)	:	test_utils.py  (glance-20.0.1)
skipping to change at line 36		skipping to change at line 36
from glance.common import exception		from glance.common import exception
from glance.common import store_utils		from glance.common import store_utils
from glance.common import utils		from glance.common import utils
from glance.tests import utils as test_utils		from glance.tests import utils as test_utils
CONF = cfg.CONF		CONF = cfg.CONF
class TestStoreUtils(test_utils.BaseTestCase):		class TestStoreUtils(test_utils.BaseTestCase):
"""Test glance.common.store_utils module"""		"""Test glance.common.store_utils module"""
def _test_update_store_in_location(self, metadata, store_id, expected):		def _test_update_store_in_location(self, metadata, store_id, expected,
		store_id_call_count=1,
		save_call_count=1):
		image = mock.Mock()
		image_repo = mock.Mock()
		image_repo.save = mock.Mock()
locations = [{		locations = [{
'url': 'rbd://aaaaaaaa/images/id',		'url': 'rbd://aaaaaaaa/images/id',
'metadata': metadata		'metadata': metadata
}]		}]
		image.locations = locations
with mock.patch.object(		with mock.patch.object(
store_utils, '_get_store_id_from_uri') as mock_get_store_id:		store_utils, '_get_store_id_from_uri') as mock_get_store_id:
mock_get_store_id.return_value = store_id		mock_get_store_id.return_value = store_id
store_utils.update_store_in_locations(locations, mock.Mock())		store_utils.update_store_in_locations(image, image_repo)
self.assertEqual(locations[0]['metadata'].get('store'), expected)		self.assertEqual(image.locations[0]['metadata'].get(
		'store'), expected)
		self.assertEqual(store_id_call_count, mock_get_store_id.call_count)
		self.assertEqual(save_call_count, image_repo.save.call_count)
def test_update_store_location_with_no_store(self):		def test_update_store_location_with_no_store(self):
		enabled_backends = {
		"rbd1": "rbd",
		"rbd2": "rbd"
		}
		self.config(enabled_backends=enabled_backends)
self._test_update_store_in_location({}, 'rbd1', 'rbd1')		self._test_update_store_in_location({}, 'rbd1', 'rbd1')
def test_update_store_location_with_different_store(self):		def test_update_store_location_with_different_store(self):
self._test_update_store_in_location({'store': 'rbd2'}, 'rbd1', 'rbd1')		enabled_backends = {
		"ceph1": "rbd",
		"ceph2": "rbd"
		}
		self.config(enabled_backends=enabled_backends)
		self._test_update_store_in_location(
		{'store': 'rbd2'}, 'ceph1', 'ceph1')
def test_update_store_location_with_same_store(self):		def test_update_store_location_with_same_store(self):
self._test_update_store_in_location({'store': 'rbd1'}, 'rbd1', 'rbd1')		enabled_backends = {
		"rbd1": "rbd",
		"rbd2": "rbd"
		}
		self.config(enabled_backends=enabled_backends)
		self._test_update_store_in_location({'store': 'rbd1'}, 'rbd1', 'rbd1',
		store_id_call_count=0,
		save_call_count=0)
def test_update_store_location_with_store_none(self):		def test_update_store_location_with_store_none(self):
self._test_update_store_in_location({}, None, None)		enabled_backends = {
		"rbd1": "rbd",
		"rbd2": "rbd"
		}
		self.config(enabled_backends=enabled_backends)
		self._test_update_store_in_location({}, None, None,
		save_call_count=0)
class TestUtils(test_utils.BaseTestCase):		class TestUtils(test_utils.BaseTestCase):
"""Test routines in glance.utils"""		"""Test routines in glance.utils"""
def test_cooperative_reader(self):		def test_cooperative_reader(self):
"""Ensure cooperative reader class accesses all bytes of file"""		"""Ensure cooperative reader class accesses all bytes of file"""
BYTES = 1024		BYTES = 1024
bytes_read = 0		bytes_read = 0
with tempfile.TemporaryFile('w+') as tmp_fd:		with tempfile.TemporaryFile('w+') as tmp_fd:
tmp_fd.write('*' * BYTES)		tmp_fd.write('*' * BYTES)
skipping to change at line 521		skipping to change at line 554
body = {"all_stores": True}		body = {"all_stores": True}
req = webob.Request.blank("/some_request")		req = webob.Request.blank("/some_request")
mp = "glance.common.utils.glance_store.get_store_from_store_identifier"		mp = "glance.common.utils.glance_store.get_store_from_store_identifier"
with mock.patch(mp) as mock_get_store:		with mock.patch(mp) as mock_get_store:
result = sorted(utils.get_stores_from_request(req, body))		result = sorted(utils.get_stores_from_request(req, body))
self.assertEqual(["ceph1", "ceph2"], result)		self.assertEqual(["ceph1", "ceph2"], result)
mock_get_store.assert_any_call("ceph1")		mock_get_store.assert_any_call("ceph1")
mock_get_store.assert_any_call("ceph2")		mock_get_store.assert_any_call("ceph2")
self.assertEqual(mock_get_store.call_count, 2)		self.assertEqual(mock_get_store.call_count, 2)
		def test_get_stores_from_request_excludes_readonly_store(self):
		enabled_backends = {
		"ceph1": "rbd",
		"ceph2": "rbd",
		"http": "http"
		}
		self.config(enabled_backends=enabled_backends)
		store.register_store_opts(CONF)
		self.config(default_backend="ceph1", group="glance_store")
		body = {"all_stores": True}
		req = webob.Request.blank("/some_request")
		mp = "glance.common.utils.glance_store.get_store_from_store_identifier"
		with mock.patch(mp) as mock_get_store:
		result = sorted(utils.get_stores_from_request(req, body))
		self.assertNotIn("http", result)
		self.assertEqual(["ceph1", "ceph2"], result)
		mock_get_store.assert_any_call("ceph1")
		mock_get_store.assert_any_call("ceph2")
		self.assertEqual(mock_get_store.call_count, 2)
def test_get_stores_from_request_raises_bad_request_with_all_stores(self):		def test_get_stores_from_request_raises_bad_request_with_all_stores(self):
enabled_backends = {		enabled_backends = {
"ceph1": "rbd",		"ceph1": "rbd",
"ceph2": "rbd"		"ceph2": "rbd"
}		}
self.config(enabled_backends=enabled_backends)		self.config(enabled_backends=enabled_backends)
store.register_store_opts(CONF)		store.register_store_opts(CONF)
self.config(default_backend="ceph1", group="glance_store")		self.config(default_backend="ceph1", group="glance_store")
headers = {"x-image-meta-store": "ceph2"}		headers = {"x-image-meta-store": "ceph2"}
body = {"stores": ["ceph1"], "all_stores": True}		body = {"stores": ["ceph1"], "all_stores": True}
End of changes. 8 change blocks.
6 lines changed or deleted		59 lines changed or added

---