from neutron.openstack.common import log as logging
from neutron.openstack.common.rpc import dispatcher
from neutron.plugins.hyperv.agent import utils
+from neutron.plugins.hyperv.agent import utilsfactory
from neutron.plugins.hyperv.common import constants
LOG = logging.getLogger(__name__)
RPC_API_VERSION = '1.0'
def __init__(self):
- self._utils = utils.HyperVUtils()
+ self._utils = utilsfactory.get_hypervutils()
self._polling_interval = CONF.AGENT.polling_interval
self._load_physical_network_mappings()
self._network_vswitch_map = {}
class HyperVException(q_exc.NeutronException):
message = _('HyperVException: %(msg)s')
+WMI_JOB_STATE_STARTED = 4096
WMI_JOB_STATE_RUNNING = 4
WMI_JOB_STATE_COMPLETED = 7
class HyperVUtils(object):
+
+ _ETHERNET_SWITCH_PORT = 'Msvm_SwitchPort'
+
+ _wmi_namespace = '//./root/virtualization'
+
def __init__(self):
self._wmi_conn = None
@property
def _conn(self):
if self._wmi_conn is None:
- self._wmi_conn = wmi.WMI(moniker='//./root/virtualization')
+ self._wmi_conn = wmi.WMI(moniker=self._wmi_namespace)
return self._wmi_conn
def get_switch_ports(self, vswitch_name):
vswitch = self._get_vswitch(vswitch_name)
vswitch_ports = vswitch.associators(
- wmi_result_class='Msvm_SwitchPort')
+ wmi_result_class=self._ETHERNET_SWITCH_PORT)
return set(p.Name for p in vswitch_ports)
def vnic_port_exists(self, port_id):
def get_vnic_ids(self):
return set(
p.ElementName
- for p in self._conn.Msvm_SyntheticEthernetPortSettingData())
+ for p in self._conn.Msvm_SyntheticEthernetPortSettingData()
+ if p.ElementName is not None)
def _get_vnic_settings(self, vnic_name):
vnic_settings = self._conn.Msvm_SyntheticEthernetPortSettingData(
vm = self._get_vm_from_res_setting_data(res_setting_data)
vs_man_svc = self._conn.Msvm_VirtualSystemManagementService()[0]
- (job_path,
- ret_val) = vs_man_svc.ModifyVirtualSystemResources(
- vm.Path_(), [res_setting_data.GetText_(1)])
+ (job_path, ret_val) = vs_man_svc.ModifyVirtualSystemResources(
+ vm.Path_(), [res_setting_data.GetText_(1)])
self._check_job_status(ret_val, job_path)
def _check_job_status(self, ret_val, jobpath):
"""Poll WMI job state for completion."""
if not ret_val:
return
- elif ret_val != WMI_JOB_STATE_RUNNING:
+ elif ret_val not in [WMI_JOB_STATE_STARTED, WMI_JOB_STATE_RUNNING]:
raise HyperVException(msg=_('Job failed with error %d') % ret_val)
job_wmi_path = jobpath.replace('\\', '/')
def _get_vswitch_external_port(self, vswitch):
vswitch_ports = vswitch.associators(
- wmi_result_class='Msvm_SwitchPort')
+ wmi_result_class=self._ETHERNET_SWITCH_PORT)
for vswitch_port in vswitch_ports:
lan_endpoints = vswitch_port.associators(
wmi_result_class='Msvm_SwitchLanEndpoint')
def get_port_by_id(self, port_id, vswitch_name):
vswitch = self._get_vswitch(vswitch_name)
- switch_ports = vswitch.associators(wmi_result_class='Msvm_SwitchPort')
+ switch_ports = vswitch.associators(
+ wmi_result_class=self._ETHERNET_SWITCH_PORT)
for switch_port in switch_ports:
if (switch_port.ElementName == port_id):
return switch_port
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 Cloudbase Solutions SRL
+# 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.
+# @author: Claudiu Belu, Cloudbase Solutions Srl
+
+import sys
+
+from oslo.config import cfg
+
+from neutron.openstack.common import log as logging
+from neutron.plugins.hyperv.agent import utils
+from neutron.plugins.hyperv.agent import utilsv2
+
+# Check needed for unit testing on Unix
+if sys.platform == 'win32':
+ import wmi
+
+hyper_opts = [
+ cfg.BoolOpt('force_hyperv_utils_v1',
+ default=False,
+ help='Force V1 WMI utility classes'),
+]
+
+CONF = cfg.CONF
+CONF.register_opts(hyper_opts, 'hyperv')
+
+LOG = logging.getLogger(__name__)
+
+
+def _get_windows_version():
+ return wmi.WMI(moniker='//./root/cimv2').Win32_OperatingSystem()[0].Version
+
+
+def _check_min_windows_version(major, minor, build=0):
+ version_str = _get_windows_version()
+ return map(int, version_str.split('.')) >= [major, minor, build]
+
+
+def _get_class(v1_class, v2_class, force_v1_flag):
+ # V2 classes are supported starting from Hyper-V Server 2012 and
+ # Windows Server 2012 (kernel version 6.2)
+ if not force_v1_flag and _check_min_windows_version(6, 2):
+ cls = v2_class
+ else:
+ cls = v1_class
+ LOG.debug("Loading class: %(module_name)s.%(class_name)s",
+ {'module_name': cls.__module__, 'class_name': cls.__name__})
+ return cls
+
+
+def get_hypervutils():
+ return _get_class(utils.HyperVUtils, utilsv2.HyperVUtilsV2,
+ CONF.hyperv.force_hyperv_utils_v1)()
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 Cloudbase Solutions SRL
+# 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.
+# @author: Alessandro Pilotti, Cloudbase Solutions Srl
+# @author: Claudiu Belu, Cloudbase Solutions Srl
+
+from neutron.plugins.hyperv.agent import utils
+
+
+class HyperVUtilsV2(utils.HyperVUtils):
+
+ _EXTERNAL_PORT = 'Msvm_ExternalEthernetPort'
+ _ETHERNET_SWITCH_PORT = 'Msvm_EthernetSwitchPort'
+ _PORT_ALLOC_SET_DATA = 'Msvm_EthernetPortAllocationSettingData'
+ _PORT_VLAN_SET_DATA = 'Msvm_EthernetSwitchPortVlanSettingData'
+ _LAN_ENDPOINT = 'Msvm_LANEndpoint'
+ _STATE_DISABLED = 3
+ _OPERATION_MODE_ACCESS = 1
+
+ _wmi_namespace = '//./root/virtualization/v2'
+
+ def __init__(self):
+ super(HyperVUtilsV2, self).__init__()
+
+ def connect_vnic_to_vswitch(self, vswitch_name, switch_port_name):
+ vnic = self._get_vnic_settings(switch_port_name)
+ vswitch = self._get_vswitch(vswitch_name)
+
+ port, found = self._get_switch_port_allocation(switch_port_name, True)
+ port.HostResource = [vswitch.path_()]
+ port.Parent = vnic.path_()
+ if not found:
+ vm = self._get_vm_from_res_setting_data(vnic)
+ self._add_virt_resource(vm, port)
+ else:
+ self._modify_virt_resource(port)
+
+ def _modify_virt_resource(self, res_setting_data):
+ vs_man_svc = self._conn.Msvm_VirtualSystemManagementService()[0]
+ (job_path, out_set_data, ret_val) = vs_man_svc.ModifyResourceSettings(
+ ResourceSettings=[res_setting_data.GetText_(1)])
+ self._check_job_status(ret_val, job_path)
+
+ def _add_virt_resource(self, vm, res_setting_data):
+ vs_man_svc = self._conn.Msvm_VirtualSystemManagementService()[0]
+ (job_path, out_set_data, ret_val) = vs_man_svc.AddResourceSettings(
+ vm.path_(), [res_setting_data.GetText_(1)])
+ self._check_job_status(ret_val, job_path)
+
+ def _remove_virt_resource(self, res_setting_data):
+ vs_man_svc = self._conn.Msvm_VirtualSystemManagementService()[0]
+ (job, ret_val) = vs_man_svc.RemoveResourceSettings(
+ ResourceSettings=[res_setting_data.path_()])
+ self._check_job_status(ret_val, job)
+
+ def disconnect_switch_port(
+ self, vswitch_name, switch_port_name, delete_port):
+ """Disconnects the switch port."""
+ sw_port, found = self._get_switch_port_allocation(switch_port_name)
+ if not sw_port:
+ # Port not found. It happens when the VM was already deleted.
+ return
+
+ if delete_port:
+ self._remove_virt_resource(sw_port)
+ else:
+ sw_port.EnabledState = self._STATE_DISABLED
+ self._modify_virt_resource(sw_port)
+
+ def _get_vswitch(self, vswitch_name):
+ vswitch = self._conn.Msvm_VirtualEthernetSwitch(
+ ElementName=vswitch_name)
+ if not len(vswitch):
+ raise utils.HyperVException(msg=_('VSwitch not found: %s') %
+ vswitch_name)
+ return vswitch[0]
+
+ def _get_vswitch_external_port(self, vswitch):
+ vswitch_ports = vswitch.associators(
+ wmi_result_class=self._ETHERNET_SWITCH_PORT)
+ for vswitch_port in vswitch_ports:
+ lan_endpoints = vswitch_port.associators(
+ wmi_result_class=self._LAN_ENDPOINT)
+ if len(lan_endpoints):
+ lan_endpoints = lan_endpoints[0].associators(
+ wmi_result_class=self._LAN_ENDPOINT)
+ if len(lan_endpoints):
+ ext_port = lan_endpoints[0].associators(
+ wmi_result_class=self._EXTERNAL_PORT)
+ if ext_port:
+ return vswitch_port
+
+ def set_vswitch_port_vlan_id(self, vlan_id, switch_port_name):
+ port_alloc, found = self._get_switch_port_allocation(switch_port_name)
+ if not found:
+ raise utils.HyperVException(
+ msg=_('Port Alloc not found: %s') % switch_port_name)
+
+ vs_man_svc = self._conn.Msvm_VirtualSystemManagementService()[0]
+ vlan_settings = self._get_vlan_setting_data_from_port_alloc(port_alloc)
+ if vlan_settings:
+ # Removing the feature because it cannot be modified
+ # due to a wmi exception.
+ (job_path, ret_val) = vs_man_svc.RemoveFeatureSettings(
+ FeatureSettings=[vlan_settings.path_()])
+ self._check_job_status(ret_val, job_path)
+
+ (vlan_settings, found) = self._get_vlan_setting_data(switch_port_name)
+ vlan_settings.AccessVlanId = vlan_id
+ vlan_settings.OperationMode = self._OPERATION_MODE_ACCESS
+ (job_path, out, ret_val) = vs_man_svc.AddFeatureSettings(
+ port_alloc.path_(), [vlan_settings.GetText_(1)])
+ self._check_job_status(ret_val, job_path)
+
+ def _get_vlan_setting_data_from_port_alloc(self, port_alloc):
+ return self._get_first_item(port_alloc.associators(
+ wmi_result_class=self._PORT_VLAN_SET_DATA))
+
+ def _get_vlan_setting_data(self, switch_port_name, create=True):
+ return self._get_setting_data(
+ self._PORT_VLAN_SET_DATA,
+ switch_port_name, create)
+
+ def _get_switch_port_allocation(self, switch_port_name, create=False):
+ return self._get_setting_data(
+ self._PORT_ALLOC_SET_DATA,
+ switch_port_name, create)
+
+ def _get_setting_data(self, class_name, element_name, create=True):
+ element_name = element_name.replace("'", '"')
+ q = self._conn.query("SELECT * FROM %(class_name)s WHERE "
+ "ElementName = '%(element_name)s'" %
+ {"class_name": class_name,
+ "element_name": element_name})
+ data = self._get_first_item(q)
+ found = data is not None
+ if not data and create:
+ data = self._get_default_setting_data(class_name)
+ data.ElementName = element_name
+ return data, found
+
+ def _get_default_setting_data(self, class_name):
+ return self._conn.query("SELECT * FROM %s WHERE InstanceID "
+ "LIKE '%%\\Default'" % class_name)[0]
+
+ def _get_first_item(self, obj):
+ if obj:
+ return obj[0]
from oslo.config import cfg
from neutron.plugins.hyperv.agent import hyperv_neutron_agent
+from neutron.plugins.hyperv.agent import utilsfactory
from neutron.tests import base
# Avoid rpc initialization for unit tests
cfg.CONF.set_override('rpc_backend',
'neutron.openstack.common.rpc.impl_fake')
+
+ utilsfactory._get_windows_version = mock.MagicMock(
+ return_value='6.2.0')
self.agent = hyperv_neutron_agent.HyperVNeutronAgent()
self.agent.plugin_rpc = mock.Mock()
self.agent.context = mock.Mock()
self.agent.agent_id = mock.Mock()
- self.agent._utils = mock.Mock()
def test_port_bound(self):
port = mock.Mock()
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 Cloudbase Solutions SRL
+# 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.
+# @author: Claudiu Belu, Cloudbase Solutions Srl
+
+"""
+Unit tests for the Hyper-V utils factory.
+"""
+
+import mock
+
+from oslo.config import cfg
+
+from neutron.plugins.hyperv.agent import utils
+from neutron.plugins.hyperv.agent import utilsfactory
+from neutron.plugins.hyperv.agent import utilsv2
+from neutron.tests import base
+
+CONF = cfg.CONF
+
+
+class TestHyperVUtilsFactory(base.BaseTestCase):
+
+ def test_get_hypervutils_v2(self):
+ self._test_returned_class(utilsv2.HyperVUtilsV2, False, '6.2.0')
+
+ def test_get_hypervutils_v1_old_version(self):
+ self._test_returned_class(utils.HyperVUtils, False, '6.1.0')
+
+ def test_get_hypervutils_v1_forced(self):
+ self._test_returned_class(utils.HyperVUtils, True, '6.2.0')
+
+ def _test_returned_class(self, expected_class, force_v1, os_version):
+ CONF.hyperv.force_hyperv_utils_v1 = force_v1
+ utilsfactory._get_windows_version = mock.MagicMock(
+ return_value=os_version)
+ actual_class = type(utilsfactory.get_hypervutils())
+ self.assertEqual(actual_class, expected_class)
--- /dev/null
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 Cloudbase Solutions SRL
+# 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.
+# @author: Alessandro Pilotti, Cloudbase Solutions Srl
+
+"""
+Unit tests for the Hyper-V utils V2.
+"""
+
+import mock
+
+from neutron.plugins.hyperv.agent import utils
+from neutron.plugins.hyperv.agent import utilsv2
+from neutron.tests import base
+
+
+class TestHyperVUtilsV2(base.BaseTestCase):
+
+ _FAKE_VSWITCH_NAME = "fake_vswitch_name"
+ _FAKE_PORT_NAME = "fake_port_name"
+ _FAKE_JOB_PATH = 'fake_job_path'
+ _FAKE_RET_VAL = 0
+ _FAKE_VM_PATH = "fake_vm_path"
+ _FAKE_RES_DATA = "fake_res_data"
+ _FAKE_RES_PATH = "fake_res_path"
+ _FAKE_VSWITCH = "fake_vswitch"
+ _FAKE_VLAN_ID = "fake_vlan_id"
+ _FAKE_CLASS_NAME = "fake_class_name"
+ _FAKE_ELEMENT_NAME = "fake_element_name"
+
+ def setUp(self):
+ super(TestHyperVUtilsV2, self).setUp()
+ self._utils = utilsv2.HyperVUtilsV2()
+ self._utils._wmi_conn = mock.MagicMock()
+
+ def test_connect_vnic_to_vswitch_found(self):
+ self._test_connect_vnic_to_vswitch(True)
+
+ def test_connect_vnic_to_vswitch_not_found(self):
+ self._test_connect_vnic_to_vswitch(False)
+
+ def _test_connect_vnic_to_vswitch(self, found):
+ self._utils._get_vnic_settings = mock.MagicMock()
+
+ if not found:
+ mock_vm = mock.MagicMock()
+ self._utils._get_vm_from_res_setting_data = mock.MagicMock(
+ return_value=mock_vm)
+ self._utils._add_virt_resource = mock.MagicMock()
+ else:
+ self._utils._modify_virt_resource = mock.MagicMock()
+
+ self._utils._get_vswitch = mock.MagicMock()
+ self._utils._get_switch_port_allocation = mock.MagicMock()
+
+ mock_port = mock.MagicMock()
+ self._utils._get_switch_port_allocation.return_value = (mock_port,
+ found)
+
+ self._utils.connect_vnic_to_vswitch(self._FAKE_VSWITCH_NAME,
+ self._FAKE_PORT_NAME)
+
+ if not found:
+ self._utils._add_virt_resource.assert_called_with(mock_vm,
+ mock_port)
+ else:
+ self._utils._modify_virt_resource.assert_called_with(mock_port)
+
+ def test_add_virt_resource(self):
+ mock_svc = self._utils._conn.Msvm_VirtualSystemManagementService()[0]
+ mock_svc.AddResourceSettings.return_value = (self._FAKE_JOB_PATH,
+ mock.MagicMock(),
+ self._FAKE_RET_VAL)
+ mock_res_setting_data = mock.MagicMock()
+ mock_res_setting_data.GetText_.return_value = self._FAKE_RES_DATA
+
+ mock_vm = mock.MagicMock()
+ mock_vm.path_.return_value = self._FAKE_VM_PATH
+
+ self._utils._check_job_status = mock.MagicMock()
+
+ self._utils._add_virt_resource(mock_vm, mock_res_setting_data)
+
+ mock_svc.AddResourceSettings.assert_called_with(self._FAKE_VM_PATH,
+ [self._FAKE_RES_DATA])
+
+ def test_modify_virt_resource(self):
+ mock_svc = self._utils._conn.Msvm_VirtualSystemManagementService()[0]
+ mock_svc.ModifyResourceSettings.return_value = (self._FAKE_JOB_PATH,
+ mock.MagicMock(),
+ self._FAKE_RET_VAL)
+ mock_res_setting_data = mock.MagicMock()
+ mock_res_setting_data.GetText_.return_value = self._FAKE_RES_DATA
+
+ self._utils._check_job_status = mock.MagicMock()
+
+ self._utils._modify_virt_resource(mock_res_setting_data)
+
+ mock_svc.ModifyResourceSettings.assert_called_with(
+ ResourceSettings=[self._FAKE_RES_DATA])
+
+ def test_remove_virt_resource(self):
+ mock_svc = self._utils._conn.Msvm_VirtualSystemManagementService()[0]
+ mock_svc.RemoveResourceSettings.return_value = (self._FAKE_JOB_PATH,
+ self._FAKE_RET_VAL)
+ mock_res_setting_data = mock.MagicMock()
+ mock_res_setting_data.path_.return_value = self._FAKE_RES_PATH
+
+ self._utils._check_job_status = mock.MagicMock()
+
+ self._utils._remove_virt_resource(mock_res_setting_data)
+
+ mock_svc.RemoveResourceSettings.assert_called_with(
+ ResourceSettings=[self._FAKE_RES_PATH])
+
+ def test_disconnect_switch_port_delete_port(self):
+ self._test_disconnect_switch_port(True)
+
+ def test_disconnect_switch_port_modify_port(self):
+ self._test_disconnect_switch_port(False)
+
+ def _test_disconnect_switch_port(self, delete_port):
+ self._utils._get_switch_port_allocation = mock.MagicMock()
+
+ mock_sw_port = mock.MagicMock()
+ self._utils._get_switch_port_allocation.return_value = (mock_sw_port,
+ True)
+
+ if delete_port:
+ self._utils._remove_virt_resource = mock.MagicMock()
+ else:
+ self._utils._modify_virt_resource = mock.MagicMock()
+
+ self._utils.disconnect_switch_port(self._FAKE_VSWITCH_NAME,
+ self._FAKE_PORT_NAME,
+ delete_port)
+
+ if delete_port:
+ self._utils._remove_virt_resource.assert_called_with(mock_sw_port)
+ else:
+ self._utils._modify_virt_resource.assert_called_with(mock_sw_port)
+
+ def test_get_vswitch(self):
+ self._utils._conn.Msvm_VirtualEthernetSwitch.return_value = [
+ self._FAKE_VSWITCH]
+ vswitch = self._utils._get_vswitch(self._FAKE_VSWITCH_NAME)
+
+ self.assertEqual(self._FAKE_VSWITCH, vswitch)
+
+ def test_get_vswitch_not_found(self):
+ self._utils._conn.Msvm_VirtualEthernetSwitch.return_value = []
+ self.assertRaises(utils.HyperVException, self._utils._get_vswitch,
+ self._FAKE_VSWITCH_NAME)
+
+ def test_get_vswitch_external_port(self):
+ mock_vswitch = mock.MagicMock()
+ mock_sw_port = mock.MagicMock()
+ mock_vswitch.associators.return_value = [mock_sw_port]
+ mock_le = mock_sw_port.associators.return_value
+ mock_le.__len__.return_value = 1
+ mock_le1 = mock_le[0].associators.return_value
+ mock_le1.__len__.return_value = 1
+
+ vswitch_port = self._utils._get_vswitch_external_port(mock_vswitch)
+
+ self.assertEqual(mock_sw_port, vswitch_port)
+
+ def test_set_vswitch_port_vlan_id(self):
+ mock_port_alloc = mock.MagicMock()
+ self._utils._get_switch_port_allocation = mock.MagicMock(return_value=(
+ mock_port_alloc, True))
+ self._utils._get_vlan_setting_data_from_port_alloc = mock.MagicMock()
+
+ mock_svc = self._utils._conn.Msvm_VirtualSystemManagementService()[0]
+ mock_svc.RemoveFeatureSettings.return_value = (self._FAKE_JOB_PATH,
+ self._FAKE_RET_VAL)
+ mock_vlan_settings = mock.MagicMock()
+ self._utils._get_vlan_setting_data = mock.MagicMock(return_value=(
+ mock_vlan_settings, True))
+
+ mock_svc.AddFeatureSettings.return_value = (self._FAKE_JOB_PATH,
+ None,
+ self._FAKE_RET_VAL)
+
+ self._utils.set_vswitch_port_vlan_id(self._FAKE_VLAN_ID,
+ self._FAKE_PORT_NAME)
+
+ self.assertTrue(mock_svc.RemoveFeatureSettings.called)
+ self.assertTrue(mock_svc.AddFeatureSettings.called)
+
+ def test_get_setting_data(self):
+ self._utils._get_first_item = mock.MagicMock(return_value=None)
+
+ mock_data = mock.MagicMock()
+ self._utils._get_default_setting_data = mock.MagicMock(
+ return_value=mock_data)
+
+ ret_val = self._utils._get_setting_data(self._FAKE_CLASS_NAME,
+ self._FAKE_ELEMENT_NAME,
+ True)
+
+ self.assertEqual(ret_val, (mock_data, False))
Code Review / openstack-build / neutron-build.git / commitdiff