Coverage for manila/tests/share/drivers/netapp/dataontap/client/test_rest_api.py: 99%

1# Copyright 2022 NetApp, 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""" 

16Tests for NetApp REST API layer 

17""" 

18 

19from unittest import mock 

20 

21import ddt 

22from oslo_serialization import jsonutils 

23from oslo_utils import netutils 

24import requests 

25from requests import auth 

26 

27from manila.share.drivers.netapp.dataontap.client import api as legacy_api 

28from manila.share.drivers.netapp.dataontap.client import rest_api as netapp_api 

29from manila import test 

30from manila.tests.share.drivers.netapp.dataontap.client import fakes as fake 

31 

32 

33@ddt.ddt 

34class NetAppRestApiServerTests(test.TestCase): 

35 """Test case for NetApp REST API server methods.""" 

36 def setUp(self): 

37 self.rest_client = netapp_api.RestNaServer('127.0.0.1') 

38 super(NetAppRestApiServerTests, self).setUp() 

39 

40 @ddt.data(None, 'my_cert') 

41 def test__init__ssl_verify(self, ssl_cert_path): 

42 client = netapp_api.RestNaServer('127.0.0.1', 

43 ssl_cert_path=ssl_cert_path) 

44 

45 if ssl_cert_path: 

46 self.assertEqual(ssl_cert_path, client._ssl_verify) 

47 else: 

48 self.assertTrue(client._ssl_verify) 

49 

50 @ddt.data(None, 'ftp') 

51 def test_set_transport_type_value_error(self, transport_type): 

52 self.assertRaises(ValueError, self.rest_client.set_transport_type, 

53 transport_type) 

54 

55 @ddt.data('!&', '80na', '') 

56 def test_set_port__value_error(self, port): 

57 self.assertRaises(ValueError, self.rest_client.set_port, port) 

58 

59 @ddt.data( 

60 {'port': None, 'protocol': 'http', 'expected_port': '80'}, 

61 {'port': None, 'protocol': 'https', 'expected_port': '443'}, 

62 {'port': '111', 'protocol': None, 'expected_port': '111'} 

63 ) 

64 @ddt.unpack 

65 def test_set_port(self, port, protocol, expected_port): 

66 self.rest_client._protocol = protocol 

67 

68 self.rest_client.set_port(port=port) 

69 

70 self.assertEqual(expected_port, self.rest_client._port) 

71 

72 @ddt.data('!&', '80na', '') 

73 def test_set_timeout_value_error(self, timeout): 

74 self.assertRaises(ValueError, self.rest_client.set_timeout, timeout) 

75 

76 @ddt.data({'params': {'major': 1, 'minor': '20a'}}, 

77 {'params': {'major': '20a', 'minor': 1}}, 

78 {'params': {'major': '!*', 'minor': '20a'}}) 

79 @ddt.unpack 

80 def test_set_api_version_value_error(self, params): 

81 self.assertRaises(ValueError, self.rest_client.set_api_version, 

82 **params) 

83 

84 def test_set_api_version_valid(self): 

85 args = {'major': '20', 'minor': 1} 

86 

87 self.rest_client.set_api_version(**args) 

88 

89 self.assertEqual(self.rest_client._api_major_version, 20) 

90 self.assertEqual(self.rest_client._api_minor_version, 1) 

91 self.assertEqual(self.rest_client._api_version, "20.1") 

92 

93 def test_invoke_successfully_naapi_error(self): 

94 self.mock_object(self.rest_client, '_build_headers') 

95 self.mock_object(self.rest_client, '_get_base_url', 

96 mock.Mock(return_value='')) 

97 self.mock_object( 

98 self.rest_client, 'send_http_request', 

99 mock.Mock(return_value=(10, fake.ERROR_RESPONSE_REST))) 

100 

101 self.assertRaises(legacy_api.NaApiError, 

102 self.rest_client.invoke_successfully, 

103 fake.FAKE_ACTION_URL, 'get') 

104 

105 @ddt.data(None, {'fields': 'fake_fields'}) 

106 def test_invoke_successfully(self, query): 

107 mock_build_header = self.mock_object( 

108 self.rest_client, '_build_headers', 

109 mock.Mock(return_value=fake.FAKE_HTTP_HEADER)) 

110 mock_base = self.mock_object( 

111 self.rest_client, '_get_base_url', 

112 mock.Mock(return_value=fake.FAKE_BASE_URL)) 

113 mock_add_query = self.mock_object( 

114 self.rest_client, '_add_query_params_to_url', 

115 mock.Mock(return_value=fake.FAKE_ACTION_URL)) 

116 http_code = 200 

117 mock_send_http = self.mock_object( 

118 self.rest_client, 'send_http_request', 

119 mock.Mock(return_value=(http_code, fake.NO_RECORDS_RESPONSE_REST))) 

120 

121 code, response = self.rest_client.invoke_successfully( 

122 fake.FAKE_ACTION_URL, 'get', body=fake.FAKE_HTTP_BODY, query=query, 

123 enable_tunneling=True) 

124 

125 self.assertEqual(response, fake.NO_RECORDS_RESPONSE_REST) 

126 self.assertEqual(code, http_code) 

127 mock_build_header.assert_called_once_with(True) 

128 mock_base.assert_called_once_with() 

129 self.assertEqual(bool(query), mock_add_query.called) 

130 mock_send_http.assert_called_once_with( 

131 'get', 

132 fake.FAKE_BASE_URL + fake.FAKE_ACTION_URL, fake.FAKE_HTTP_BODY, 

133 fake.FAKE_HTTP_HEADER) 

134 

135 @ddt.data( 

136 {'error': requests.HTTPError(), 'raised': legacy_api.NaApiError}, 

137 {'error': Exception, 'raised': legacy_api.NaApiError}) 

138 @ddt.unpack 

139 def test_send_http_request_http_error(self, error, raised): 

140 self.mock_object(netapp_api, 'LOG') 

141 self.mock_object(self.rest_client, '_build_session') 

142 self.rest_client._session = mock.Mock() 

143 self.mock_object( 

144 self.rest_client, '_get_request_method', mock.Mock( 

145 return_value=mock.Mock(side_effect=error))) 

146 

147 self.assertRaises(raised, self.rest_client.send_http_request, 

148 'get', fake.FAKE_ACTION_URL, fake.FAKE_HTTP_BODY, 

149 fake.FAKE_HTTP_HEADER) 

150 

151 @ddt.data( 

152 { 

153 'resp_content': fake.NO_RECORDS_RESPONSE_REST, 

154 'body': fake.FAKE_HTTP_BODY, 

155 'timeout': 10, 

156 }, 

157 { 

158 'resp_content': fake.NO_RECORDS_RESPONSE_REST, 

159 'body': fake.FAKE_HTTP_BODY, 

160 'timeout': None, 

161 }, 

162 { 

163 'resp_content': fake.NO_RECORDS_RESPONSE_REST, 

164 'body': None, 

165 'timeout': None, 

166 }, 

167 { 

168 'resp_content': None, 

169 'body': None, 

170 'timeout': None, 

171 } 

172 ) 

173 @ddt.unpack 

174 def test_send_http_request(self, resp_content, body, timeout): 

175 if timeout: 

176 self.rest_client._timeout = timeout 

177 self.mock_object(netapp_api, 'LOG') 

178 mock_json_dumps = self.mock_object( 

179 jsonutils, 'dumps', mock.Mock(return_value='fake_dump_body')) 

180 mock_build_session = self.mock_object( 

181 self.rest_client, '_build_session') 

182 _mock_session = mock.Mock() 

183 self.rest_client._session = _mock_session 

184 response = mock.Mock() 

185 response.content = resp_content 

186 response.status_code = 10 

187 mock_post = mock.Mock(return_value=response) 

188 mock_get_request_method = self.mock_object( 

189 self.rest_client, '_get_request_method', mock.Mock( 

190 return_value=mock_post)) 

191 mock_json_loads = self.mock_object( 

192 jsonutils, 'loads', 

193 mock.Mock(return_value='fake_loads_response')) 

194 

195 code, res = self.rest_client.send_http_request( 

196 'post', fake.FAKE_ACTION_URL, body, fake.FAKE_HTTP_HEADER) 

197 

198 expected_res = 'fake_loads_response' if resp_content else {} 

199 self.assertEqual(expected_res, res) 

200 self.assertEqual(10, code) 

201 self.assertEqual(bool(body), mock_json_dumps.called) 

202 self.assertEqual(bool(resp_content), mock_json_loads.called) 

203 mock_build_session.assert_called_once_with(fake.FAKE_HTTP_HEADER) 

204 mock_get_request_method.assert_called_once_with('post', _mock_session) 

205 expected_data = 'fake_dump_body' if body else {} 

206 if timeout: 

207 mock_post.assert_called_once_with( 

208 fake.FAKE_ACTION_URL, data=expected_data, timeout=timeout) 

209 else: 

210 mock_post.assert_called_once_with(fake.FAKE_ACTION_URL, 

211 data=expected_data) 

212 

213 @ddt.data( 

214 {'host': '192.168.1.0', 'port': '80', 'protocol': 'http'}, 

215 {'host': '0.0.0.0', 'port': '443', 'protocol': 'https'}, 

216 {'host': '::ffff:8', 'port': '80', 'protocol': 'http'}, 

217 {'host': 'fdf8:f53b:82e4::53', 'port': '443', 'protocol': 'https'}) 

218 @ddt.unpack 

219 def test__get_base_url(self, host, port, protocol): 

220 client = netapp_api.RestNaServer(host, port=port, 

221 transport_type=protocol) 

222 expected_host = netutils.escape_ipv6(host) 

223 expected_url = '%s://%s:%s/api' % (protocol, expected_host, port) 

224 

225 url = client._get_base_url() 

226 

227 self.assertEqual(expected_url, url) 

228 

229 def test__add_query_params_to_url(self): 

230 formatted_url = self.rest_client._add_query_params_to_url( 

231 fake.FAKE_ACTION_URL, fake.FAKE_HTTP_QUERY) 

232 

233 expected_formatted_url = fake.FAKE_ACTION_URL 

234 expected_formatted_url += fake.FAKE_FORMATTED_HTTP_QUERY 

235 self.assertEqual(expected_formatted_url, formatted_url) 

236 

237 @ddt.data('post', 'get', 'put', 'delete', 'patch') 

238 def test_get_request_method(self, method): 

239 _mock_session = mock.Mock() 

240 _mock_session.post = mock.Mock() 

241 _mock_session.get = mock.Mock() 

242 _mock_session.put = mock.Mock() 

243 _mock_session.delete = mock.Mock() 

244 _mock_session.patch = mock.Mock() 

245 

246 res = self.rest_client._get_request_method(method, _mock_session) 

247 

248 expected_method = getattr(_mock_session, method) 

249 self.assertEqual(expected_method, res) 

250 

251 def test__str__(self): 

252 fake_host = 'fake_host' 

253 client = netapp_api.RestNaServer(fake_host) 

254 

255 expected_str = "server: %s" % fake_host 

256 self.assertEqual(expected_str, str(client)) 

257 

258 def test_get_transport_type(self): 

259 expected_protocol = 'fake_protocol' 

260 self.rest_client._protocol = expected_protocol 

261 

262 res = self.rest_client.get_transport_type() 

263 

264 self.assertEqual(expected_protocol, res) 

265 

266 @ddt.data(None, ('1', '0')) 

267 def test_get_api_version(self, api_version): 

268 if api_version: 

269 self.rest_client._api_version = str(api_version) 

270 (self.rest_client._api_major_version, _) = api_version 

271 (_, self.rest_client._api_minor_version) = api_version 

272 

273 res = self.rest_client.get_api_version() 

274 

275 self.assertEqual(api_version, res) 

276 

277 @ddt.data(None, '9.10') 

278 def test_get_ontap_version(self, ontap_version): 

279 if ontap_version: 

280 self.rest_client._ontap_version = ontap_version 

281 

282 res = self.rest_client.get_ontap_version() 

283 

284 self.assertEqual(ontap_version, res) 

285 

286 def test_set_vserver(self): 

287 expected_vserver = 'fake_vserver' 

288 self.rest_client.set_vserver(expected_vserver) 

289 

290 self.assertEqual(expected_vserver, self.rest_client._vserver) 

291 

292 def test_get_vserver(self): 

293 expected_vserver = 'fake_vserver' 

294 self.rest_client._vserver = expected_vserver 

295 

296 res = self.rest_client.get_vserver() 

297 

298 self.assertEqual(expected_vserver, res) 

299 

300 def test__build_session_with_basic_auth(self): 

301 """Tests whether build session works with """ 

302 """default(basic auth) parameters""" 

303 fake_session = mock.Mock() 

304 mock_requests_session = self.mock_object( 

305 requests, 'Session', mock.Mock(return_value=fake_session)) 

306 mock_auth = self.mock_object( 

307 self.rest_client, '_create_basic_auth_handler', 

308 mock.Mock(return_value='fake_auth')) 

309 self.rest_client._ssl_verify = 'fake_ssl' 

310 

311 self.rest_client._build_session(fake.FAKE_HTTP_HEADER) 

312 

313 self.assertEqual(fake_session, self.rest_client._session) 

314 self.assertEqual('fake_auth', self.rest_client._session.auth) 

315 self.assertEqual('fake_ssl', self.rest_client._session.verify) 

316 self.assertEqual(fake.FAKE_HTTP_HEADER, 

317 self.rest_client._session.headers) 

318 mock_requests_session.assert_called_once_with() 

319 mock_auth.assert_called_once_with() 

320 

321 def test__build_session__certificate_auth(self): 

322 """Tests whether build session works with """ 

323 """valid certificate parameters""" 

324 self.rest_client._private_key_file = 'fake_key.pem' 

325 self.rest_client._certificate_file = 'fake_cert.pem' 

326 self.rest_client._certificate_host_validation = False 

327 fake_session = mock.Mock() 

328 mock_requests_session = self.mock_object( 

329 requests, 'Session', mock.Mock(return_value=fake_session)) 

330 mock_cert = self.mock_object( 

331 self.rest_client, '_create_certificate_auth_handler', 

332 mock.Mock(return_value=('fake_cert', 'fake_verify'))) 

333 

334 self.rest_client._build_session(fake.FAKE_HTTP_HEADER) 

335 

336 self.assertEqual(fake_session, self.rest_client._session) 

337 self.assertEqual(('fake_cert', 'fake_verify'), 

338 (self.rest_client._session.cert, 

339 self.rest_client._session.verify)) 

340 self.assertEqual(fake.FAKE_HTTP_HEADER, 

341 self.rest_client._session.headers) 

342 mock_requests_session.assert_called_once_with() 

343 mock_cert.assert_called_once_with() 

344 

345 @ddt.data(True, False) 

346 def test__build_headers(self, enable_tunneling): 

347 self.rest_client._vserver = fake.VSERVER_NAME 

348 

349 res = self.rest_client._build_headers(enable_tunneling) 

350 

351 expected = { 

352 "Accept": "application/json", 

353 "Content-Type": "application/json" 

354 } 

355 if enable_tunneling: 

356 expected["X-Dot-SVM-Name"] = fake.VSERVER_NAME 

357 self.assertEqual(expected, res) 

358 

359 def test__create_basic_auth_handler(self): 

360 username = 'fake_username' 

361 password = 'fake_password' 

362 client = netapp_api.RestNaServer('10.1.1.1', username=username, 

363 password=password) 

364 

365 res = client._create_basic_auth_handler() 

366 

367 expected = auth.HTTPBasicAuth(username, password) 

368 self.assertEqual(expected.__dict__, res.__dict__) 

369 

370 def test__create_certificate_auth_handler_default(self): 

371 """Test whether create certificate auth handler """ 

372 """works with default params""" 

373 self.rest_client._private_key_file = 'fake_key.pem' 

374 self.rest_client._certificate_file = 'fake_cert.pem' 

375 self.rest_client._certificate_host_validation = False 

376 cert = self.rest_client._certificate_file, \ 

377 self.rest_client._private_key_file 

378 self.rest_client._session = mock.Mock() 

379 if not self.rest_client._certificate_host_validation: 379 ↛ 381line 379 didn't jump to line 381 because the condition on line 379 was always true

380 self.assertFalse(self.rest_client._certificate_host_validation) 

381 res = self.rest_client._create_certificate_auth_handler() 

382 self.assertEqual(res, 

383 (cert, self.rest_client._certificate_host_validation)) 

384 

385 def test__create_certificate_auth_handler_with_host_validation(self): 

386 """Test whether create certificate auth handler """ 

387 """works with host validation enabled""" 

388 self.rest_client._private_key_file = 'fake_key.pem' 

389 self.rest_client._certificate_file = 'fake_cert.pem' 

390 self.rest_client._ca_certificate_file = 'fake_ca_cert.crt' 

391 self.rest_client._certificate_host_validation = True 

392 cert = self.rest_client._certificate_file, \ 

393 self.rest_client._private_key_file 

394 self.rest_client._session = mock.Mock() 

395 if self.rest_client._certificate_host_validation: 395 ↛ 397line 395 didn't jump to line 397 because the condition on line 395 was always true

396 self.assertTrue(self.rest_client._certificate_host_validation) 

397 res = self.rest_client._create_certificate_auth_handler() 

398 self.assertEqual(res, (cert, self.rest_client._ca_certificate_file))