# Copyright 2018-2019 RedHat Inc. # All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import contextlib import errno import hashlib import mock import os import six import socket import tempfile import time import uuid import fixtures from os_brick.initiator import connector from oslo_concurrency import processutils from oslo_config import cfg from oslo_utils import units import glance_store as store from glance_store._drivers import cinder from glance_store import exceptions from glance_store import location from glance_store.tests import base from glance_store.tests.unit import test_store_capabilities as test_cap class FakeObject(object): def __init__(self, **kwargs): for name, value in kwargs.items(): setattr(self, name, value) class TestMultiCinderStore(base.MultiStoreBaseTest, test_cap.TestStoreCapabilitiesChecking): # NOTE(flaper87): temporary until we # can move to a fully-local lib. # (Swift store's fault) _CONF = cfg.ConfigOpts() def setUp(self): super(TestMultiCinderStore, self).setUp() enabled_backends = { "cinder1": "cinder", "cinder2": "cinder" } self.conf = self._CONF self.conf(args=[]) self.conf.register_opt(cfg.DictOpt('enabled_backends')) self.config(enabled_backends=enabled_backends) store.register_store_opts(self.conf) self.config(default_backend='cinder1', group='glance_store') # Ensure stores + locations cleared location.SCHEME_TO_CLS_BACKEND_MAP = {} store.create_multi_stores(self.conf) self.addCleanup(setattr, location, 'SCHEME_TO_CLS_BACKEND_MAP', dict()) self.test_dir = self.useFixture(fixtures.TempDir()).path self.addCleanup(self.conf.reset) self.store = cinder.Store(self.conf, backend="cinder1") self.store.configure() self.register_store_backend_schemes(self.store, 'cinder', 'cinder1') self.store.READ_CHUNKSIZE = 4096 self.store.WRITE_CHUNKSIZE = 4096 fake_sc = [{u'endpoints': [{u'publicURL': u'http://foo/public_url'}], u'endpoints_links': [], u'name': u'cinder', u'type': u'volumev2'}] self.context = FakeObject(service_catalog=fake_sc, user='fake_user', auth_token='fake_token', tenant='fake_tenant') def test_get_cinderclient(self): cc = cinder.get_cinderclient(self.conf, self.context, backend="cinder1") self.assertEqual('fake_token', cc.client.auth_token) self.assertEqual('http://foo/public_url', cc.client.management_url) def test_get_cinderclient_with_user_overriden(self): self.config(cinder_store_user_name='test_user', group="cinder1") self.config(cinder_store_password='test_password', group="cinder1") self.config(cinder_store_project_name='test_project', group="cinder1") self.config(cinder_store_auth_address='test_address', group="cinder1") cc = cinder.get_cinderclient(self.conf, self.context, backend="cinder1") self.assertIsNone(cc.client.auth_token) self.assertEqual('test_address', cc.client.management_url) def test_temporary_chown(self): class fake_stat(object): st_uid = 1 with mock.patch.object(os, 'stat', return_value=fake_stat()), \ mock.patch.object(os, 'getuid', return_value=2), \ mock.patch.object(processutils, 'execute') as mock_execute, \ mock.patch.object(cinder, 'get_root_helper', return_value='sudo'): with cinder.temporary_chown('test'): pass expected_calls = [mock.call('chown', 2, 'test', run_as_root=True, root_helper='sudo'), mock.call('chown', 1, 'test', run_as_root=True, root_helper='sudo')] self.assertEqual(expected_calls, mock_execute.call_args_list) @mock.patch.object(time, 'sleep') def test_wait_volume_status(self, mock_sleep): fake_manager = FakeObject(get=mock.Mock()) volume_available = FakeObject(manager=fake_manager, id='fake-id', status='available') volume_in_use = FakeObject(manager=fake_manager, id='fake-id', status='in-use') fake_manager.get.side_effect = [volume_available, volume_in_use] self.assertEqual(volume_in_use, self.store._wait_volume_status( volume_available, 'available', 'in-use')) fake_manager.get.assert_called_with('fake-id') mock_sleep.assert_called_once_with(0.5) @mock.patch.object(time, 'sleep') def test_wait_volume_status_unexpected(self, mock_sleep): fake_manager = FakeObject(get=mock.Mock()) volume_available = FakeObject(manager=fake_manager, id='fake-id', status='error') fake_manager.get.return_value = volume_available self.assertRaises(exceptions.BackendException, self.store._wait_volume_status, volume_available, 'available', 'in-use') fake_manager.get.assert_called_with('fake-id') @mock.patch.object(time, 'sleep') def test_wait_volume_status_timeout(self, mock_sleep): fake_manager = FakeObject(get=mock.Mock()) volume_available = FakeObject(manager=fake_manager, id='fake-id', status='available') fake_manager.get.return_value = volume_available self.assertRaises(exceptions.BackendException, self.store._wait_volume_status, volume_available, 'available', 'in-use') fake_manager.get.assert_called_with('fake-id') def _test_open_cinder_volume(self, open_mode, attach_mode, error): fake_volume = mock.MagicMock(id=str(uuid.uuid4()), status='available') fake_volumes = FakeObject(get=lambda id: fake_volume, detach=mock.Mock()) fake_client = FakeObject(volumes=fake_volumes) _, fake_dev_path = tempfile.mkstemp(dir=self.test_dir) fake_devinfo = {'path': fake_dev_path} fake_connector = FakeObject( connect_volume=mock.Mock(return_value=fake_devinfo), disconnect_volume=mock.Mock()) @contextlib.contextmanager def fake_chown(path, backend=None): yield def do_open(): with self.store._open_cinder_volume( fake_client, fake_volume, open_mode): if error: raise error def fake_factory(protocol, root_helper, **kwargs): self.assertEqual(fake_volume.initialize_connection.return_value, kwargs['conn']) return fake_connector root_helper = "sudo glance-rootwrap /etc/glance/rootwrap.conf" with mock.patch.object(cinder.Store, '_wait_volume_status', return_value=fake_volume), \ mock.patch.object(cinder, 'temporary_chown', side_effect=fake_chown), \ mock.patch.object(cinder, 'get_root_helper', return_value=root_helper), \ mock.patch.object(connector, 'get_connector_properties'), \ mock.patch.object(connector.InitiatorConnector, 'factory', side_effect=fake_factory): if error: self.assertRaises(error, do_open) else: do_open() fake_connector.connect_volume.assert_called_once_with(mock.ANY) fake_connector.disconnect_volume.assert_called_once_with( mock.ANY, fake_devinfo) fake_volume.attach.assert_called_once_with( None, 'glance_store', attach_mode, host_name=socket.gethostname()) fake_volumes.detach.assert_called_once_with(fake_volume) def test_open_cinder_volume_rw(self): self._test_open_cinder_volume('wb', 'rw', None) def test_open_cinder_volume_ro(self): self._test_open_cinder_volume('rb', 'ro', None) def test_open_cinder_volume_error(self): self._test_open_cinder_volume('wb', 'rw', IOError) def test_cinder_configure_add(self): self.assertRaises(exceptions.BadStoreConfiguration, self.store._check_context, None) self.assertRaises(exceptions.BadStoreConfiguration, self.store._check_context, FakeObject(service_catalog=None)) self.store._check_context(FakeObject(service_catalog='fake')) def test_cinder_get(self): expected_size = 5 * units.Ki expected_file_contents = b"*" * expected_size volume_file = six.BytesIO(expected_file_contents) fake_client = FakeObject(auth_token=None, management_url=None) fake_volume_uuid = str(uuid.uuid4()) fake_volume = mock.MagicMock(id=fake_volume_uuid, metadata={'image_size': expected_size}, status='available') fake_volume.manager.get.return_value = fake_volume fake_volumes = FakeObject(get=lambda id: fake_volume) @contextlib.contextmanager def fake_open(client, volume, mode): self.assertEqual('rb', mode) yield volume_file with mock.patch.object(cinder, 'get_cinderclient') as mock_cc, \ mock.patch.object(self.store, '_open_cinder_volume', side_effect=fake_open): mock_cc.return_value = FakeObject(client=fake_client, volumes=fake_volumes) uri = "cinder://%s" % fake_volume_uuid loc = location.get_location_from_uri_and_backend(uri, "cinder1", conf=self.conf) (image_file, image_size) = self.store.get(loc, context=self.context) expected_num_chunks = 2 data = b"" num_chunks = 0 for chunk in image_file: num_chunks += 1 data += chunk self.assertEqual(expected_num_chunks, num_chunks) self.assertEqual(expected_file_contents, data) def test_cinder_get_size(self): fake_client = FakeObject(auth_token=None, management_url=None) fake_volume_uuid = str(uuid.uuid4()) fake_volume = FakeObject(size=5, metadata={}) fake_volumes = {fake_volume_uuid: fake_volume} with mock.patch.object(cinder, 'get_cinderclient') as mocked_cc: mocked_cc.return_value = FakeObject(client=fake_client, volumes=fake_volumes) uri = 'cinder://%s' % fake_volume_uuid loc = location.get_location_from_uri_and_backend(uri, "cinder1", conf=self.conf) image_size = self.store.get_size(loc, context=self.context) self.assertEqual(fake_volume.size * units.Gi, image_size) def test_cinder_get_size_with_metadata(self): fake_client = FakeObject(auth_token=None, management_url=None) fake_volume_uuid = str(uuid.uuid4()) expected_image_size = 4500 * units.Mi fake_volume = FakeObject(size=5, metadata={'image_size': expected_image_size}) fake_volumes = {fake_volume_uuid: fake_volume} with mock.patch.object(cinder, 'get_cinderclient') as mocked_cc: mocked_cc.return_value = FakeObject(client=fake_client, volumes=fake_volumes) uri = 'cinder://%s' % fake_volume_uuid loc = location.get_location_from_uri_and_backend(uri, "cinder1", conf=self.conf) image_size = self.store.get_size(loc, context=self.context) self.assertEqual(expected_image_size, image_size) def _test_cinder_add(self, fake_volume, volume_file, size_kb=5, verifier=None, backend="cinder1"): expected_image_id = str(uuid.uuid4()) expected_size = size_kb * units.Ki expected_file_contents = b"*" * expected_size image_file = six.BytesIO(expected_file_contents) expected_checksum = hashlib.md5(expected_file_contents).hexdigest() expected_location = 'cinder://%s' % fake_volume.id fake_client = FakeObject(auth_token=None, management_url=None) fake_volume.manager.get.return_value = fake_volume fake_volumes = FakeObject(create=mock.Mock(return_value=fake_volume)) self.config(cinder_volume_type='some_type', group=backend) @contextlib.contextmanager def fake_open(client, volume, mode): self.assertEqual('wb', mode) yield volume_file with mock.patch.object(cinder, 'get_cinderclient') as mock_cc, \ mock.patch.object(self.store, '_open_cinder_volume', side_effect=fake_open): mock_cc.return_value = FakeObject(client=fake_client, volumes=fake_volumes) loc, size, checksum, metadata = self.store.add(expected_image_id, image_file, expected_size, self.context, verifier) self.assertEqual(expected_location, loc) self.assertEqual(expected_size, size) self.assertEqual(expected_checksum, checksum) fake_volumes.create.assert_called_once_with( 1, name='image-%s' % expected_image_id, metadata={'image_owner': self.context.tenant, 'glance_image_id': expected_image_id, 'image_size': str(expected_size)}, volume_type='some_type') self.assertEqual(backend, metadata["store"]) def test_cinder_add(self): fake_volume = mock.MagicMock(id=str(uuid.uuid4()), status='available', size=1) volume_file = six.BytesIO() self._test_cinder_add(fake_volume, volume_file) def test_cinder_add_with_verifier(self): fake_volume = mock.MagicMock(id=str(uuid.uuid4()), status='available', size=1) volume_file = six.BytesIO() verifier = mock.MagicMock() self._test_cinder_add(fake_volume, volume_file, 1, verifier) verifier.update.assert_called_with(b"*" * units.Ki) def test_cinder_add_volume_full(self): e = IOError() volume_file = six.BytesIO() e.errno = errno.ENOSPC fake_volume = mock.MagicMock(id=str(uuid.uuid4()), status='available', size=1) with mock.patch.object(volume_file, 'write', side_effect=e): self.assertRaises(exceptions.StorageFull, self._test_cinder_add, fake_volume, volume_file) fake_volume.delete.assert_called_once_with() def test_cinder_delete(self): fake_client = FakeObject(auth_token=None, management_url=None) fake_volume_uuid = str(uuid.uuid4()) fake_volume = FakeObject(delete=mock.Mock()) fake_volumes = {fake_volume_uuid: fake_volume} with mock.patch.object(cinder, 'get_cinderclient') as mocked_cc: mocked_cc.return_value = FakeObject(client=fake_client, volumes=fake_volumes) uri = 'cinder://%s' % fake_volume_uuid loc = location.get_location_from_uri_and_backend(uri, "cinder1", conf=self.conf) self.store.delete(loc, context=self.context) fake_volume.delete.assert_called_once_with() def test_cinder_add_different_backend(self): self.store = cinder.Store(self.conf, backend="cinder2") self.store.configure() self.register_store_backend_schemes(self.store, 'cinder', 'cinder2') fake_volume = mock.MagicMock(id=str(uuid.uuid4()), status='available', size=1) volume_file = six.BytesIO() self._test_cinder_add(fake_volume, volume_file, backend="cinder2")