x/vmware-nsx
Code Issues Proposed changes

Adds support for the Hyper-V WMI V2 namespace

Browse Source 
Blueprint: hyper-v-wmi-v2

The Hyper-V APIs are mainly based on WMI. The original 2008 Hyper-V
release introduced the "root\virtualization" namespace which got
superseded in Hyper-V Server / Windows Server 2012 by the
"root\virtualization\v2" namespace (referred as V2 in the sources).

The original namespace has been dropped in the upcoming Hyper-V 2012
R2 (currently available in preview), which means that the Grizzly code
will not be compatible with it as is.

The Hyper-V driver is structured with a clear decoupling between OS
interaction classes (so called *utils modules and classes) and the
agent's logic.

This allows us to provide an implementation of the V2 API without
impacting the rest of the agent's code, based on a factory module
added to instantiate the proper version of the *utils classes: the
original "V1" ones for versions of the OS predating 2012 and the
newer "V2" ones starting from Hyper-V 2012 (Windows kernel version
6.2).

Change-Id: I380d8c65715242d8a5b94b5061ebe4f30e6c2090
This commit is contained in:
Claudiu Belu 2013-08-07 07:15:09 -07:00
parent 6008e688a1
commit 5d2a06df39
7 changed files with 515 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
neutron/plugins/hyperv/agent/hyperv_neutron_agent.py
View File

@ -33,6 +33,7 @@ from neutron import context
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__)
@ -63,7 +64,7 @@ class HyperVNeutronAgent(object):
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 = {}

23
neutron/plugins/hyperv/agent/utils.py
View File

@ -37,24 +37,30 @@ LOG = logging.getLogger(__name__)
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):
@ -67,7 +73,8 @@ class HyperVUtils(object):
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(
@ -99,8 +106,7 @@ class HyperVUtils(object):
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(
(job_path, ret_val) = vs_man_svc.ModifyVirtualSystemResources(
vm.Path_(), [res_setting_data.GetText_(1)])
self._check_job_status(ret_val, job_path)
@ -108,7 +114,7 @@ class HyperVUtils(object):
"""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('\\', '/')
@ -204,7 +210,7 @@ class HyperVUtils(object):
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')
@ -232,7 +238,8 @@ class HyperVUtils(object):
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

66
neutron/plugins/hyperv/agent/utilsfactory.py Normal file
View File

@ -0,0 +1,66 @@
# 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)()

161
neutron/plugins/hyperv/agent/utilsv2.py Normal file
View File

@ -0,0 +1,161 @@
# 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]

5
neutron/tests/unit/hyperv/test_hyperv_neutron_agent.py
View File

@ -24,6 +24,7 @@ import mock
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
@ -35,11 +36,13 @@ class TestHyperVNeutronAgent(base.BaseTestCase):
# 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()

51
neutron/tests/unit/hyperv/test_hyperv_utilsfactory.py Normal file
View File

@ -0,0 +1,51 @@
# 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)

215
neutron/tests/unit/hyperv/test_hyperv_utilsv2.py Normal file
View File

@ -0,0 +1,215 @@
# 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))