openstack/neutron-lbaas
Code Issues Proposed changes

Merge "Minor refactoring for Hyper-V utils and tests"

Browse Source
This commit is contained in:
Jenkins 2014-03-05 08:27:55 +00:00 committed by Gerrit Code Review
commit ac8c0c645d
2 changed files with 48 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
neutron/plugins/hyperv/agent/utilsv2.py
View File

@ -204,18 +204,12 @@ class HyperVUtilsV2(utils.HyperVUtils):
acls = port.associators(wmi_result_class=self._PORT_ALLOC_ACL_SET_DATA) acls = port.associators(wmi_result_class=self._PORT_ALLOC_ACL_SET_DATA)
for acl_type in [self._ACL_TYPE_IPV4, self._ACL_TYPE_IPV6]: for acl_type in [self._ACL_TYPE_IPV4, self._ACL_TYPE_IPV6]:
for acl_dir in [self._ACL_DIR_IN, self._ACL_DIR_OUT]: for acl_dir in [self._ACL_DIR_IN, self._ACL_DIR_OUT]:
acls = [v for v in acls _acls = self._filter_acls(
if v.Action == self._ACL_ACTION_METER and acls, self._ACL_ACTION_METER, acl_dir, acl_type)
v.Applicability == self._ACL_APPLICABILITY_LOCAL and
v.Direction == acl_dir and if not _acls:
v.AclType == acl_type] acl = self._create_acl(
if not acls: acl_dir, acl_type, self._ACL_ACTION_METER)
acl = self._get_default_setting_data(
self._PORT_ALLOC_ACL_SET_DATA)
acl.AclType = acl_type
acl.Direction = acl_dir
acl.Action = self._ACL_ACTION_METER
acl.Applicability = self._ACL_APPLICABILITY_LOCAL
self._add_virt_feature(port, acl) self._add_virt_feature(port, acl)
def create_security_rule(self, switch_port_name, direction, acl_type, def create_security_rule(self, switch_port_name, direction, acl_type,

107
neutron/tests/unit/hyperv/test_hyperv_utilsv2.py
View File

@ -88,83 +88,60 @@ class TestHyperVUtilsV2(base.BaseTestCase):
self._utils._modify_virt_resource.assert_called_with(mock_port) self._utils._modify_virt_resource.assert_called_with(mock_port)
def test_add_virt_resource(self): def test_add_virt_resource(self):
mock_svc = self._utils._conn.Msvm_VirtualSystemManagementService()[0] self._test_virt_method('AddResourceSettings', 3, '_add_virt_resource',
mock_svc.AddResourceSettings.return_value = (self._FAKE_JOB_PATH, True, self._FAKE_VM_PATH, [self._FAKE_RES_DATA])
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_add_virt_feature(self): def test_add_virt_feature(self):
mock_svc = self._utils._conn.Msvm_VirtualSystemManagementService()[0] self._test_virt_method('AddFeatureSettings', 3, '_add_virt_feature',
mock_svc.AddFeatureSettings.return_value = (self._FAKE_JOB_PATH, True, self._FAKE_VM_PATH, [self._FAKE_RES_DATA])
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_feature(mock_vm, mock_res_setting_data)
mock_svc.AddFeatureSettings.assert_called_once_with(
self._FAKE_VM_PATH, [self._FAKE_RES_DATA])
def test_modify_virt_resource(self): def test_modify_virt_resource(self):
mock_svc = self._utils._conn.Msvm_VirtualSystemManagementService()[0] self._test_virt_method('ModifyResourceSettings', 3,
mock_svc.ModifyResourceSettings.return_value = (self._FAKE_JOB_PATH, '_modify_virt_resource', False,
mock.MagicMock(), ResourceSettings=[self._FAKE_RES_DATA])
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): def test_remove_virt_resource(self):
self._test_virt_method('RemoveResourceSettings', 2,
'_remove_virt_resource', False,
ResourceSettings=[self._FAKE_RES_PATH])
def test_remove_virt_feature(self):
self._test_virt_method('RemoveFeatureSettings', 2,
'_remove_virt_feature', False,
FeatureSettings=[self._FAKE_RES_PATH])
def _test_virt_method(self, vsms_method_name, return_count,
utils_method_name, with_mock_vm, *args, **kwargs):
mock_svc = self._utils._conn.Msvm_VirtualSystemManagementService()[0] mock_svc = self._utils._conn.Msvm_VirtualSystemManagementService()[0]
mock_svc.RemoveResourceSettings.return_value = (self._FAKE_JOB_PATH, vsms_method = getattr(mock_svc, vsms_method_name)
self._FAKE_RET_VAL) mock_rsd = self._mock_vsms_method(vsms_method, return_count)
if with_mock_vm:
mock_vm = mock.MagicMock()
mock_vm.path_.return_value = self._FAKE_VM_PATH
getattr(self._utils, utils_method_name)(mock_vm, mock_rsd)
else:
getattr(self._utils, utils_method_name)(mock_rsd)
if args:
vsms_method.assert_called_once_with(*args)
else:
vsms_method.assert_called_once_with(**kwargs)
def _mock_vsms_method(self, vsms_method, return_count):
args = None
if return_count == 3:
args = (self._FAKE_JOB_PATH, mock.MagicMock(), self._FAKE_RET_VAL)
else:
args = (self._FAKE_JOB_PATH, self._FAKE_RET_VAL)
vsms_method.return_value = args
mock_res_setting_data = mock.MagicMock() mock_res_setting_data = mock.MagicMock()
mock_res_setting_data.GetText_.return_value = self._FAKE_RES_DATA
mock_res_setting_data.path_.return_value = self._FAKE_RES_PATH mock_res_setting_data.path_.return_value = self._FAKE_RES_PATH
self._utils._check_job_status = mock.MagicMock() self._utils._check_job_status = mock.MagicMock()
self._utils._remove_virt_resource(mock_res_setting_data) return mock_res_setting_data
mock_svc.RemoveResourceSettings.assert_called_with(
ResourceSettings=[self._FAKE_RES_PATH])
@mock.patch('neutron.plugins.hyperv.agent.utilsv2.HyperVUtilsV2'
'._check_job_status')
def test_remove_virt_feature(self, mock_check_job_status):
mock_svc = self._utils._conn.Msvm_VirtualSystemManagementService()[0]
mock_svc.RemoveFeatureSettings.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._remove_virt_feature(mock_res_setting_data)
mock_svc.RemoveFeatureSettings.assert_called_with(
FeatureSettings=[self._FAKE_RES_PATH])
def test_disconnect_switch_port_delete_port(self): def test_disconnect_switch_port_delete_port(self):
self._test_disconnect_switch_port(True) self._test_disconnect_switch_port(True)