Coverage for manila/tests/share/drivers/dell_emc/common/enas/test_connector.py: 100%

1# Copyright (c) 2015 EMC Corporation. 

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 

16from unittest import mock 

17from urllib import error as url_error 

18from urllib import request as url_request 

19 

20from eventlet import greenthread 

21from oslo_concurrency import processutils 

22 

23from manila import exception 

24from manila.share import configuration as conf 

25from manila.share.drivers.dell_emc.common.enas import connector 

26from manila import ssh_utils 

27from manila import test 

28from manila.tests.share.drivers.dell_emc.common.enas import fakes 

29from manila.tests.share.drivers.dell_emc.common.enas import utils as enas_utils 

30 

31 

32class XMLAPIConnectorTestData(object): 

33 FAKE_BODY = '<fakebody></fakebody>' 

34 FAKE_RESP = '<Response></Response>' 

35 FAKE_METHOD = 'fake_method' 

36 

37 FAKE_KEY = 'key' 

38 FAKE_VALUE = 'value' 

39 

40 @staticmethod 

41 def req_auth_url(): 

42 return 'https://' + fakes.FakeData.emc_nas_server + '/Login' 

43 

44 @staticmethod 

45 def req_credential(): 

46 return ( 

47 'user=' + fakes.FakeData.emc_nas_login 

48 + '&password=' + fakes.FakeData.emc_nas_password 

49 + '&Login=Login' 

50 ).encode() 

51 

52 @staticmethod 

53 def req_url_encode(): 

54 return {'Content-Type': 'application/x-www-form-urlencoded'} 

55 

56 @staticmethod 

57 def req_url(): 

58 return ( 

59 'https://' 

60 + fakes.FakeData.emc_nas_server 

61 + '/servlets/CelerraManagementServices' 

62 ) 

63 

64 

65XML_CONN_TD = XMLAPIConnectorTestData 

66 

67 

68class XMLAPIConnectorTest(test.TestCase): 

69 @mock.patch.object(url_request, 'Request', mock.Mock()) 

70 def setUp(self): 

71 super(XMLAPIConnectorTest, self).setUp() 

72 

73 emc_share_driver = fakes.FakeEMCShareDriver() 

74 

75 self.configuration = emc_share_driver.configuration 

76 

77 xml_socket = mock.Mock() 

78 xml_socket.read = mock.Mock(return_value=XML_CONN_TD.FAKE_RESP) 

79 opener = mock.Mock() 

80 opener.open = mock.Mock(return_value=xml_socket) 

81 

82 with mock.patch.object(url_request, 'build_opener', 

83 mock.Mock(return_value=opener)): 

84 self.XmlConnector = connector.XMLAPIConnector( 

85 configuration=self.configuration, debug=False) 

86 

87 expected_calls = [ 

88 mock.call(XML_CONN_TD.req_auth_url(), 

89 XML_CONN_TD.req_credential(), 

90 XML_CONN_TD.req_url_encode()), 

91 ] 

92 

93 url_request.Request.assert_has_calls(expected_calls) 

94 

95 def test_request_with_debug(self): 

96 self.XmlConnector.debug = True 

97 

98 request = mock.Mock() 

99 request.headers = {XML_CONN_TD.FAKE_KEY: XML_CONN_TD.FAKE_VALUE} 

100 request.get_full_url = mock.Mock( 

101 return_value=XML_CONN_TD.FAKE_VALUE) 

102 

103 with mock.patch.object(url_request, 'Request', 

104 mock.Mock(return_value=request)): 

105 rsp = self.XmlConnector.request(XML_CONN_TD.FAKE_BODY, 

106 XML_CONN_TD.FAKE_METHOD) 

107 

108 self.assertEqual(XML_CONN_TD.FAKE_RESP, rsp) 

109 

110 def test_request_with_no_authorized_exception(self): 

111 xml_socket = mock.Mock() 

112 xml_socket.read = mock.Mock(return_value=XML_CONN_TD.FAKE_RESP) 

113 

114 hook = enas_utils.RequestSideEffect() 

115 hook.append(ex=url_error.HTTPError(XML_CONN_TD.req_url(), 

116 '403', 'fake_message', None, None)) 

117 hook.append(xml_socket) 

118 hook.append(xml_socket) 

119 

120 self.XmlConnector.url_opener.open = mock.Mock(side_effect=hook) 

121 

122 self.XmlConnector.request(XML_CONN_TD.FAKE_BODY) 

123 

124 def test_request_with_general_exception(self): 

125 hook = enas_utils.RequestSideEffect() 

126 hook.append(ex=url_error.HTTPError(XML_CONN_TD.req_url(), 

127 'error_code', 'fake_message', 

128 None, None)) 

129 self.XmlConnector.url_opener.open = mock.Mock(side_effect=hook) 

130 

131 self.assertRaises(exception.ManilaException, 

132 self.XmlConnector.request, 

133 XML_CONN_TD.FAKE_BODY) 

134 

135 

136class MockSSH(object): 

137 def __enter__(self): 

138 return self 

139 

140 def __exit__(self, type, value, traceback): 

141 pass 

142 

143 

144class MockSSHPool(object): 

145 def __init__(self): 

146 self.ssh = MockSSH() 

147 

148 def item(self): 

149 try: 

150 return self.ssh 

151 finally: 

152 pass 

153 

154 

155class CmdConnectorTest(test.TestCase): 

156 def setUp(self): 

157 super(CmdConnectorTest, self).setUp() 

158 

159 self.configuration = conf.Configuration(None) 

160 self.configuration.append_config_values = mock.Mock(return_value=0) 

161 self.configuration.emc_nas_login = fakes.FakeData.emc_nas_login 

162 self.configuration.emc_nas_password = fakes.FakeData.emc_nas_password 

163 self.configuration.emc_nas_server = fakes.FakeData.emc_nas_server 

164 self.configuration.emc_ssl_cert_verify = False 

165 self.configuration.emc_ssl_cert_path = None 

166 

167 self.sshpool = MockSSHPool() 

168 with mock.patch.object(ssh_utils, "SSHPool", 

169 mock.Mock(return_value=self.sshpool)): 

170 self.CmdHelper = connector.SSHConnector( 

171 configuration=self.configuration, debug=False) 

172 

173 ssh_utils.SSHPool.assert_called_once_with( 

174 ip=fakes.FakeData.emc_nas_server, 

175 port=22, 

176 conn_timeout=None, 

177 login=fakes.FakeData.emc_nas_login, 

178 password=fakes.FakeData.emc_nas_password) 

179 

180 def test_run_ssh(self): 

181 with mock.patch.object(processutils, "ssh_execute", 

182 mock.Mock(return_value=('fake_output', ''))): 

183 cmd_list = ['fake', 'cmd'] 

184 self.CmdHelper.run_ssh(cmd_list) 

185 

186 processutils.ssh_execute.assert_called_once_with( 

187 self.sshpool.item(), 'fake cmd', check_exit_code=False) 

188 

189 def test_run_ssh_with_debug(self): 

190 self.CmdHelper.debug = True 

191 

192 with mock.patch.object(processutils, "ssh_execute", 

193 mock.Mock(return_value=('fake_output', ''))): 

194 cmd_list = ['fake', 'cmd'] 

195 self.CmdHelper.run_ssh(cmd_list) 

196 

197 processutils.ssh_execute.assert_called_once_with( 

198 self.sshpool.item(), 'fake cmd', check_exit_code=False) 

199 

200 @mock.patch.object( 

201 processutils, "ssh_execute", 

202 mock.Mock(side_effect=processutils.ProcessExecutionError)) 

203 def test_run_ssh_exception(self): 

204 cmd_list = ['fake', 'cmd'] 

205 

206 self.mock_object(greenthread, 'sleep', mock.Mock()) 

207 

208 sshpool = MockSSHPool() 

209 

210 with mock.patch.object(ssh_utils, "SSHPool", 

211 mock.Mock(return_value=sshpool)): 

212 self.CmdHelper = connector.SSHConnector(self.configuration) 

213 

214 self.assertRaises(processutils.ProcessExecutionError, 

215 self.CmdHelper.run_ssh, 

216 cmd_list, 

217 True) 

218 

219 ssh_utils.SSHPool.assert_called_once_with( 

220 ip=fakes.FakeData.emc_nas_server, 

221 port=22, 

222 conn_timeout=None, 

223 login=fakes.FakeData.emc_nas_login, 

224 password=fakes.FakeData.emc_nas_password) 

225 

226 processutils.ssh_execute.assert_called_once_with( 

227 sshpool.item(), 'fake cmd', check_exit_code=True)