¡@
OpenStack Study: test_brcd_fc_zone_client_cli.py**** CubicPower OpenStack Study ****
# (c) Copyright 2014 Brocade Communications Systems Inc. # All Rights Reserved. # # Copyright 2014 OpenStack Foundation # # 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. # """Unit tests for brcd fc zone client cli.""" import mock from cinder import exception from cinder.openstack.common import log as logging from cinder.openstack.common import processutils from cinder import test from cinder.zonemanager.drivers.brocade.brcd_fc_zone_client_cli \ import BrcdFCZoneClientCLI import cinder.zonemanager.drivers.brocade.fc_zone_constants as ZoneConstant from mock import patch LOG = logging.getLogger(__name__) nsshow = '20:1a:00:05:1e:e8:e3:29' switch_data = [' N 011a00;2,3;20:1a:00:05:1e:e8:e3:29;\ 20:1a:00:05:1e:e8:e3:29;na', ' Fabric Port Name: 20:1a:00:05:1e:e8:e3:29'] cfgactvshow = ['Effective configuration:\n', ' cfg:\tOpenStack_Cfg\t\n', ' zone:\topenstack50060b0000c26604201900051ee8e329\t\n', '\t\t50:06:0b:00:00:c2:66:04\n', '\t\t20:19:00:05:1e:e8:e3:29\n'] active_zoneset = { 'zones': { 'openstack50060b0000c26604201900051ee8e329': ['50:06:0b:00:00:c2:66:04', '20:19:00:05:1e:e8:e3:29']}, 'active_zone_config': 'OpenStack_Cfg'} active_zoneset_multiple_zones = { 'zones': { 'openstack50060b0000c26604201900051ee8e329': ['50:06:0b:00:00:c2:66:04', '20:19:00:05:1e:e8:e3:29'], 'openstack50060b0000c26602201900051ee8e327': ['50:06:0b:00:00:c2:66:02', '20:19:00:05:1e:e8:e3:27']}, 'active_zone_config': 'OpenStack_Cfg'} new_zone = {'openstack10000012345678902001009876543210': ['10:00:00:12:34:56:78:90', '20:01:00:98:76:54:32:10']} new_zones = {'openstack10000012345678902001009876543210': ['10:00:00:12:34:56:78:90', '20:01:00:98:76:54:32:10'], 'openstack10000011111111112001001111111111': ['10:00:00:11:11:11:11:11', '20:01:00:11:11:11:11:11']} zone_names_to_delete = 'openstack50060b0000c26604201900051ee8e329' supported_firmware = ['Kernel: 2.6', 'Fabric OS: v7.0.1'] unsupported_firmware = ['Fabric OS: v6.2.1']
**** CubicPower OpenStack Study **** class TestBrcdFCZoneClientCLI(BrcdFCZoneClientCLI, test.TestCase):
**** CubicPower OpenStack Study **** def setUp(self): super(TestBrcdFCZoneClientCLI, self).setUp() # override some of the functions
**** CubicPower OpenStack Study **** def __init__(self, *args, **kwargs): test.TestCase.__init__(self, *args, **kwargs) @patch.object(BrcdFCZoneClientCLI, '_get_switch_info')
**** CubicPower OpenStack Study **** def test_get_active_zone_set(self, get_switch_info_mock): cmd_list = [ZoneConstant.GET_ACTIVE_ZONE_CFG] get_switch_info_mock.return_value = cfgactvshow active_zoneset_returned = self.get_active_zone_set() get_switch_info_mock.assert_called_once_with(cmd_list) self.assertDictMatch(active_zoneset_returned, active_zoneset) @patch.object(BrcdFCZoneClientCLI, '_run_ssh')
**** CubicPower OpenStack Study **** def test_get_active_zone_set_ssh_error(self, run_ssh_mock): run_ssh_mock.side_effect = processutils.ProcessExecutionError self.assertRaises(exception.BrocadeZoningCliException, self.get_active_zone_set) @mock.patch.object(BrcdFCZoneClientCLI, 'get_active_zone_set') @mock.patch.object(BrcdFCZoneClientCLI, 'apply_zone_change') @mock.patch.object(BrcdFCZoneClientCLI, '_cfg_save')
**** CubicPower OpenStack Study **** def test_add_zones_new_zone_no_activate(self, get_active_zs_mock, apply_zone_change_mock, cfg_save_mock): get_active_zs_mock.return_value = active_zoneset self.add_zones(new_zones, False) get_active_zs_mock.assert_called_once() apply_zone_change_mock.assert_called_twice() cfg_save_mock.assert_called_once() @mock.patch.object(BrcdFCZoneClientCLI, 'get_active_zone_set') @mock.patch.object(BrcdFCZoneClientCLI, 'apply_zone_change') @mock.patch.object(BrcdFCZoneClientCLI, '_cfg_save') @mock.patch.object(BrcdFCZoneClientCLI, 'activate_zoneset')
**** CubicPower OpenStack Study **** def test_add_zones_new_zone_activate(self, get_active_zs_mock, apply_zone_change_mock, cfg_save_mock, activate_zoneset_mock): get_active_zs_mock.return_value = active_zoneset self.add_zones(new_zone, True) get_active_zs_mock.assert_called_once() apply_zone_change_mock.assert_called_once() cfg_save_mock.assert_called_once() activate_zoneset_mock.assert_called_once() @mock.patch.object(BrcdFCZoneClientCLI, '_ssh_execute')
**** CubicPower OpenStack Study **** def test_activate_zoneset(self, ssh_execute_mock): ssh_execute_mock.return_value = True return_value = self.activate_zoneset('zoneset1') self.assertTrue(return_value) @mock.patch.object(BrcdFCZoneClientCLI, '_ssh_execute')
**** CubicPower OpenStack Study **** def test_deactivate_zoneset(self, ssh_execute_mock): ssh_execute_mock.return_value = True return_value = self.deactivate_zoneset() self.assertTrue(return_value) @mock.patch.object(BrcdFCZoneClientCLI, 'get_active_zone_set') @mock.patch.object(BrcdFCZoneClientCLI, 'apply_zone_change') @mock.patch.object(BrcdFCZoneClientCLI, '_cfg_save')
**** CubicPower OpenStack Study **** def test_delete_zones_activate_false(self, get_active_zs_mock, apply_zone_change_mock, cfg_save_mock): get_active_zs_mock.return_value = active_zoneset_multiple_zones with mock.patch.object(self, '_zone_delete') \ as zone_delete_mock: self.delete_zones(zone_names_to_delete, False) get_active_zs_mock.assert_called_once() apply_zone_change_mock.assert_called_once() zone_delete_mock.assert_called_once_with(zone_names_to_delete) cfg_save_mock.assert_called_once() @patch.object(BrcdFCZoneClientCLI, 'get_active_zone_set') @patch.object(BrcdFCZoneClientCLI, 'apply_zone_change') @patch.object(BrcdFCZoneClientCLI, '_cfg_save') @patch.object(BrcdFCZoneClientCLI, 'activate_zoneset')
**** CubicPower OpenStack Study **** def test_delete_zones_activate_true(self, get_active_zs_mock, apply_zone_change_mock, cfg_save_mock, activate_zs_mock): get_active_zs_mock.return_value = active_zoneset_multiple_zones with mock.patch.object(self, '_zone_delete') \ as zone_delete_mock: self.delete_zones(zone_names_to_delete, True) get_active_zs_mock.assert_called_once() apply_zone_change_mock.assert_called_once() zone_delete_mock.assert_called_once_with(zone_names_to_delete) cfg_save_mock.assert_called_once() activate_zs_mock.assert_called_once() @patch.object(BrcdFCZoneClientCLI, '_get_switch_info')
**** CubicPower OpenStack Study **** def test_get_nameserver_info(self, get_switch_info_mock): ns_info_list = [] ns_info_list_expected = ['20:1a:00:05:1e:e8:e3:29', '20:1a:00:05:1e:e8:e3:29'] get_switch_info_mock.return_value = (switch_data) ns_info_list = self.get_nameserver_info() self.assertEqual(ns_info_list, ns_info_list_expected) @patch.object(BrcdFCZoneClientCLI, '_run_ssh')
**** CubicPower OpenStack Study **** def test_get_nameserver_info_ssh_error(self, run_ssh_mock): run_ssh_mock.side_effect = processutils.ProcessExecutionError self.assertRaises(exception.BrocadeZoningCliException, self.get_nameserver_info) @patch.object(BrcdFCZoneClientCLI, '_ssh_execute')
**** CubicPower OpenStack Study **** def test__cfg_save(self, ssh_execute_mock): cmd_list = [ZoneConstant.CFG_SAVE] self._cfg_save() ssh_execute_mock.assert_called_once_with(cmd_list, True, 1) @patch.object(BrcdFCZoneClientCLI, 'apply_zone_change')
**** CubicPower OpenStack Study **** def test__zone_delete(self, apply_zone_change_mock): zone_name = 'testzone' cmd_list = ['zonedelete', '"testzone"'] self._zone_delete(zone_name) apply_zone_change_mock.assert_called_once_with(cmd_list) @patch.object(BrcdFCZoneClientCLI, 'apply_zone_change')
**** CubicPower OpenStack Study **** def test__cfg_trans_abort(self, apply_zone_change_mock): cmd_list = [ZoneConstant.CFG_ZONE_TRANS_ABORT] with mock.patch.object(self, '_is_trans_abortable') \ as is_trans_abortable_mock: is_trans_abortable_mock.return_value = True self._cfg_trans_abort() is_trans_abortable_mock.assert_called_once() apply_zone_change_mock.assert_called_once_with(cmd_list) @patch.object(BrcdFCZoneClientCLI, '_run_ssh')
**** CubicPower OpenStack Study **** def test__is_trans_abortable_true(self, run_ssh_mock): cmd_list = [ZoneConstant.CFG_SHOW_TRANS] run_ssh_mock.return_value = (Stream(ZoneConstant.TRANS_ABORTABLE), None) data = self._is_trans_abortable() self.assertTrue(data) run_ssh_mock.assert_called_once_with(cmd_list, True, 1) @patch.object(BrcdFCZoneClientCLI, '_run_ssh')
**** CubicPower OpenStack Study **** def test__is_trans_abortable_ssh_error(self, run_ssh_mock): run_ssh_mock.return_value = (Stream(), Stream()) self.assertRaises(exception.BrocadeZoningCliException, self._is_trans_abortable) @patch.object(BrcdFCZoneClientCLI, '_run_ssh')
**** CubicPower OpenStack Study **** def test__is_trans_abortable_false(self, run_ssh_mock): cmd_list = [ZoneConstant.CFG_SHOW_TRANS] cfgtransshow = 'There is no outstanding zoning transaction' run_ssh_mock.return_value = (Stream(cfgtransshow), None) data = self._is_trans_abortable() self.assertFalse(data) run_ssh_mock.assert_called_once_with(cmd_list, True, 1) @patch.object(BrcdFCZoneClientCLI, '_run_ssh')
**** CubicPower OpenStack Study **** def test_apply_zone_change(self, run_ssh_mock): cmd_list = [ZoneConstant.CFG_SAVE] run_ssh_mock.return_value = (None, None) self.apply_zone_change(cmd_list) run_ssh_mock.assert_called_once_with(cmd_list, True, 1) @patch.object(BrcdFCZoneClientCLI, '_run_ssh')
**** CubicPower OpenStack Study **** def test__get_switch_info(self, run_ssh_mock): cmd_list = [ZoneConstant.NS_SHOW] nsshow_list = [nsshow] run_ssh_mock.return_value = (Stream(nsshow), Stream()) switch_data = self._get_switch_info(cmd_list) self.assertEqual(switch_data, nsshow_list) run_ssh_mock.assert_called_once_with(cmd_list, True, 1)
**** CubicPower OpenStack Study **** def test__parse_ns_output(self): invalid_switch_data = [' N 011a00;20:1a:00:05:1e:e8:e3:29'] return_wwn_list = [] expected_wwn_list = ['20:1a:00:05:1e:e8:e3:29'] return_wwn_list = self._parse_ns_output(switch_data) self.assertEqual(return_wwn_list, expected_wwn_list) self.assertRaises(exception.InvalidParameterValue, self._parse_ns_output, invalid_switch_data) @patch.object(BrcdFCZoneClientCLI, '_execute_shell_cmd')
**** CubicPower OpenStack Study **** def test_is_supported_firmware(self, exec_shell_cmd_mock): exec_shell_cmd_mock.return_value = (supported_firmware, None) self.assertTrue(self.is_supported_firmware()) @patch.object(BrcdFCZoneClientCLI, '_execute_shell_cmd')
**** CubicPower OpenStack Study **** def test_is_supported_firmware_invalid(self, exec_shell_cmd_mock): exec_shell_cmd_mock.return_value = (unsupported_firmware, None) self.assertFalse(self.is_supported_firmware()) @patch.object(BrcdFCZoneClientCLI, '_execute_shell_cmd')
**** CubicPower OpenStack Study **** def test_is_supported_firmware_no_ssh_response(self, exec_shell_cmd_mock): exec_shell_cmd_mock.return_value = (None, Stream()) self.assertFalse(self.is_supported_firmware()) @patch.object(BrcdFCZoneClientCLI, '_execute_shell_cmd')
**** CubicPower OpenStack Study **** def test_is_supported_firmware_ssh_error(self, exec_shell_cmd_mock): exec_shell_cmd_mock.side_effect = processutils.ProcessExecutionError self.assertRaises(exception.BrocadeZoningCliException, self.is_supported_firmware)
**** CubicPower OpenStack Study **** class Channel(object):
**** CubicPower OpenStack Study **** def recv_exit_status(self): return 0
**** CubicPower OpenStack Study **** class Stream(object):
**** CubicPower OpenStack Study **** def __init__(self, buffer=''): self.buffer = buffer self.channel = Channel()
**** CubicPower OpenStack Study **** def readlines(self): return self.buffer
**** CubicPower OpenStack Study **** def splitlines(self): return self.buffer.splitlines()
**** CubicPower OpenStack Study **** def close(self): pass
**** CubicPower OpenStack Study **** def flush(self): self.buffer = ''
|