Coverage for manila/tests/share/drivers/windows/test_winrm_helper.py: 100%

1# Copyright (c) 2015 Cloudbase Solutions SRL 

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 

17 

18import ddt 

19from oslo_concurrency import processutils 

20from oslo_utils import importutils 

21from oslo_utils import strutils 

22 

23from manila import exception 

24from manila.share.drivers.windows import winrm_helper 

25from manila import test 

26 

27 

28@ddt.ddt 

29class WinRMHelperTestCase(test.TestCase): 

30 _FAKE_SERVER = {'ip': mock.sentinel.ip} 

31 

32 @mock.patch.object(importutils, 'import_module') 

33 def setUp(self, mock_import_module): 

34 self._winrm = winrm_helper.WinRMHelper() 

35 super(WinRMHelperTestCase, self).setUp() 

36 

37 @ddt.data({'import_exc': None}, 

38 {'import_exc': ImportError}) 

39 @mock.patch.object(importutils, 'import_module') 

40 @ddt.unpack 

41 def test_setup_winrm(self, mock_import_module, import_exc): 

42 winrm_helper.winrm = None 

43 mock_import_module.side_effect = import_exc 

44 

45 if import_exc: 

46 self.assertRaises(exception.ShareBackendException, 

47 winrm_helper.setup_winrm) 

48 else: 

49 winrm_helper.setup_winrm() 

50 self.assertEqual(mock_import_module.return_value, 

51 winrm_helper.winrm) 

52 mock_import_module.assert_called_once_with('winrm') 

53 

54 @mock.patch.object(winrm_helper.WinRMHelper, '_get_auth') 

55 @mock.patch.object(winrm_helper, 'WinRMConnection') 

56 def test_get_conn(self, mock_conn_cls, mock_get_auth): 

57 mock_auth = {'mock_auth_key': mock.sentinel.auth_opt} 

58 mock_get_auth.return_value = mock_auth 

59 

60 conn = self._winrm._get_conn(self._FAKE_SERVER) 

61 

62 mock_get_auth.assert_called_once_with(self._FAKE_SERVER) 

63 mock_conn_cls.assert_called_once_with( 

64 ip=self._FAKE_SERVER['ip'], 

65 conn_timeout=self._winrm._config.winrm_conn_timeout, 

66 operation_timeout=self._winrm._config.winrm_operation_timeout, 

67 **mock_auth) 

68 self.assertEqual(mock_conn_cls.return_value, conn) 

69 

70 @ddt.data({}, 

71 {'exit_code': 1}, 

72 {'exit_code': 1, 'check_exit_code': False}) 

73 @mock.patch.object(strutils, 'mask_password') 

74 @mock.patch.object(winrm_helper.WinRMHelper, '_parse_command') 

75 @mock.patch.object(winrm_helper.WinRMHelper, '_get_conn') 

76 @ddt.unpack 

77 def test_execute(self, mock_get_conn, mock_parse_command, 

78 mock_mask_password, 

79 check_exit_code=True, exit_code=0): 

80 mock_parse_command.return_value = (mock.sentinel.parsed_cmd, 

81 mock.sentinel.sanitized_cmd) 

82 mock_conn = mock_get_conn.return_value 

83 mock_conn.execute.return_value = (mock.sentinel.stdout, 

84 mock.sentinel.stderr, 

85 exit_code) 

86 

87 if exit_code == 0 or not check_exit_code: 

88 result = self._winrm.execute(mock.sentinel.server, 

89 mock.sentinel.command, 

90 check_exit_code=check_exit_code, 

91 retry=False) 

92 expected_result = (mock.sentinel.stdout, mock.sentinel.stderr) 

93 self.assertEqual(expected_result, result) 

94 else: 

95 self.assertRaises(processutils.ProcessExecutionError, 

96 self._winrm.execute, 

97 mock.sentinel.server, 

98 mock.sentinel.command, 

99 check_exit_code=check_exit_code, 

100 retry=False) 

101 

102 mock_get_conn.assert_called_once_with(mock.sentinel.server) 

103 mock_parse_command.assert_called_once_with(mock.sentinel.command) 

104 mock_conn.execute.assert_called_once_with(mock.sentinel.parsed_cmd) 

105 mock_mask_password.assert_has_calls([mock.call(mock.sentinel.stdout), 

106 mock.call(mock.sentinel.stderr)]) 

107 

108 @mock.patch('base64.b64encode') 

109 @mock.patch.object(strutils, 'mask_password') 

110 def test_parse_command(self, mock_mask_password, mock_base64): 

111 mock_mask_password.return_value = mock.sentinel.sanitized_cmd 

112 mock_base64.return_value = mock.sentinel.encoded_string 

113 

114 cmd = ('Get-Disk', '-Number', 1) 

115 result = self._winrm._parse_command(cmd) 

116 

117 joined_cmd = 'Get-Disk -Number 1' 

118 expected_command = ("powershell.exe -ExecutionPolicy RemoteSigned " 

119 "-NonInteractive -EncodedCommand %s" % 

120 mock.sentinel.encoded_string) 

121 expected_result = expected_command, mock.sentinel.sanitized_cmd 

122 

123 mock_mask_password.assert_called_once_with(joined_cmd) 

124 mock_base64.assert_called_once_with(joined_cmd.encode("utf_16_le")) 

125 self.assertEqual(expected_result, result) 

126 

127 def _test_get_auth(self, use_cert_auth=False): 

128 mock_server = {'use_cert_auth': use_cert_auth, 

129 'cert_pem_path': mock.sentinel.pem_path, 

130 'cert_key_pem_path': mock.sentinel.key_path, 

131 'username': mock.sentinel.username, 

132 'password': mock.sentinel.password} 

133 

134 result = self._winrm._get_auth(mock_server) 

135 

136 expected_result = {'username': mock_server['username']} 

137 if use_cert_auth: 

138 expected_result['cert_pem_path'] = mock_server['cert_pem_path'] 

139 expected_result['cert_key_pem_path'] = ( 

140 mock_server['cert_key_pem_path']) 

141 else: 

142 expected_result['password'] = mock_server['password'] 

143 

144 self.assertEqual(expected_result, result) 

145 

146 def test_get_auth_using_certificates(self): 

147 self._test_get_auth(use_cert_auth=True) 

148 

149 def test_get_auth_using_password(self): 

150 self._test_get_auth() 

151 

152 

153class WinRMConnectionTestCase(test.TestCase): 

154 @mock.patch.object(winrm_helper, 'setup_winrm') 

155 @mock.patch.object(winrm_helper, 'winrm') 

156 @mock.patch.object(winrm_helper.WinRMConnection, '_get_url') 

157 @mock.patch.object(winrm_helper.WinRMConnection, '_get_default_port') 

158 def setUp(self, mock_get_port, mock_get_url, mock_winrm, 

159 mock_setup_winrm): 

160 self._winrm = winrm_helper.WinRMConnection() 

161 self._mock_conn = mock_winrm.protocol.Protocol.return_value 

162 super(WinRMConnectionTestCase, self).setUp() 

163 

164 @mock.patch.object(winrm_helper, 'setup_winrm') 

165 @mock.patch.object(winrm_helper, 'winrm') 

166 @mock.patch.object(winrm_helper.WinRMConnection, '_get_url') 

167 @mock.patch.object(winrm_helper.WinRMConnection, '_get_default_port') 

168 def test_init_conn(self, mock_get_port, mock_get_url, mock_winrm, 

169 mock_setup_winrm): 

170 # certificates are passed so we expect cert auth to be used 

171 cert_auth = True 

172 winrm_conn = winrm_helper.WinRMConnection( 

173 ip=mock.sentinel.ip, username=mock.sentinel.username, 

174 password=mock.sentinel.password, 

175 cert_pem_path=mock.sentinel.cert_pem_path, 

176 cert_key_pem_path=mock.sentinel.cert_key_pem_path, 

177 operation_timeout=mock.sentinel.operation_timeout, 

178 conn_timeout=mock.sentinel.conn_timeout) 

179 

180 mock_get_port.assert_called_once_with(cert_auth) 

181 mock_get_url.assert_called_once_with(mock.sentinel.ip, 

182 mock_get_port.return_value, 

183 cert_auth) 

184 mock_winrm.protocol.Protocol.assert_called_once_with( 

185 endpoint=mock_get_url.return_value, 

186 transport=winrm_helper.TRANSPORT_SSL, 

187 username=mock.sentinel.username, 

188 password=mock.sentinel.password, 

189 cert_pem=mock.sentinel.cert_pem_path, 

190 cert_key_pem=mock.sentinel.cert_key_pem_path) 

191 self.assertEqual(mock_winrm.protocol.Protocol.return_value, 

192 winrm_conn._conn) 

193 self.assertEqual(mock.sentinel.conn_timeout, 

194 winrm_conn._conn.transport.timeout) 

195 winrm_conn._conn.set_timeout.assert_called_once_with( 

196 mock.sentinel.operation_timeout) 

197 

198 def test_get_default_port_https(self): 

199 port = self._winrm._get_default_port(use_ssl=True) 

200 self.assertEqual(winrm_helper.DEFAULT_PORT_HTTPS, port) 

201 

202 def test_get_default_port_http(self): 

203 port = self._winrm._get_default_port(use_ssl=False) 

204 self.assertEqual(winrm_helper.DEFAULT_PORT_HTTP, port) 

205 

206 def _test_get_url(self, ip=None, use_ssl=True): 

207 if not ip: 

208 self.assertRaises(exception.ShareBackendException, 

209 self._winrm._get_url, 

210 ip=ip, 

211 port=mock.sentinel.port, 

212 use_ssl=use_ssl) 

213 else: 

214 url = self._winrm._get_url(ip=ip, 

215 port=mock.sentinel.port, 

216 use_ssl=use_ssl) 

217 expected_protocol = 'https' if use_ssl else 'http' 

218 expected_url = self._winrm._URL_TEMPLATE % dict( 

219 protocol=expected_protocol, 

220 port=mock.sentinel.port, 

221 ip=ip) 

222 self.assertEqual(expected_url, url) 

223 

224 def test_get_url_using_ssl(self): 

225 self._test_get_url(ip=mock.sentinel.ip) 

226 

227 def test_get_url_using_plaintext(self): 

228 self._test_get_url(ip=mock.sentinel.ip, use_ssl=False) 

229 

230 def test_get_url_missing_ip(self): 

231 self._test_get_url() 

232 

233 def _test_execute(self, get_output_exception=None): 

234 self._mock_conn.open_shell.return_value = mock.sentinel.shell_id 

235 self._mock_conn.run_command.return_value = mock.sentinel.cmd_id 

236 

237 command_output = (mock.sentinel.stdout, 

238 mock.sentinel.stderr, 

239 mock.sentinel.exit_code) 

240 if get_output_exception: 

241 self._mock_conn.get_command_output.side_effect = ( 

242 get_output_exception) 

243 self.assertRaises( 

244 get_output_exception, 

245 self._winrm.execute, 

246 mock.sentinel.cmd) 

247 else: 

248 self._mock_conn.get_command_output.return_value = command_output 

249 result = self._winrm.execute(mock.sentinel.cmd) 

250 self.assertEqual(command_output, result) 

251 

252 self._mock_conn.open_shell.assert_called_once_with() 

253 self._mock_conn.run_command.assert_called_once_with( 

254 mock.sentinel.shell_id, mock.sentinel.cmd) 

255 

256 self._mock_conn.cleanup_command.assert_called_once_with( 

257 mock.sentinel.shell_id, mock.sentinel.cmd_id) 

258 self._mock_conn.close_shell.assert_called_once_with( 

259 mock.sentinel.shell_id) 

260 

261 def test_execute(self): 

262 self._test_execute() 

263 

264 def test_execute_exception(self): 

265 self._test_execute(get_output_exception=Exception)