Coverage for manila/tests/api/v2/test_share_export_locations.py: 99%

1# Copyright (c) 2015 Mirantis 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 

16from unittest import mock 

17 

18import ddt 

19from webob import exc 

20 

21from manila.api.openstack import api_version_request as api_version 

22from manila.api.v2 import share_export_locations as export_locations 

23from manila.common import constants 

24from manila import context 

25from manila import db 

26from manila import exception 

27from manila import policy 

28from manila import test 

29from manila.tests.api import fakes 

30from manila.tests import db_utils 

31 

32 

33@ddt.ddt 

34class ShareExportLocationsAPITest(test.TestCase): 

35 

36 def _get_request(self, version="2.9", use_admin_context=True): 

37 req = fakes.HTTPRequest.blank( 

38 '/v2/shares/%s/export_locations' % self.share_instance_id, 

39 version=version, use_admin_context=use_admin_context) 

40 return req 

41 

42 def setUp(self): 

43 super(ShareExportLocationsAPITest, self).setUp() 

44 self.controller = ( 

45 export_locations.ShareExportLocationController()) 

46 self.resource_name = self.controller.resource_name 

47 self.ctxt = { 

48 'admin': context.RequestContext('admin', 'fake', True), 

49 'user': context.RequestContext('fake', 'fake'), 

50 } 

51 self.mock_policy_check = self.mock_object( 

52 policy, 'check_policy', mock.Mock(return_value=True)) 

53 self.share = db_utils.create_share() 

54 self.share_instance_id = self.share.instance.id 

55 self.req = self._get_request() 

56 paths = ['fake1/1/', 'fake2/2', 'fake3/3'] 

57 db.export_locations_update( 

58 self.ctxt['admin'], self.share_instance_id, paths, False) 

59 

60 @ddt.data({'role': 'admin', 'version': '2.9'}, 

61 {'role': 'user', 'version': '2.9'}, 

62 {'role': 'admin', 'version': '2.13'}, 

63 {'role': 'user', 'version': '2.13'}) 

64 @ddt.unpack 

65 def test_list_and_show(self, role, version): 

66 

67 summary_keys = ['id', 'path'] 

68 admin_summary_keys = summary_keys + [ 

69 'share_instance_id', 'is_admin_only'] 

70 detail_keys = summary_keys + ['created_at', 'updated_at'] 

71 admin_detail_keys = admin_summary_keys + ['created_at', 'updated_at'] 

72 

73 self._test_list_and_show(role, version, summary_keys, detail_keys, 

74 admin_summary_keys, admin_detail_keys) 

75 

76 @ddt.data('admin', 'user') 

77 def test_list_and_show_with_preferred_flag(self, role): 

78 

79 summary_keys = ['id', 'path', 'preferred'] 

80 admin_summary_keys = summary_keys + [ 

81 'share_instance_id', 'is_admin_only'] 

82 detail_keys = summary_keys + ['created_at', 'updated_at'] 

83 admin_detail_keys = admin_summary_keys + ['created_at', 'updated_at'] 

84 

85 self._test_list_and_show(role, '2.14', summary_keys, detail_keys, 

86 admin_summary_keys, admin_detail_keys) 

87 

88 def _test_list_and_show(self, role, version, summary_keys, detail_keys, 

89 admin_summary_keys, admin_detail_keys): 

90 

91 req = self._get_request(version=version, 

92 use_admin_context=(role == 'admin')) 

93 index_result = self.controller.index(req, self.share['id']) 

94 

95 self.assertIn('export_locations', index_result) 

96 self.assertEqual(1, len(index_result)) 

97 self.assertEqual(3, len(index_result['export_locations'])) 

98 

99 for index_el in index_result['export_locations']: 

100 self.assertIn('id', index_el) 

101 show_result = self.controller.show( 

102 req, self.share['id'], index_el['id']) 

103 self.assertIn('export_location', show_result) 

104 self.assertEqual(1, len(show_result)) 

105 

106 show_el = show_result['export_location'] 

107 

108 # Check summary keys in index result & detail keys in show result 

109 if role == 'admin': 

110 self.assertEqual(len(admin_summary_keys), len(index_el)) 

111 for key in admin_summary_keys: 

112 self.assertIn(key, index_el) 

113 self.assertEqual(len(admin_detail_keys), len(show_el)) 

114 for key in admin_detail_keys: 

115 self.assertIn(key, show_el) 

116 else: 

117 self.assertEqual(len(summary_keys), len(index_el)) 

118 for key in summary_keys: 

119 self.assertIn(key, index_el) 

120 self.assertEqual(len(detail_keys), len(show_el)) 

121 for key in detail_keys: 

122 self.assertIn(key, show_el) 

123 

124 # Ensure keys common to index & show results have matching values 

125 for key in summary_keys: 

126 self.assertEqual(index_el[key], show_el[key]) 

127 

128 def test_list_export_locations_share_not_found(self): 

129 self.assertRaises( 

130 exc.HTTPNotFound, 

131 self.controller.index, 

132 self.req, 'inexistent_share_id', 

133 ) 

134 

135 def test_show_export_location_share_not_found(self): 

136 index_result = self.controller.index(self.req, self.share['id']) 

137 el_id = index_result['export_locations'][0]['id'] 

138 self.assertRaises( 

139 exc.HTTPNotFound, 

140 self.controller.show, 

141 self.req, 'inexistent_share_id', el_id, 

142 ) 

143 

144 def test_show_export_location_not_found(self): 

145 self.assertRaises( 

146 exc.HTTPNotFound, 

147 self.controller.show, 

148 self.req, self.share['id'], 'inexistent_export_location', 

149 ) 

150 

151 def test_get_admin_export_location(self): 

152 el_data = { 

153 'path': '/admin/export/location', 

154 'is_admin_only': True, 

155 'metadata': {'foo': 'bar'}, 

156 } 

157 db.export_locations_update( 

158 self.ctxt['admin'], self.share_instance_id, el_data, True) 

159 index_result = self.controller.index(self.req, self.share['id']) 

160 el_id = index_result['export_locations'][0]['id'] 

161 

162 # Not found for member 

163 member_req = self._get_request(use_admin_context=False) 

164 self.assertRaises( 

165 exc.HTTPForbidden, 

166 self.controller.show, 

167 member_req, self.share['id'], el_id, 

168 ) 

169 

170 # Ok for admin 

171 el = self.controller.show(self.req, self.share['id'], el_id) 

172 for k, v in el.items(): 

173 self.assertEqual(v, el[k]) 

174 

175 @ddt.data(*set(('2.46', '2.47', api_version._MAX_API_VERSION))) 

176 def test_list_export_locations_replicated_share(self, version): 

177 """Test the export locations API changes between 2.46 and 2.47 

178 

179 For API version <= 2.46, non-active replica export locations are 

180 included in the API response. They are not included in and beyond 

181 version 2.47. 

182 """ 

183 # Setup data 

184 share = db_utils.create_share( 

185 replication_type=constants.REPLICATION_TYPE_READABLE, 

186 replica_state=constants.REPLICA_STATE_ACTIVE) 

187 active_replica_id = share.instance.id 

188 exports = [ 

189 {'path': 'myshare.mydomain/active-replica-exp1', 

190 'is_admin_only': False}, 

191 {'path': 'myshare.mydomain/active-replica-exp2', 

192 'is_admin_only': False}, 

193 ] 

194 db.export_locations_update( 

195 self.ctxt['user'], active_replica_id, exports) 

196 

197 # Replicas 

198 share_replica2 = db_utils.create_share_replica( 

199 share_id=share.id, replica_state=constants.REPLICA_STATE_IN_SYNC) 

200 share_replica3 = db_utils.create_share_replica( 

201 share_id=share.id, 

202 replica_state=constants.REPLICA_STATE_OUT_OF_SYNC) 

203 replica2_exports = [ 

204 {'path': 'myshare.mydomain/insync-replica-exp', 

205 'is_admin_only': False} 

206 ] 

207 replica3_exports = [ 

208 {'path': 'myshare.mydomain/outofsync-replica-exp', 

209 'is_admin_only': False} 

210 ] 

211 db.export_locations_update( 

212 self.ctxt['user'], share_replica2.id, replica2_exports) 

213 db.export_locations_update( 

214 self.ctxt['user'], share_replica3.id, replica3_exports) 

215 

216 req = self._get_request(version=version) 

217 index_result = self.controller.index(req, share['id']) 

218 

219 actual_paths = [el['path'] for el in index_result['export_locations']] 

220 if self.is_microversion_ge(version, '2.47'): 

221 self.assertEqual(2, len(index_result['export_locations'])) 

222 self.assertNotIn( 

223 'myshare.mydomain/insync-replica-exp', actual_paths) 

224 self.assertNotIn( 

225 'myshare.mydomain/outofsync-replica-exp', actual_paths) 

226 else: 

227 self.assertEqual(4, len(index_result['export_locations'])) 

228 self.assertIn('myshare.mydomain/insync-replica-exp', actual_paths) 

229 self.assertIn( 

230 'myshare.mydomain/outofsync-replica-exp', actual_paths) 

231 

232 @ddt.data('1.0', '2.0', '2.8') 

233 def test_list_with_unsupported_version(self, version): 

234 self.assertRaises( 

235 exception.VersionNotFoundForAPIMethod, 

236 self.controller.index, 

237 self._get_request(version), 

238 self.share_instance_id, 

239 ) 

240 

241 @ddt.data('1.0', '2.0', '2.8') 

242 def test_show_with_unsupported_version(self, version): 

243 index_result = self.controller.index(self.req, self.share['id']) 

244 

245 self.assertRaises( 

246 exception.VersionNotFoundForAPIMethod, 

247 self.controller.show, 

248 self._get_request(version), 

249 self.share['id'], 

250 index_result['export_locations'][0]['id'] 

251 ) 

252 

253 @ddt.data('2.86', '2.87') 

254 def test_index_metadata(self, version): 

255 req = self._get_request(version, use_admin_context=True) 

256 index_result = self.controller.index(req, self.share['id']) 

257 if version == '2.86': 

258 assert 'metadata' not in index_result['export_locations'][0] 

259 elif version == '2.87': 259 ↛ exitline 259 didn't return from function 'test_index_metadata' because the condition on line 259 was always true

260 assert 'metadata' in index_result['export_locations'][0] 

261 

262 @ddt.data('2.86', '2.87') 

263 def test_show_metadata(self, version): 

264 req = self._get_request(version, use_admin_context=True) 

265 index_result = self.controller.index(req, self.share['id']) 

266 el_id = index_result['export_locations'][0]['id'] 

267 show_result = self.controller.show(req, self.share['id'], el_id) 

268 

269 if version == '2.86': 

270 assert 'metadata' not in show_result['export_location'] 

271 elif version == '2.87': 271 ↛ exitline 271 didn't return from function 'test_show_metadata' because the condition on line 271 was always true

272 assert 'metadata' in show_result['export_location'] 

273 

274 def test_validate_metadata_for_update(self): 

275 index_result = self.controller.index(self.req, self.share['id']) 

276 el_id = index_result['export_locations'][0]['id'] 

277 metadata = {"foo": "bar", "preferred": "False"} 

278 

279 req = fakes.HTTPRequest.blank( 

280 '/v2/shares/%s/export_locations/%s/metadata' % ( 

281 self.share_instance_id, el_id), 

282 version="2.87", use_admin_context=True) 

283 result = self.controller._validate_metadata_for_update( 

284 req, el_id, metadata) 

285 

286 self.assertEqual(metadata, result) 

287 

288 def test_validate_metadata_for_update_invalid(self): 

289 index_result = self.controller.index(self.req, self.share['id']) 

290 el_id = index_result['export_locations'][0]['id'] 

291 metadata = {"foo": "bar", "preferred": "False"} 

292 

293 self.mock_policy_check = self.mock_object( 

294 policy, 'check_policy', mock.Mock( 

295 side_effect=exception.PolicyNotAuthorized( 

296 action="update_admin_only_metadata"))) 

297 

298 req = fakes.HTTPRequest.blank( 

299 '/v2/shares/%s/export_locations/%s/metadata' % ( 

300 self.share_instance_id, el_id), 

301 version="2.87", use_admin_context=False) 

302 

303 self.assertRaises(exc.HTTPForbidden, 

304 self.controller._validate_metadata_for_update, 

305 req, el_id, metadata) 

306 self.mock_policy_check.assert_called_once_with( 

307 req.environ['manila.context'], 'share_export_location', 

308 'update_admin_only_metadata') 

309 

310 def test_create_metadata(self): 

311 index_result = self.controller.index(self.req, self.share['id']) 

312 el_id = index_result['export_locations'][0]['id'] 

313 body = {'metadata': {'key1': 'val1', 'key2': 'val2'}} 

314 mock_validate = self.mock_object( 

315 self.controller, '_validate_metadata_for_update', 

316 mock.Mock(return_value=body['metadata'])) 

317 mock_create = self.mock_object( 

318 self.controller, '_create_metadata', 

319 mock.Mock(return_value=body)) 

320 

321 req = fakes.HTTPRequest.blank( 

322 '/v2/shares/%s/export_locations/%s/metadata' % ( 

323 self.share_instance_id, el_id), 

324 version="2.87", use_admin_context=True) 

325 

326 res = self.controller.create_metadata(req, self.share['id'], el_id, 

327 body=body) 

328 self.assertEqual(body, res) 

329 mock_validate.assert_called_once_with(req, el_id, body['metadata'], 

330 delete=False) 

331 mock_create.assert_called_once_with(req, el_id, body) 

332 

333 def test_update_all_metadata(self): 

334 index_result = self.controller.index(self.req, self.share['id']) 

335 el_id = index_result['export_locations'][0]['id'] 

336 body = {'metadata': {'key1': 'val1', 'key2': 'val2'}} 

337 mock_validate = self.mock_object( 

338 self.controller, '_validate_metadata_for_update', 

339 mock.Mock(return_value=body['metadata'])) 

340 mock_update = self.mock_object( 

341 self.controller, '_update_all_metadata', 

342 mock.Mock(return_value=body)) 

343 

344 req = fakes.HTTPRequest.blank( 

345 '/v2/shares/%s/export_locations/%s/metadata' % ( 

346 self.share_instance_id, el_id), 

347 version="2.87", use_admin_context=True) 

348 

349 res = self.controller.update_all_metadata(req, self.share['id'], el_id, 

350 body=body) 

351 self.assertEqual(body, res) 

352 mock_validate.assert_called_once_with(req, el_id, body['metadata']) 

353 mock_update.assert_called_once_with(req, el_id, body) 

354 

355 def test_delete_metadata(self): 

356 index_result = self.controller.index(self.req, self.share['id']) 

357 el_id = index_result['export_locations'][0]['id'] 

358 mock_delete = self.mock_object( 

359 self.controller, '_delete_metadata', mock.Mock()) 

360 

361 req = fakes.HTTPRequest.blank( 

362 '/v2/shares/%s/export_locations/%s/metadata/fake_key' % ( 

363 self.share_instance_id, el_id), 

364 version="2.87", use_admin_context=True) 

365 self.controller.delete_metadata(req, self.share['id'], el_id, 

366 'fake_key') 

367 mock_delete.assert_called_once_with(req, el_id, 'fake_key')