1# (c) Copyright 2014 Cisco Systems Inc. 2# All Rights Reserved. 3# 4# Licensed under the Apache License, Version 2.0 (the "License"); you may 5# not use this file except in compliance with the License. You may obtain 6# a copy of the License at 7# 8# http://www.apache.org/licenses/LICENSE-2.0 9# 10# Unless required by applicable law or agreed to in writing, software 11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 13# License for the specific language governing permissions and limitations 14# under the License. 15# 16 17 18"""Unit tests for Cisco fc san lookup service.""" 19 20import mock 21from oslo_config import cfg 22 23from cinder import exception 24from cinder import test 25from cinder.volume import configuration as conf 26import cinder.zonemanager.drivers.cisco.cisco_fc_san_lookup_service \ 27 as cisco_lookup 28import cinder.zonemanager.drivers.cisco.fc_zone_constants as ZoneConstant 29from cinder.zonemanager import utils as zm_utils 30 31nsshow = '20:1a:00:05:1e:e8:e3:29' 32switch_data = ['VSAN 304\n', 33 '------------------------------------------------------\n', 34 'FCID TYPE PWWN (VENDOR) \n', 35 '------------------------------------------------------\n', 36 '0x030001 N 20:1a:00:05:1e:e8:e3:29 (Cisco) ipfc\n', 37 '0x030101 NL 10:00:00:00:77:99:60:2c (Interphase)\n', 38 '0x030200 N 10:00:00:49:c9:28:c7:01\n'] 39 40nsshow_data = ['10:00:8c:7c:ff:52:3b:01', '20:24:00:02:ac:00:0a:50'] 41 42_device_map_to_verify = { 43 'CISCO_FAB_2': { 44 'initiator_port_wwn_list': ['10008c7cff523b01'], 45 'target_port_wwn_list': ['20240002ac000a50']}} 46 47 48class TestCiscoFCSanLookupService(cisco_lookup.CiscoFCSanLookupService, 49 test.TestCase): 50 51 def setUp(self): 52 super(TestCiscoFCSanLookupService, self).setUp() 53 self.configuration = conf.Configuration(None) 54 self.configuration.set_default('fc_fabric_names', 'CISCO_FAB_2', 55 'fc-zone-manager') 56 self.configuration.fc_fabric_names = 'CISCO_FAB_2' 57 self.create_configuration() 58 self.fabric_vsan = '304' 59 60 # override some of the functions 61 def __init__(self, *args, **kwargs): 62 test.TestCase.__init__(self, *args, **kwargs) 63 64 def create_configuration(self): 65 fc_fabric_opts = [] 66 fc_fabric_opts.append(cfg.StrOpt('cisco_fc_fabric_address', 67 default='172.24.173.142', help='')) 68 fc_fabric_opts.append(cfg.StrOpt('cisco_fc_fabric_user', 69 default='admin', help='')) 70 fc_fabric_opts.append(cfg.StrOpt('cisco_fc_fabric_password', 71 default='admin1234', help='', 72 secret=True)) 73 fc_fabric_opts.append(cfg.PortOpt('cisco_fc_fabric_port', 74 default=22, help='')) 75 fc_fabric_opts.append(cfg.StrOpt('cisco_zoning_vsan', 76 default='304', help='')) 77 config = conf.Configuration(fc_fabric_opts, 'CISCO_FAB_2') 78 self.fabric_configs = {'CISCO_FAB_2': config} 79 80 @mock.patch.object(cisco_lookup.CiscoFCSanLookupService, 81 'get_nameserver_info') 82 def test_get_device_mapping_from_network(self, get_nameserver_info_mock): 83 initiator_list = ['10008c7cff523b01'] 84 target_list = ['20240002ac000a50', '20240002ac000a40'] 85 get_nameserver_info_mock.return_value = (nsshow_data) 86 device_map = self.get_device_mapping_from_network( 87 initiator_list, target_list) 88 self.assertDictEqual(_device_map_to_verify, device_map) 89 90 @mock.patch.object(cisco_lookup.CiscoFCSanLookupService, 91 '_get_switch_info') 92 def test_get_nameserver_info(self, get_switch_data_mock): 93 ns_info_list = [] 94 ns_info_list_expected = ['20:1a:00:05:1e:e8:e3:29', 95 '10:00:00:49:c9:28:c7:01'] 96 get_switch_data_mock.return_value = (switch_data) 97 ns_info_list = self.get_nameserver_info('304') 98 self.assertEqual(ns_info_list_expected, ns_info_list) 99 100 def test_parse_ns_output(self): 101 invalid_switch_data = [' N 011a00;20:1a:00:05:1e:e8:e3:29'] 102 return_wwn_list = [] 103 expected_wwn_list = ['20:1a:00:05:1e:e8:e3:29', 104 '10:00:00:49:c9:28:c7:01'] 105 return_wwn_list = self._parse_ns_output(switch_data) 106 self.assertEqual(expected_wwn_list, return_wwn_list) 107 self.assertRaises(exception.InvalidParameterValue, 108 self._parse_ns_output, invalid_switch_data) 109 110 def test_get_formatted_wwn(self): 111 wwn_list = ['10008c7cff523b01'] 112 return_wwn_list = [] 113 expected_wwn_list = ['10:00:8c:7c:ff:52:3b:01'] 114 return_wwn_list.append(zm_utils.get_formatted_wwn(wwn_list[0])) 115 self.assertEqual(expected_wwn_list, return_wwn_list) 116 117 @mock.patch.object(cisco_lookup.CiscoFCSanLookupService, 118 '_run_ssh') 119 def test__get_switch_info(self, run_ssh_mock): 120 cmd_list = [ZoneConstant.FCNS_SHOW, self.fabric_vsan, 121 ' | no-more'] 122 nsshow_list = [nsshow] 123 run_ssh_mock.return_value = (Stream(nsshow), Stream()) 124 switch_data = self._get_switch_info(cmd_list) 125 self.assertEqual(nsshow_list, switch_data) 126 run_ssh_mock.assert_called_once_with(cmd_list, True, 1) 127 128 129class Channel(object): 130 def recv_exit_status(self): 131 return 0 132 133 134class Stream(object): 135 def __init__(self, buffer=''): 136 self.buffer = buffer 137 self.channel = Channel() 138 139 def readlines(self): 140 return self.buffer 141 142 def splitlines(self): 143 return self.buffer.splitlines() 144 145 def close(self): 146 pass 147 148 def flush(self): 149 self.buffer = '' 150