Coverage for manila/tests/api/v1/test_share_servers.py: 100%
232 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright 2014 OpenStack Foundation
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.
16from unittest import mock
18import copy
19import ddt
20from webob import exc
22from manila.api.openstack import api_version_request as api_version
23from manila.api.v1 import share_servers
24from manila.common import constants
25from manila import context
26from manila.db import api as db_api
27from manila import exception
28from manila import policy
29from manila import test
30from manila.tests.api import fakes
32fake_share_server_list = {
33 'share_servers': [
34 {
35 'status': constants.STATUS_ACTIVE,
36 'updated_at': None,
37 'host': 'fake_host',
38 'share_network_name': 'fake_sn_name',
39 'share_network_id': 'fake_sn_id',
40 'share_network_subnet_ids': ['fake_sns_id'],
41 'project_id': 'fake_project_id',
42 'id': 'fake_server_id',
43 'is_auto_deletable': False,
44 'task_state': None,
45 'source_share_server_id': None,
46 'identifier': 'fake_id',
47 'security_service_update_support': False,
48 'network_allocation_update_support': False,
49 'encryption_key_ref': None
50 },
51 {
52 'status': constants.STATUS_ERROR,
53 'updated_at': None,
54 'host': 'fake_host_2',
55 'share_network_name': 'fake_sn_id_2',
56 'share_network_id': 'fake_sn_id_2',
57 'share_network_subnet_ids': ['fake_sns_id_2'],
58 'project_id': 'fake_project_id_2',
59 'id': 'fake_server_id_2',
60 'is_auto_deletable': True,
61 'task_state': None,
62 'source_share_server_id': None,
63 'identifier': 'fake_id_2',
64 'security_service_update_support': False,
65 'network_allocation_update_support': False,
66 'encryption_key_ref': None
68 },
69 ]
70}
72fake_share_network_get_list = {
73 'share_networks': [
74 {
75 'name': 'fake_sn_name',
76 'id': 'fake_sn_id',
77 'project_id': 'fake_project_id',
78 },
79 {
80 'name': None,
81 'id': 'fake_sn_id_2',
82 'project_id': 'fake_project_id_2',
83 }
84 ]
85}
87fake_share_server_get_result = {
88 'share_server': {
89 'status': constants.STATUS_ACTIVE,
90 'created_at': None,
91 'updated_at': None,
92 'host': 'fake_host',
93 'share_network_name': 'fake_sn_name',
94 'share_network_id': 'fake_sn_id',
95 'share_network_subnet_ids': ['fake_sns_id'],
96 'project_id': 'fake_project_id',
97 'id': 'fake_server_id',
98 'backend_details': {
99 'fake_key_1': 'fake_value_1',
100 'fake_key_2': 'fake_value_2',
101 },
102 'is_auto_deletable': False,
103 'task_state': None,
104 'source_share_server_id': None,
105 'identifier': 'fake_id',
106 'security_service_update_support': False,
107 'network_allocation_update_support': False,
108 'encryption_key_ref': None
109 }
110}
112share_server_backend_details = {
113 'fake_key_1': 'fake_value_1',
114 'fake_key_2': 'fake_value_2',
115}
117fake_share_server_backend_details_get_result = {
118 'details': share_server_backend_details
119}
122CONTEXT = context.get_admin_context()
125class FakeShareServer(object):
127 def __init__(self, *args, **kwargs):
128 super(FakeShareServer, self).__init__()
129 self.id = kwargs.get('id', 'fake_server_id')
130 if 'created_at' in kwargs:
131 self.created_at = kwargs.get('created_at', None)
132 self.updated_at = kwargs.get('updated_at', None)
133 self.host = kwargs.get('host', 'fake_host')
134 self.share_network_subnets = kwargs.get('share_network_subnets', [{
135 'id': 'fake_sns_id', 'share_network_id': 'fake_sn_id'}])
136 self.share_network_subnet_ids = kwargs.get(
137 'share_network_subnet_ids',
138 [sn['id'] for sn in self.share_network_subnets])
139 self.status = kwargs.get('status', constants.STATUS_ACTIVE)
140 self.project_id = 'fake_project_id'
141 self.identifier = kwargs.get('identifier', 'fake_id')
142 self.is_auto_deletable = kwargs.get('is_auto_deletable', False)
143 self.task_state = kwargs.get('task_state')
144 self.source_share_server_id = kwargs.get('source_share_server_id')
145 self.backend_details = share_server_backend_details
146 self.security_service_update_support = kwargs.get(
147 'security_service_update_support', False)
148 self.network_allocation_update_support = kwargs.get(
149 'network_allocation_update_support', False)
150 self.share_network_id = kwargs.get('share_network_id', 'fake_sn_id')
151 self.encryption_key_ref = kwargs.get('encryption_key_ref', None)
153 def __getitem__(self, item):
154 return getattr(self, item)
157def fake_share_server_get_all():
158 fake_share_servers = [
159 FakeShareServer(),
160 FakeShareServer(id='fake_server_id_2',
161 host='fake_host_2',
162 share_network_subnets=[{
163 'id': 'fake_sns_id_2',
164 'share_network_id': 'fake_sn_id_2',
165 }],
166 share_network_id='fake_sn_id_2',
167 identifier='fake_id_2',
168 task_state=None,
169 is_auto_deletable=True,
170 status=constants.STATUS_ERROR,
171 security_service_update_support=False,
172 network_allocation_update_support=False),
173 ]
174 return fake_share_servers
177def fake_share_server_get():
178 return FakeShareServer(created_at=None)
181class FakeRequestAdmin(object):
182 environ = {"manila.context": CONTEXT}
183 GET = {}
186class FakeRequestWithHost(FakeRequestAdmin):
187 GET = {'host': fake_share_server_list['share_servers'][0]['host']}
190class FakeRequestWithStatus(FakeRequestAdmin):
191 GET = {'status': constants.STATUS_ERROR}
194class FakeRequestWithProjectId(FakeRequestAdmin):
195 GET = {'project_id': fake_share_server_get_all()[0].project_id}
198class FakeRequestWithShareNetworkSubnetId(FakeRequestAdmin):
199 GET = {
200 'share_network_subnet_id':
201 fake_share_server_get_all()[0].share_network_subnet_ids,
202 }
205class FakeRequestWithFakeFilter(FakeRequestAdmin):
206 GET = {'fake_key': 'fake_value'}
209@ddt.ddt
210class ShareServerAPITest(test.TestCase):
212 def setUp(self):
213 super(ShareServerAPITest, self).setUp()
214 self.controller = share_servers.ShareServerController()
215 self.resource_name = self.controller.resource_name
216 self.mock_object(policy, 'check_policy',
217 mock.Mock(return_value=True))
218 self.mock_object(db_api, 'share_server_get_all',
219 mock.Mock(return_value=fake_share_server_get_all()))
220 self.mock_object(db_api, 'share_server_get_all_with_filters',
221 mock.Mock(return_value=fake_share_server_get_all()))
223 def _prepare_request(self, url, use_admin_context,
224 version=api_version._MAX_API_VERSION):
225 request = fakes.HTTPRequest.blank(url,
226 use_admin_context=use_admin_context,
227 version=version)
228 ctxt = request.environ['manila.context']
229 return request, ctxt
231 def test_index_no_filters(self):
232 request, ctxt = self._prepare_request(url='/v2/share-servers/',
233 use_admin_context=True)
234 self.mock_object(db_api, 'share_network_get', mock.Mock(
235 side_effect=[fake_share_network_get_list['share_networks'][0],
236 fake_share_network_get_list['share_networks'][1]]))
237 result = self.controller.index(request)
238 policy.check_policy.assert_called_once_with(
239 ctxt, self.resource_name, 'index')
240 db_api.share_server_get_all_with_filters.assert_called_once_with(
241 ctxt, {})
242 self.assertEqual(fake_share_server_list, result)
244 def test_index_host_filter(self):
245 request, ctxt = self._prepare_request(
246 url='/index?host=%s'
247 % fake_share_server_list['share_servers'][0]['host'],
248 use_admin_context=True)
249 self.mock_object(db_api, 'share_network_get', mock.Mock(
250 return_value=fake_share_network_get_list['share_networks'][0])),
251 self.mock_object(
252 db_api, 'share_server_get_all_with_filters',
253 mock.Mock(return_value=[
254 s for s in fake_share_server_get_all() if s['host'] ==
255 fake_share_server_list['share_servers'][0]['host']
256 ]))
257 result = self.controller.index(request)
258 policy.check_policy.assert_called_once_with(
259 ctxt, self.resource_name, 'index')
260 db_api.share_server_get_all_with_filters.assert_called_once_with(
261 ctxt, {
262 'host': fake_share_server_list['share_servers'][0]['host'],
263 })
264 self.assertEqual([fake_share_server_list['share_servers'][0]],
265 result['share_servers'])
267 def test_index_status_filter(self):
268 request, ctxt = self._prepare_request(url='/index?status=%s' %
269 constants.STATUS_ERROR,
270 use_admin_context=True)
271 self.mock_object(
272 db_api, 'share_server_get_all_with_filters',
273 mock.Mock(return_value=[
274 s for s in fake_share_server_get_all()
275 if s['status'] == constants.STATUS_ERROR
276 ]))
277 self.mock_object(db_api, 'share_network_get', mock.Mock(
278 return_value=fake_share_network_get_list['share_networks'][1]))
279 result = self.controller.index(request)
280 policy.check_policy.assert_called_once_with(
281 ctxt, self.resource_name, 'index')
282 db_api.share_server_get_all_with_filters.assert_called_once_with(
283 ctxt, {'status': constants.STATUS_ERROR})
284 self.assertEqual([fake_share_server_list['share_servers'][1]],
285 result['share_servers'])
287 def test_index_project_id_filter(self):
288 request, ctxt = self._prepare_request(
289 url='/index?project_id=%s'
290 % fake_share_server_get_all()[0].project_id,
291 use_admin_context=True)
292 self.mock_object(db_api, 'share_network_get', mock.Mock(
293 side_effect=[fake_share_network_get_list['share_networks'][0],
294 fake_share_network_get_list['share_networks'][1]]))
295 result = self.controller.index(request)
296 policy.check_policy.assert_called_once_with(
297 ctxt, self.resource_name, 'index')
298 db_api.share_server_get_all_with_filters.assert_called_once_with(
299 ctxt, {})
300 self.assertEqual([fake_share_server_list['share_servers'][0]],
301 result['share_servers'])
303 def test_index_share_network_filter_by_name(self):
304 request, ctxt = self._prepare_request(
305 url='/index?share_network=%s'
306 % fake_share_network_get_list['share_networks'][0]['name'],
307 use_admin_context=True)
308 self.mock_object(db_api, 'share_network_get', mock.Mock(
309 side_effect=[fake_share_network_get_list['share_networks'][0],
310 fake_share_network_get_list['share_networks'][1]]))
311 result = self.controller.index(request)
312 policy.check_policy.assert_called_once_with(
313 ctxt, self.resource_name, 'index')
314 db_api.share_server_get_all_with_filters.assert_called_once_with(
315 ctxt, {})
316 self.assertEqual([fake_share_server_list['share_servers'][0]],
317 result['share_servers'])
319 def test_index_share_network_filter_by_id(self):
320 request, ctxt = self._prepare_request(
321 url='/index?share_network=%s'
322 % fake_share_network_get_list['share_networks'][0]['id'],
323 use_admin_context=True)
324 self.mock_object(db_api, 'share_network_get', mock.Mock(
325 side_effect=[fake_share_network_get_list['share_networks'][0],
326 fake_share_network_get_list['share_networks'][1]]))
327 result = self.controller.index(request)
328 policy.check_policy.assert_called_once_with(
329 ctxt, self.resource_name, 'index')
330 db_api.share_server_get_all_with_filters.assert_called_once_with(
331 ctxt, {})
332 self.assertEqual([fake_share_server_list['share_servers'][0]],
333 result['share_servers'])
335 def test_index_fake_filter(self):
336 request, ctxt = self._prepare_request(url='/index?fake_key=fake_value',
337 use_admin_context=True)
338 self.mock_object(db_api, 'share_network_get', mock.Mock(
339 side_effect=[fake_share_network_get_list['share_networks'][0],
340 fake_share_network_get_list['share_networks'][1]]))
341 result = self.controller.index(request)
342 policy.check_policy.assert_called_once_with(
343 ctxt, self.resource_name, 'index')
344 db_api.share_server_get_all_with_filters.assert_called_once_with(
345 ctxt, {})
346 self.assertEqual(0, len(result['share_servers']))
348 def test_index_share_network_not_found(self):
349 request, ctxt = self._prepare_request(
350 url='/index?identifier=%s'
351 % fake_share_server_get_all()[0].identifier,
352 use_admin_context=True)
353 self.mock_object(
354 db_api, 'share_server_get_all_with_filters',
355 mock.Mock(return_value=[
356 s for s in fake_share_server_get_all()
357 if s['identifier'] == fake_share_server_get_all()[0].identifier
358 ]))
359 self.mock_object(
360 db_api, 'share_network_get',
361 mock.Mock(side_effect=exception.ShareNetworkNotFound(
362 share_network_id='fake')))
364 result = self.controller.index(request)
365 db_api.share_server_get_all_with_filters.assert_called_once_with(
366 ctxt, {
367 'identifier': fake_share_server_get_all()[0].identifier,
368 })
369 policy.check_policy.assert_called_once_with(
370 ctxt, self.resource_name, 'index')
371 exp_share_server = fake_share_server_list['share_servers'][0].copy()
372 exp_share_server['project_id'] = ''
373 exp_share_server['share_network_name'] = ''
374 self.assertEqual([exp_share_server],
375 result['share_servers'])
377 def test_index_share_network_not_found_filter_project(self):
378 request, ctxt = self._prepare_request(
379 url='/index?project_id=%s'
380 % fake_share_server_get_all()[0].project_id,
381 use_admin_context=True)
382 self.mock_object(
383 db_api, 'share_network_get',
384 mock.Mock(side_effect=exception.ShareNetworkNotFound(
385 share_network_id='fake')))
387 result = self.controller.index(request)
388 db_api.share_server_get_all_with_filters.assert_called_once_with(
389 ctxt, {})
390 policy.check_policy.assert_called_once_with(
391 ctxt, self.resource_name, 'index')
392 self.assertEqual(0, len(result['share_servers']))
394 @ddt.data({'version': '2.70', 'share_network_name': ''},
395 {'version': '2.90', 'share_network_name': 'fake_sn_name'},
396 {'version': '2.70', 'share_network_name': 'fake_sn_name'},
397 {'version': '2.68', 'share_network_name': 'fake_sn_name'})
398 @ddt.unpack
399 def test_show(self, version, share_network_name):
400 self.mock_object(db_api, 'share_server_get',
401 mock.Mock(return_value=fake_share_server_get()))
402 request, ctxt = self._prepare_request('/show', use_admin_context=True,
403 version=version)
405 share_network = copy.deepcopy(
406 fake_share_network_get_list['share_networks'][0])
407 share_server = copy.deepcopy(
408 fake_share_server_get_result['share_server'])
410 if version == '2.68':
411 share_server['share_network_subnet_id'] = \
412 share_server['share_network_subnet_ids'][0]
413 share_server.pop('share_network_subnet_ids')
414 share_server.pop('network_allocation_update_support')
416 share_network['name'] = share_network_name
417 if share_network['name']:
418 share_server['share_network_name'] = share_network['name']
419 else:
420 share_server['share_network_name'] = share_network['id']
422 if version < '2.90':
423 share_server.pop('encryption_key_ref')
425 self.mock_object(db_api, 'share_network_get', mock.Mock(
426 return_value=share_network))
427 result = self.controller.show(
428 request,
429 share_server['id'])
430 policy.check_policy.assert_called_once_with(
431 ctxt, self.resource_name, 'show')
432 db_api.share_server_get.assert_called_once_with(
433 ctxt, share_server['id'])
434 self.assertEqual(share_server,
435 result['share_server'])
437 @ddt.data(
438 {'share_server_side_effect': exception.ShareServerNotFound(
439 share_server_id="foo"),
440 'share_net_side_effect': mock.Mock()},
441 {'share_server_side_effect': mock.Mock(
442 return_value=fake_share_server_get()),
443 'share_net_side_effect': exception.ShareNetworkNotFound(
444 share_network_id="foo")})
445 @ddt.unpack
446 def test_show_server_not_found(self, share_server_side_effect,
447 share_net_side_effect):
448 self.mock_object(db_api, 'share_server_get',
449 mock.Mock(side_effect=share_server_side_effect))
450 request, ctxt = self._prepare_request('/show', use_admin_context=True)
451 self.mock_object(db_api, 'share_network_get', mock.Mock(
452 side_effect=share_net_side_effect))
453 self.assertRaises(
454 exc.HTTPNotFound, self.controller.show, request,
455 fake_share_server_get_result['share_server']['id'])
457 policy.check_policy.assert_called_once_with(
458 ctxt, self.resource_name, 'show')
459 db_api.share_server_get.assert_called_once_with(
460 ctxt, fake_share_server_get_result['share_server']['id'])
461 if isinstance(share_net_side_effect, exception.ShareNetworkNotFound):
462 exp_share_net_id = (fake_share_server_get()
463 .share_network_subnets[0]['share_network_id'])
464 db_api.share_network_get.assert_called_once_with(
465 ctxt, exp_share_net_id)
467 def test_details(self):
468 self.mock_object(db_api, 'share_server_get',
469 mock.Mock(return_value=fake_share_server_get()))
470 result = self.controller.details(
471 FakeRequestAdmin,
473 fake_share_server_get_result['share_server']['id'])
474 policy.check_policy.assert_called_once_with(
475 CONTEXT, self.resource_name, 'details')
476 db_api.share_server_get.assert_called_once_with(
477 CONTEXT, fake_share_server_get_result['share_server']['id'])
478 self.assertEqual(fake_share_server_backend_details_get_result,
479 result)
481 def test_details_share_server_not_found(self):
482 share_server_id = 'fake'
483 self.mock_object(
484 db_api, 'share_server_get',
485 mock.Mock(side_effect=exception.ShareServerNotFound(
486 share_server_id=share_server_id)))
487 self.assertRaises(exc.HTTPNotFound,
488 self.controller.details,
489 FakeRequestAdmin,
490 share_server_id)
491 policy.check_policy.assert_called_once_with(
492 CONTEXT, self.resource_name, 'details')
493 db_api.share_server_get.assert_called_once_with(
494 CONTEXT, share_server_id)
496 def test_delete_active_server(self):
497 share_server = FakeShareServer(status=constants.STATUS_ACTIVE)
498 self.mock_object(db_api, 'share_server_get',
499 mock.Mock(return_value=share_server))
500 self.mock_object(self.controller.share_api, 'delete_share_server')
501 self.controller.delete(
502 FakeRequestAdmin,
503 fake_share_server_get_result['share_server']['id'])
504 policy.check_policy.assert_called_once_with(
505 CONTEXT, self.resource_name, 'delete')
506 db_api.share_server_get.assert_called_once_with(
507 CONTEXT, fake_share_server_get_result['share_server']['id'])
508 self.controller.share_api.delete_share_server.assert_called_once_with(
509 CONTEXT, share_server)
511 def test_delete_error_server(self):
512 share_server = FakeShareServer(status=constants.STATUS_ERROR)
513 self.mock_object(db_api, 'share_server_get',
514 mock.Mock(return_value=share_server))
515 self.mock_object(self.controller.share_api, 'delete_share_server')
516 self.controller.delete(
517 FakeRequestAdmin,
518 fake_share_server_get_result['share_server']['id'])
519 policy.check_policy.assert_called_once_with(
520 CONTEXT, self.resource_name, 'delete')
521 db_api.share_server_get.assert_called_once_with(
522 CONTEXT, fake_share_server_get_result['share_server']['id'])
523 self.controller.share_api.delete_share_server.assert_called_once_with(
524 CONTEXT, share_server)
526 def test_delete_used_server(self):
527 share_server_id = fake_share_server_get_result['share_server']['id']
529 def raise_not_share_server_in_use(*args, **kwargs):
530 raise exception.ShareServerInUse(share_server_id=share_server_id)
532 share_server = fake_share_server_get()
533 self.mock_object(db_api, 'share_server_get',
534 mock.Mock(return_value=share_server))
535 self.mock_object(self.controller.share_api, 'delete_share_server',
536 mock.Mock(side_effect=raise_not_share_server_in_use))
537 self.assertRaises(exc.HTTPConflict,
538 self.controller.delete,
539 FakeRequestAdmin,
540 share_server_id)
541 db_api.share_server_get.assert_called_once_with(CONTEXT,
542 share_server_id)
543 self.controller.share_api.delete_share_server.assert_called_once_with(
544 CONTEXT, share_server)
546 def test_delete_not_found(self):
547 share_server_id = fake_share_server_get_result['share_server']['id']
549 def raise_not_found(*args, **kwargs):
550 raise exception.ShareServerNotFound(
551 share_server_id=share_server_id)
553 self.mock_object(db_api, 'share_server_get',
554 mock.Mock(side_effect=raise_not_found))
555 self.assertRaises(exc.HTTPNotFound,
556 self.controller.delete,
557 FakeRequestAdmin,
558 share_server_id)
559 db_api.share_server_get.assert_called_once_with(
560 CONTEXT, share_server_id)
561 policy.check_policy.assert_called_once_with(
562 CONTEXT, self.resource_name, 'delete')
564 def test_delete_creating_server(self):
565 share_server = FakeShareServer(status=constants.STATUS_CREATING)
566 self.mock_object(db_api, 'share_server_get',
567 mock.Mock(return_value=share_server))
568 self.assertRaises(exc.HTTPForbidden,
569 self.controller.delete,
570 FakeRequestAdmin,
571 share_server['id'])
572 policy.check_policy.assert_called_once_with(
573 CONTEXT,
574 self.resource_name, 'delete')
576 def test_delete_deleting_server(self):
577 share_server = FakeShareServer(status=constants.STATUS_DELETING)
578 self.mock_object(db_api, 'share_server_get',
579 mock.Mock(return_value=share_server))
580 self.assertRaises(exc.HTTPForbidden,
581 self.controller.delete,
582 FakeRequestAdmin,
583 share_server['id'])
584 policy.check_policy.assert_called_once_with(
585 CONTEXT, self.resource_name, 'delete')