diff options
Diffstat (limited to 'nova/tests/unit/virt/powervm/test_driver.py')
| -rw-r--r-- | nova/tests/unit/virt/powervm/test_driver.py | 649 |
1 files changed, 0 insertions, 649 deletions
diff --git a/nova/tests/unit/virt/powervm/test_driver.py b/nova/tests/unit/virt/powervm/test_driver.py deleted file mode 100644 index 025d823d15..0000000000 --- a/nova/tests/unit/virt/powervm/test_driver.py +++ /dev/null @@ -1,649 +0,0 @@ -# Copyright 2016, 2018 IBM Corp. -# -# 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 fixtures -import mock -from oslo_serialization import jsonutils -from oslo_utils.fixture import uuidsentinel as uuids -from pypowervm import const as pvm_const -from pypowervm import exceptions as pvm_exc -from pypowervm.helpers import log_helper as pvm_hlp_log -from pypowervm.helpers import vios_busy as pvm_hlp_vbusy -from pypowervm.utils import transaction as pvm_tx -from pypowervm.wrappers import virtual_io_server as pvm_vios - -from nova import block_device as nova_block_device -from nova.compute import provider_tree -from nova import conf as cfg -from nova import exception -from nova.objects import block_device as bdmobj -from nova import test -from nova.tests.unit.virt import powervm -from nova.virt import block_device as nova_virt_bdm -from nova.virt import driver as nova_driver -from nova.virt.driver import ComputeDriver -from nova.virt import hardware -from nova.virt.powervm.disk import ssp -from nova.virt.powervm import driver - -CONF = cfg.CONF - - -class TestPowerVMDriver(test.NoDBTestCase): - - def setUp(self): - super(TestPowerVMDriver, self).setUp() - self.drv = driver.PowerVMDriver('virtapi') - self.adp = self.useFixture(fixtures.MockPatch( - 'pypowervm.adapter.Adapter', autospec=True)).mock - self.drv.adapter = self.adp - self.sess = self.useFixture(fixtures.MockPatch( - 'pypowervm.adapter.Session', autospec=True)).mock - - self.pwron = self.useFixture(fixtures.MockPatch( - 'nova.virt.powervm.vm.power_on')).mock - self.pwroff = self.useFixture(fixtures.MockPatch( - 'nova.virt.powervm.vm.power_off')).mock - - # Create an instance to test with - self.inst = powervm.TEST_INSTANCE - - def test_driver_capabilities(self): - """Test the driver capabilities.""" - # check that the driver reports all capabilities - self.assertEqual(set(ComputeDriver.capabilities), - set(self.drv.capabilities)) - # check the values for each capability - self.assertFalse(self.drv.capabilities['has_imagecache']) - self.assertFalse(self.drv.capabilities['supports_evacuate']) - self.assertFalse( - self.drv.capabilities['supports_migrate_to_same_host']) - self.assertTrue(self.drv.capabilities['supports_attach_interface']) - self.assertFalse(self.drv.capabilities['supports_device_tagging']) - self.assertFalse( - self.drv.capabilities['supports_tagged_attach_interface']) - self.assertFalse( - self.drv.capabilities['supports_tagged_attach_volume']) - self.assertTrue(self.drv.capabilities['supports_extend_volume']) - self.assertFalse(self.drv.capabilities['supports_multiattach']) - - @mock.patch('nova.image.glance.API') - @mock.patch('pypowervm.tasks.storage.ComprehensiveScrub', autospec=True) - @mock.patch('oslo_utils.importutils.import_object_ns', autospec=True) - @mock.patch('pypowervm.wrappers.managed_system.System', autospec=True) - @mock.patch('pypowervm.tasks.partition.validate_vios_ready', autospec=True) - def test_init_host(self, mock_vvr, mock_sys, mock_import, mock_scrub, - mock_img): - mock_hostw = mock.Mock(uuid='uuid') - mock_sys.get.return_value = [mock_hostw] - self.drv.init_host('host') - self.sess.assert_called_once_with(conn_tries=60) - self.adp.assert_called_once_with( - self.sess.return_value, helpers=[ - pvm_hlp_log.log_helper, pvm_hlp_vbusy.vios_busy_retry_helper]) - mock_vvr.assert_called_once_with(self.drv.adapter) - mock_sys.get.assert_called_once_with(self.drv.adapter) - self.assertEqual(mock_hostw, self.drv.host_wrapper) - mock_scrub.assert_called_once_with(self.drv.adapter) - mock_scrub.return_value.execute.assert_called_once_with() - mock_import.assert_called_once_with( - 'nova.virt.powervm.disk', 'localdisk.LocalStorage', - self.drv.adapter, 'uuid') - self.assertEqual(mock_import.return_value, self.drv.disk_dvr) - mock_img.assert_called_once_with() - self.assertEqual(mock_img.return_value, self.drv.image_api) - - @mock.patch('nova.virt.powervm.vm.get_pvm_uuid') - @mock.patch('nova.virt.powervm.vm.get_vm_qp') - @mock.patch('nova.virt.powervm.vm._translate_vm_state') - def test_get_info(self, mock_tx_state, mock_qp, mock_uuid): - mock_tx_state.return_value = 'fake-state' - self.assertEqual(hardware.InstanceInfo('fake-state'), - self.drv.get_info('inst')) - mock_uuid.assert_called_once_with('inst') - mock_qp.assert_called_once_with( - self.drv.adapter, mock_uuid.return_value, 'PartitionState') - mock_tx_state.assert_called_once_with(mock_qp.return_value) - - @mock.patch('nova.virt.powervm.vm.get_lpar_names') - def test_list_instances(self, mock_names): - mock_names.return_value = ['one', 'two', 'three'] - self.assertEqual(['one', 'two', 'three'], self.drv.list_instances()) - mock_names.assert_called_once_with(self.adp) - - def test_get_available_nodes(self): - self.flags(host='hostname') - self.assertEqual(['hostname'], self.drv.get_available_nodes('node')) - - @mock.patch('pypowervm.wrappers.managed_system.System', autospec=True) - @mock.patch('nova.virt.powervm.host.build_host_resource_from_ms') - def test_get_available_resource(self, mock_bhrfm, mock_sys): - mock_sys.get.return_value = ['sys'] - mock_bhrfm.return_value = {'foo': 'bar'} - self.drv.disk_dvr = mock.create_autospec(ssp.SSPDiskAdapter, - instance=True) - self.assertEqual( - {'foo': 'bar', 'local_gb': self.drv.disk_dvr.capacity, - 'local_gb_used': self.drv.disk_dvr.capacity_used}, - self.drv.get_available_resource('node')) - mock_sys.get.assert_called_once_with(self.adp) - mock_bhrfm.assert_called_once_with('sys') - self.assertEqual('sys', self.drv.host_wrapper) - - @contextlib.contextmanager - def _update_provider_tree(self, allocations=None): - """Host resource dict gets converted properly to provider tree inv.""" - - with mock.patch('nova.virt.powervm.host.' - 'build_host_resource_from_ms') as mock_bhrfm: - mock_bhrfm.return_value = { - 'vcpus': 8, - 'memory_mb': 2048, - } - self.drv.host_wrapper = 'host_wrapper' - # Validate that this gets converted to int with floor - self.drv.disk_dvr = mock.Mock(capacity=2091.8) - exp_inv = { - 'VCPU': { - 'total': 8, - 'max_unit': 8, - 'allocation_ratio': 16.0, - 'reserved': 0, - }, - 'MEMORY_MB': { - 'total': 2048, - 'max_unit': 2048, - 'allocation_ratio': 1.5, - 'reserved': 512, - }, - 'DISK_GB': { - 'total': 2091, - 'max_unit': 2091, - 'allocation_ratio': 1.0, - 'reserved': 0, - }, - } - ptree = provider_tree.ProviderTree() - ptree.new_root('compute_host', uuids.cn) - # Let the caller muck with these - yield ptree, exp_inv - self.drv.update_provider_tree(ptree, 'compute_host', - allocations=allocations) - self.assertEqual(exp_inv, ptree.data('compute_host').inventory) - mock_bhrfm.assert_called_once_with('host_wrapper') - - def test_update_provider_tree(self): - # Basic: no inventory already on the provider, no extra providers, no - # aggregates or traits. - with self._update_provider_tree(): - pass - - def test_update_provider_tree_ignore_allocations(self): - with self._update_provider_tree(allocations="This is ignored"): - pass - - def test_update_provider_tree_conf_overrides(self): - # Non-default CONF values for allocation ratios and reserved. - self.flags(cpu_allocation_ratio=12.3, - reserved_host_cpus=4, - ram_allocation_ratio=4.5, - reserved_host_memory_mb=32, - disk_allocation_ratio=6.7, - # This gets int(ceil)'d - reserved_host_disk_mb=5432.1) - with self._update_provider_tree() as (_, exp_inv): - exp_inv['VCPU']['allocation_ratio'] = 12.3 - exp_inv['VCPU']['reserved'] = 4 - exp_inv['MEMORY_MB']['allocation_ratio'] = 4.5 - exp_inv['MEMORY_MB']['reserved'] = 32 - exp_inv['DISK_GB']['allocation_ratio'] = 6.7 - exp_inv['DISK_GB']['reserved'] = 6 - - def test_update_provider_tree_complex_ptree(self): - # Overrides inventory already on the provider; leaves other providers - # and aggregates/traits alone. - with self._update_provider_tree() as (ptree, exp_inv): - ptree.update_inventory('compute_host', { - # these should get blown away - 'VCPU': { - 'total': 16, - 'max_unit': 2, - 'allocation_ratio': 1.0, - 'reserved': 10, - }, - 'CUSTOM_BOGUS': { - 'total': 1234, - } - }) - ptree.update_aggregates('compute_host', - [uuids.ss_agg, uuids.other_agg]) - ptree.update_traits('compute_host', ['CUSTOM_FOO', 'CUSTOM_BAR']) - ptree.new_root('ssp', uuids.ssp) - ptree.update_inventory('ssp', {'sentinel': 'inventory', - 'for': 'ssp'}) - ptree.update_aggregates('ssp', [uuids.ss_agg]) - ptree.new_child('sriov', 'compute_host', uuid=uuids.sriov) - # Since CONF.cpu_allocation_ratio is not set and this is not - # the initial upt call (so CONF.initial_cpu_allocation_ratio would - # be used), the existing allocation ratio value from the tree is - # used. - exp_inv['VCPU']['allocation_ratio'] = 1.0 - - # Make sure the compute's agg and traits were left alone - cndata = ptree.data('compute_host') - self.assertEqual(set([uuids.ss_agg, uuids.other_agg]), - cndata.aggregates) - self.assertEqual(set(['CUSTOM_FOO', 'CUSTOM_BAR']), cndata.traits) - # And the other providers were left alone - self.assertEqual(set([uuids.cn, uuids.ssp, uuids.sriov]), - set(ptree.get_provider_uuids())) - # ...including the ssp's aggregates - self.assertEqual(set([uuids.ss_agg]), ptree.data('ssp').aggregates) - - @mock.patch('nova.virt.powervm.tasks.storage.AttachVolume.execute') - @mock.patch('nova.virt.powervm.tasks.network.PlugMgmtVif.execute') - @mock.patch('nova.virt.powervm.tasks.network.PlugVifs.execute') - @mock.patch('nova.virt.powervm.media.ConfigDrivePowerVM') - @mock.patch('nova.virt.configdrive.required_by') - @mock.patch('nova.virt.powervm.vm.create_lpar') - @mock.patch('pypowervm.tasks.partition.build_active_vio_feed_task', - autospec=True) - @mock.patch('pypowervm.tasks.storage.add_lpar_storage_scrub_tasks', - autospec=True) - def test_spawn_ops(self, mock_scrub, mock_bldftsk, mock_crt_lpar, - mock_cdrb, mock_cfg_drv, mock_plug_vifs, - mock_plug_mgmt_vif, mock_attach_vol): - """Validates the 'typical' spawn flow of the spawn of an instance. """ - mock_cdrb.return_value = True - self.drv.host_wrapper = mock.Mock() - self.drv.disk_dvr = mock.create_autospec(ssp.SSPDiskAdapter, - instance=True) - mock_ftsk = pvm_tx.FeedTask('fake', [mock.Mock(spec=pvm_vios.VIOS)]) - mock_bldftsk.return_value = mock_ftsk - block_device_info = self._fake_bdms() - self.drv.spawn('context', self.inst, 'img_meta', 'files', 'password', - 'allocs', network_info='netinfo', - block_device_info=block_device_info) - mock_crt_lpar.assert_called_once_with( - self.adp, self.drv.host_wrapper, self.inst) - mock_bldftsk.assert_called_once_with( - self.adp, xag={pvm_const.XAG.VIO_SMAP, pvm_const.XAG.VIO_FMAP}) - self.assertTrue(mock_plug_vifs.called) - self.assertTrue(mock_plug_mgmt_vif.called) - mock_scrub.assert_called_once_with( - [mock_crt_lpar.return_value.id], mock_ftsk, lpars_exist=True) - self.drv.disk_dvr.create_disk_from_image.assert_called_once_with( - 'context', self.inst, 'img_meta') - self.drv.disk_dvr.attach_disk.assert_called_once_with( - self.inst, self.drv.disk_dvr.create_disk_from_image.return_value, - mock_ftsk) - self.assertEqual(2, mock_attach_vol.call_count) - mock_cfg_drv.assert_called_once_with(self.adp) - mock_cfg_drv.return_value.create_cfg_drv_vopt.assert_called_once_with( - self.inst, 'files', 'netinfo', mock_ftsk, admin_pass='password', - mgmt_cna=mock.ANY) - self.pwron.assert_called_once_with(self.adp, self.inst) - - mock_cfg_drv.reset_mock() - mock_attach_vol.reset_mock() - - # No config drive, no bdms - mock_cdrb.return_value = False - self.drv.spawn('context', self.inst, 'img_meta', 'files', 'password', - 'allocs') - mock_cfg_drv.assert_not_called() - mock_attach_vol.assert_not_called() - - @mock.patch('nova.virt.powervm.tasks.storage.DetachVolume.execute') - @mock.patch('nova.virt.powervm.tasks.network.UnplugVifs.execute') - @mock.patch('nova.virt.powervm.vm.delete_lpar') - @mock.patch('nova.virt.powervm.media.ConfigDrivePowerVM') - @mock.patch('nova.virt.configdrive.required_by') - @mock.patch('pypowervm.tasks.partition.build_active_vio_feed_task', - autospec=True) - def test_destroy(self, mock_bldftsk, mock_cdrb, mock_cfgdrv, - mock_dlt_lpar, mock_unplug, mock_detach_vol): - """Validates PowerVM destroy.""" - self.drv.host_wrapper = mock.Mock() - self.drv.disk_dvr = mock.create_autospec(ssp.SSPDiskAdapter, - instance=True) - - mock_ftsk = pvm_tx.FeedTask('fake', [mock.Mock(spec=pvm_vios.VIOS)]) - mock_bldftsk.return_value = mock_ftsk - block_device_info = self._fake_bdms() - - # Good path, with config drive, destroy disks - mock_cdrb.return_value = True - self.drv.destroy('context', self.inst, [], - block_device_info=block_device_info) - self.pwroff.assert_called_once_with( - self.adp, self.inst, force_immediate=True) - mock_bldftsk.assert_called_once_with( - self.adp, xag=[pvm_const.XAG.VIO_SMAP]) - mock_unplug.assert_called_once() - mock_cdrb.assert_called_once_with(self.inst) - mock_cfgdrv.assert_called_once_with(self.adp) - mock_cfgdrv.return_value.dlt_vopt.assert_called_once_with( - self.inst, stg_ftsk=mock_bldftsk.return_value) - self.assertEqual(2, mock_detach_vol.call_count) - self.drv.disk_dvr.detach_disk.assert_called_once_with( - self.inst) - self.drv.disk_dvr.delete_disks.assert_called_once_with( - self.drv.disk_dvr.detach_disk.return_value) - mock_dlt_lpar.assert_called_once_with(self.adp, self.inst) - - self.pwroff.reset_mock() - mock_bldftsk.reset_mock() - mock_unplug.reset_mock() - mock_cdrb.reset_mock() - mock_cfgdrv.reset_mock() - self.drv.disk_dvr.detach_disk.reset_mock() - self.drv.disk_dvr.delete_disks.reset_mock() - mock_detach_vol.reset_mock() - mock_dlt_lpar.reset_mock() - - # No config drive, preserve disks, no block device info - mock_cdrb.return_value = False - self.drv.destroy('context', self.inst, [], block_device_info={}, - destroy_disks=False) - mock_cfgdrv.return_value.dlt_vopt.assert_not_called() - mock_detach_vol.assert_not_called() - self.drv.disk_dvr.delete_disks.assert_not_called() - - # Non-forced power_off, since preserving disks - self.pwroff.assert_called_once_with( - self.adp, self.inst, force_immediate=False) - mock_bldftsk.assert_called_once_with( - self.adp, xag=[pvm_const.XAG.VIO_SMAP]) - mock_unplug.assert_called_once() - mock_cdrb.assert_called_once_with(self.inst) - mock_cfgdrv.assert_not_called() - mock_cfgdrv.return_value.dlt_vopt.assert_not_called() - self.drv.disk_dvr.detach_disk.assert_called_once_with( - self.inst) - self.drv.disk_dvr.delete_disks.assert_not_called() - mock_dlt_lpar.assert_called_once_with(self.adp, self.inst) - - self.pwroff.reset_mock() - mock_bldftsk.reset_mock() - mock_unplug.reset_mock() - mock_cdrb.reset_mock() - mock_cfgdrv.reset_mock() - self.drv.disk_dvr.detach_disk.reset_mock() - self.drv.disk_dvr.delete_disks.reset_mock() - mock_dlt_lpar.reset_mock() - - # InstanceNotFound exception, non-forced - self.pwroff.side_effect = exception.InstanceNotFound( - instance_id='something') - self.drv.destroy('context', self.inst, [], block_device_info={}, - destroy_disks=False) - self.pwroff.assert_called_once_with( - self.adp, self.inst, force_immediate=False) - self.drv.disk_dvr.detach_disk.assert_not_called() - mock_unplug.assert_not_called() - self.drv.disk_dvr.delete_disks.assert_not_called() - mock_dlt_lpar.assert_not_called() - - self.pwroff.reset_mock() - self.pwroff.side_effect = None - mock_unplug.reset_mock() - - # Convertible (PowerVM) exception - mock_dlt_lpar.side_effect = pvm_exc.TimeoutError("Timed out") - self.assertRaises(exception.InstanceTerminationFailure, - self.drv.destroy, 'context', self.inst, [], - block_device_info={}) - - # Everything got called - self.pwroff.assert_called_once_with( - self.adp, self.inst, force_immediate=True) - mock_unplug.assert_called_once() - self.drv.disk_dvr.detach_disk.assert_called_once_with(self.inst) - self.drv.disk_dvr.delete_disks.assert_called_once_with( - self.drv.disk_dvr.detach_disk.return_value) - mock_dlt_lpar.assert_called_once_with(self.adp, self.inst) - - # Other random exception raises directly - mock_dlt_lpar.side_effect = ValueError() - self.assertRaises(ValueError, - self.drv.destroy, 'context', self.inst, [], - block_device_info={}) - - @mock.patch('nova.virt.powervm.tasks.image.UpdateTaskState.' - 'execute', autospec=True) - @mock.patch('nova.virt.powervm.tasks.storage.InstanceDiskToMgmt.' - 'execute', autospec=True) - @mock.patch('nova.virt.powervm.tasks.image.StreamToGlance.execute') - @mock.patch('nova.virt.powervm.tasks.storage.RemoveInstanceDiskFromMgmt.' - 'execute') - def test_snapshot(self, mock_rm, mock_stream, mock_conn, mock_update): - self.drv.disk_dvr = mock.Mock() - self.drv.image_api = mock.Mock() - mock_conn.return_value = 'stg_elem', 'vios_wrap', 'disk_path' - self.drv.snapshot('context', self.inst, 'image_id', - 'update_task_state') - self.assertEqual(2, mock_update.call_count) - self.assertEqual(1, mock_conn.call_count) - mock_stream.assert_called_once_with(disk_path='disk_path') - mock_rm.assert_called_once_with( - stg_elem='stg_elem', vios_wrap='vios_wrap', disk_path='disk_path') - - self.drv.disk_dvr.capabilities = {'snapshot': False} - self.assertRaises(exception.NotSupportedWithOption, self.drv.snapshot, - 'context', self.inst, 'image_id', 'update_task_state') - - def test_power_on(self): - self.drv.power_on('context', self.inst, 'network_info') - self.pwron.assert_called_once_with(self.adp, self.inst) - - def test_power_off(self): - self.drv.power_off(self.inst) - self.pwroff.assert_called_once_with( - self.adp, self.inst, force_immediate=True, timeout=None) - - def test_power_off_timeout(self): - # Long timeout (retry interval means nothing on powervm) - self.drv.power_off(self.inst, timeout=500, retry_interval=10) - self.pwroff.assert_called_once_with( - self.adp, self.inst, force_immediate=False, timeout=500) - - @mock.patch('nova.virt.powervm.vm.reboot') - def test_reboot_soft(self, mock_reboot): - inst = mock.Mock() - self.drv.reboot('context', inst, 'network_info', 'SOFT') - mock_reboot.assert_called_once_with(self.adp, inst, False) - - @mock.patch('nova.virt.powervm.vm.reboot') - def test_reboot_hard(self, mock_reboot): - inst = mock.Mock() - self.drv.reboot('context', inst, 'network_info', 'HARD') - mock_reboot.assert_called_once_with(self.adp, inst, True) - - @mock.patch('nova.virt.powervm.driver.PowerVMDriver.plug_vifs') - def test_attach_interface(self, mock_plug_vifs): - self.drv.attach_interface('context', 'inst', 'image_meta', 'vif') - mock_plug_vifs.assert_called_once_with('inst', ['vif']) - - @mock.patch('nova.virt.powervm.driver.PowerVMDriver.unplug_vifs') - def test_detach_interface(self, mock_unplug_vifs): - self.drv.detach_interface('context', 'inst', 'vif') - mock_unplug_vifs.assert_called_once_with('inst', ['vif']) - - @mock.patch('nova.virt.powervm.tasks.vm.Get', autospec=True) - @mock.patch('nova.virt.powervm.tasks.base.run', autospec=True) - @mock.patch('nova.virt.powervm.tasks.network.PlugVifs', autospec=True) - @mock.patch('taskflow.patterns.linear_flow.Flow', autospec=True) - def test_plug_vifs(self, mock_tf, mock_plug_vifs, mock_tf_run, mock_get): - # Successful plug - mock_inst = mock.Mock() - self.drv.plug_vifs(mock_inst, 'net_info') - mock_get.assert_called_once_with(self.adp, mock_inst) - mock_plug_vifs.assert_called_once_with( - self.drv.virtapi, self.adp, mock_inst, 'net_info') - add_calls = [mock.call(mock_get.return_value), - mock.call(mock_plug_vifs.return_value)] - mock_tf.return_value.add.assert_has_calls(add_calls) - mock_tf_run.assert_called_once_with( - mock_tf.return_value, instance=mock_inst) - - # InstanceNotFound and generic exception both raise - mock_tf_run.side_effect = exception.InstanceNotFound('id') - exc = self.assertRaises(exception.VirtualInterfacePlugException, - self.drv.plug_vifs, mock_inst, 'net_info') - self.assertIn('instance', str(exc)) - mock_tf_run.side_effect = Exception - exc = self.assertRaises(exception.VirtualInterfacePlugException, - self.drv.plug_vifs, mock_inst, 'net_info') - self.assertIn('unexpected', str(exc)) - - @mock.patch('nova.virt.powervm.tasks.base.run', autospec=True) - @mock.patch('nova.virt.powervm.tasks.network.UnplugVifs', autospec=True) - @mock.patch('taskflow.patterns.linear_flow.Flow', autospec=True) - def test_unplug_vifs(self, mock_tf, mock_unplug_vifs, mock_tf_run): - # Successful unplug - mock_inst = mock.Mock() - self.drv.unplug_vifs(mock_inst, 'net_info') - mock_unplug_vifs.assert_called_once_with(self.adp, mock_inst, - 'net_info') - mock_tf.return_value.add.assert_called_once_with( - mock_unplug_vifs.return_value) - mock_tf_run.assert_called_once_with(mock_tf.return_value, - instance=mock_inst) - - # InstanceNotFound should pass - mock_tf_run.side_effect = exception.InstanceNotFound(instance_id='1') - self.drv.unplug_vifs(mock_inst, 'net_info') - - # Raise InterfaceDetachFailed otherwise - mock_tf_run.side_effect = Exception - self.assertRaises(exception.InterfaceDetachFailed, - self.drv.unplug_vifs, mock_inst, 'net_info') - - @mock.patch('pypowervm.tasks.vterm.open_remotable_vnc_vterm', - autospec=True) - @mock.patch('nova.virt.powervm.vm.get_pvm_uuid', - new=mock.Mock(return_value='uuid')) - def test_get_vnc_console(self, mock_vterm): - # Success - mock_vterm.return_value = '10' - resp = self.drv.get_vnc_console(mock.ANY, self.inst) - self.assertEqual('127.0.0.1', resp.host) - self.assertEqual('10', resp.port) - self.assertEqual('uuid', resp.internal_access_path) - mock_vterm.assert_called_once_with( - mock.ANY, 'uuid', mock.ANY, vnc_path='uuid') - - # VNC failure - exception is raised directly - mock_vterm.side_effect = pvm_exc.VNCBasedTerminalFailedToOpen(err='xx') - self.assertRaises(pvm_exc.VNCBasedTerminalFailedToOpen, - self.drv.get_vnc_console, mock.ANY, self.inst) - - # 404 - mock_vterm.side_effect = pvm_exc.HttpError(mock.Mock(status=404)) - self.assertRaises(exception.InstanceNotFound, self.drv.get_vnc_console, - mock.ANY, self.inst) - - @mock.patch('nova.virt.powervm.volume.fcvscsi.FCVscsiVolumeAdapter') - def test_attach_volume(self, mock_vscsi_adpt): - """Validates the basic PowerVM attach volume.""" - # BDMs - mock_bdm = self._fake_bdms()['block_device_mapping'][0] - - with mock.patch.object(self.inst, 'save') as mock_save: - # Invoke the method. - self.drv.attach_volume('context', mock_bdm.get('connection_info'), - self.inst, mock.sentinel.stg_ftsk) - - # Verify the connect volume was invoked - mock_vscsi_adpt.return_value.attach_volume.assert_called_once_with() - mock_save.assert_called_once_with() - - @mock.patch('nova.virt.powervm.volume.fcvscsi.FCVscsiVolumeAdapter') - def test_detach_volume(self, mock_vscsi_adpt): - """Validates the basic PowerVM detach volume.""" - # BDMs - mock_bdm = self._fake_bdms()['block_device_mapping'][0] - - # Invoke the method, good path test. - self.drv.detach_volume('context', mock_bdm.get('connection_info'), - self.inst, mock.sentinel.stg_ftsk) - # Verify the disconnect volume was invoked - mock_vscsi_adpt.return_value.detach_volume.assert_called_once_with() - - @mock.patch('nova.virt.powervm.volume.fcvscsi.FCVscsiVolumeAdapter') - def test_extend_volume(self, mock_vscsi_adpt): - mock_bdm = self._fake_bdms()['block_device_mapping'][0] - self.drv.extend_volume( - 'context', mock_bdm.get('connection_info'), self.inst, 0) - mock_vscsi_adpt.return_value.extend_volume.assert_called_once_with() - - def test_vol_drv_iter(self): - block_device_info = self._fake_bdms() - bdms = nova_driver.block_device_info_get_mapping(block_device_info) - vol_adpt = mock.Mock() - - def _get_results(bdms): - # Patch so we get the same mock back each time. - with mock.patch('nova.virt.powervm.volume.fcvscsi.' - 'FCVscsiVolumeAdapter', return_value=vol_adpt): - return [ - (bdm, vol_drv) for bdm, vol_drv in self.drv._vol_drv_iter( - 'context', self.inst, bdms)] - - results = _get_results(bdms) - self.assertEqual( - 'fake_vol1', - results[0][0]['connection_info']['data']['volume_id']) - self.assertEqual(vol_adpt, results[0][1]) - self.assertEqual( - 'fake_vol2', - results[1][0]['connection_info']['data']['volume_id']) - self.assertEqual(vol_adpt, results[1][1]) - - # Test with empty bdms - self.assertEqual([], _get_results([])) - - @staticmethod - def _fake_bdms(): - def _fake_bdm(volume_id, target_lun): - connection_info = {'driver_volume_type': 'fibre_channel', - 'data': {'volume_id': volume_id, - 'target_lun': target_lun, - 'initiator_target_map': - {'21000024F5': ['50050768']}}} - mapping_dict = {'source_type': 'volume', 'volume_id': volume_id, - 'destination_type': 'volume', - 'connection_info': - jsonutils.dumps(connection_info), - } - bdm_dict = nova_block_device.BlockDeviceDict(mapping_dict) - bdm_obj = bdmobj.BlockDeviceMapping(**bdm_dict) - - return nova_virt_bdm.DriverVolumeBlockDevice(bdm_obj) - - bdm_list = [_fake_bdm('fake_vol1', 0), _fake_bdm('fake_vol2', 1)] - block_device_info = {'block_device_mapping': bdm_list} - - return block_device_info - - @mock.patch('nova.virt.powervm.volume.fcvscsi.wwpns', autospec=True) - def test_get_volume_connector(self, mock_wwpns): - vol_connector = self.drv.get_volume_connector(mock.Mock()) - self.assertEqual(mock_wwpns.return_value, vol_connector['wwpns']) - self.assertFalse(vol_connector['multipath']) - self.assertEqual(vol_connector['host'], CONF.host) - self.assertIsNone(vol_connector['initiator']) |