Coverage for manila/tests/lock/test_api.py: 100%
165 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Licensed under the Apache License, Version 2.0 (the "License"); you may
2# not use this file except in compliance with the License. You may obtain
3# a copy of the License at
4#
5# http://www.apache.org/licenses/LICENSE-2.0
6#
7# Unless required by applicable law or agreed to in writing, software
8# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
9# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
10# License for the specific language governing permissions and limitations
11# under the License.
13from unittest import mock
15import ddt
16from oslo_config import cfg
18from manila.common import constants
19from manila import context
20from manila import exception
21from manila.lock import api as lock_api
22from manila import policy
23from manila import test
24from manila.tests import utils as test_utils
25from manila import utils
27CONF = cfg.CONF
30@ddt.ddt
31class ResourceLockApiTest(test.TestCase):
33 def setUp(self):
34 super(ResourceLockApiTest, self).setUp()
35 self.lock_api = lock_api.API()
36 self.mock_object(self.lock_api, 'db')
37 self.ctxt = context.RequestContext('fakeuser',
38 'fakeproject',
39 is_admin=False)
40 self.mock_object(policy, 'check_policy')
42 @ddt.data(
43 test_utils.annotated(
44 'admin_context',
45 (context.RequestContext('fake', 'fake', is_admin=True), 'admin'),
46 ),
47 test_utils.annotated(
48 'admin_also_service_context',
49 (context.RequestContext('fake', 'fake', service_roles=['service'],
50 is_admin=True), 'service'),
51 ),
52 test_utils.annotated(
53 'service_context',
54 (context.RequestContext('fake', 'fake', service_roles=['service'],
55 is_admin=False), 'service'),
56 ),
57 test_utils.annotated(
58 'user_context',
59 (context.RequestContext('fake', 'fake', is_admin=False), 'user')
60 ),
61 )
62 @ddt.unpack
63 def test__get_lock_context(self, ctxt, expected_lock_context):
64 result = self.lock_api._get_lock_context(ctxt)
66 self.assertEqual(expected_lock_context, result['lock_context'])
67 self.assertEqual(('fake', 'fake'),
68 (result['user_id'], result['project_id']))
70 @ddt.data(
71 test_utils.annotated(
72 'user_manipulating_admin_lock',
73 (context.RequestContext('fake', 'fake', is_admin=False), 'admin'),
74 ),
75 test_utils.annotated(
76 'user_manipulating_service_lock',
77 (context.RequestContext('fake', 'fake', is_admin=False),
78 'service'),
79 ),
80 test_utils.annotated(
81 'service_manipulating_admin_lock',
82 (context.RequestContext('fake', 'fake', is_admin=False,
83 service_roles=['service']), 'admin'),
84 ),
85 )
86 @ddt.unpack
87 def test__check_allow_lock_manipulation_not_allowed(self, ctxt, lock_ctxt):
88 self.assertRaises(exception.NotAuthorized,
89 self.lock_api._check_allow_lock_manipulation,
90 ctxt, {'lock_context': lock_ctxt})
92 @ddt.data(
93 test_utils.annotated(
94 'user_manipulating_user_lock',
95 (context.RequestContext('fake', 'fake', is_admin=False), 'user'),
96 ),
97 test_utils.annotated(
98 'service_manipulating_service_lock',
99 (context.RequestContext(
100 'fake', 'fake', is_admin=False, service_roles=['service']),
101 'service'),
102 ),
103 test_utils.annotated(
104 'service_manipulating_user_lock',
105 (context.RequestContext(
106 'fake', 'fake', is_admin=False, service_roles=['service']),
107 'user'),
108 ),
109 test_utils.annotated(
110 'admin_manipulating_service_lock',
111 (context.RequestContext('fake', 'fake', is_admin=True), 'service'),
112 ),
113 test_utils.annotated(
114 'admin_manipulating_user_lock',
115 (context.RequestContext('fake', 'fake', is_admin=True), 'user'),
116 ),
117 )
118 @ddt.unpack
119 def test__check_allow_lock_manipulation_allowed(self, ctxt, lock_ctxt):
121 result = self.lock_api._check_allow_lock_manipulation(
122 ctxt,
123 {'lock_context': lock_ctxt}
124 )
125 self.assertIsNone(result)
127 @ddt.data(
128 test_utils.annotated(
129 'service_manipulating_user_lock',
130 (context.RequestContext(
131 'fake', 'fake', is_admin=False,
132 service_roles=['service']),
133 'user',
134 'user_b'),
135 ),
136 test_utils.annotated(
137 'admin_manipulating_user_lock',
138 (context.RequestContext('fake', 'fake', is_admin=True),
139 'admin',
140 'user_a'),
141 ),
142 test_utils.annotated(
143 'user_manipulating_locks_they_own',
144 (context.RequestContext('user_a', 'fake', is_admin=False),
145 'user',
146 'user_a'),
147 ),
148 test_utils.annotated(
149 'user_manipulating_other_users_lock',
150 (context.RequestContext('user_a', 'fake', is_admin=False),
151 'user',
152 'user_b'),
153 ),
154 )
155 @ddt.unpack
156 def test_access_is_restricted(self, ctxt, lock_ctxt, lock_user):
157 resource_lock = {
158 'user_id': lock_user,
159 'lock_context': lock_ctxt
160 }
161 is_restricted = (
162 (not ctxt.is_admin and not ctxt.is_service)
163 and lock_user != ctxt.user_id)
164 expected_mock_policy = {}
165 if is_restricted:
166 expected_mock_policy['side_effect'] = exception.NotAuthorized
167 self.mock_object(self.lock_api, '_check_allow_lock_manipulation')
168 self.mock_object(policy, 'check_policy',
169 mock.Mock(**expected_mock_policy))
171 result = self.lock_api.access_is_restricted(
172 ctxt,
173 resource_lock
174 )
175 self.assertEqual(is_restricted, result)
177 def test_access_is_restricted_not_authorized(self):
178 resource_lock = {
179 'user_id': 'fakeuserid',
180 'lock_context': 'user'
181 }
182 ctxt = context.RequestContext('fake', 'fake')
183 self.mock_object(self.lock_api, '_check_allow_lock_manipulation',
184 mock.Mock(side_effect=exception.NotAuthorized()))
186 result = self.lock_api.access_is_restricted(
187 ctxt,
188 resource_lock
189 )
190 self.assertTrue(result)
192 def test_get_all_all_projects_ignored(self):
193 self.mock_object(policy, 'check_policy', mock.Mock(return_value=False))
194 self.mock_object(self.lock_api.db, 'resource_lock_get_all',
195 mock.Mock(return_value=('list of locks', None)))
197 locks, count = self.lock_api.get_all(
198 self.ctxt,
199 search_opts={
200 'all_projects': True,
201 'project_id': '5dca5323e33b49fca4a5b261c72e612c',
202 })
203 self.lock_api.db.resource_lock_get_all.assert_called_once_with(
204 utils.IsAMatcher(context.RequestContext),
205 filters={},
206 limit=None,
207 offset=None,
208 sort_key='created_at',
209 sort_dir='desc',
210 show_count=False,
211 )
212 self.assertEqual(('list of locks', None), (locks, count))
214 def test_get_all_with_filters(self):
215 self.mock_object(self.lock_api.db, 'resource_lock_get_all',
216 mock.Mock(return_value=('list of locks', 4)))
217 search_opts = {
218 'all_projects': True,
219 'project_id': '5dca5323e33b49fca4a5b261c72e612c',
220 'resource_type': 'snapshot',
221 }
222 locks = self.lock_api.get_all(
223 self.ctxt,
224 limit=3,
225 offset=3,
226 search_opts=search_opts,
227 show_count=True
228 )
229 self.lock_api.db.resource_lock_get_all.assert_called_once_with(
230 utils.IsAMatcher(context.RequestContext),
231 filters=search_opts,
232 limit=3,
233 offset=3,
234 sort_key='created_at',
235 sort_dir='desc',
236 show_count=True,
237 )
238 self.assertEqual('list of locks', locks[0])
239 self.assertEqual(4, locks[1])
241 def test_create_lock_resource_not_owned_by_user(self):
242 self.mock_object(
243 policy,
244 'check_policy',
245 mock.Mock(side_effect=exception.PolicyNotAuthorized(
246 action="resource_lock:create")),
247 )
249 self.assertRaises(exception.PolicyNotAuthorized,
250 self.lock_api.create,
251 self.ctxt,
252 resource_id='19529cea-0471-4972-adaa-fee8694b7538',
253 resource_type='share',
254 resource_action='delete')
255 self.lock_api.db.share_get.assert_called_once_with(
256 utils.IsAMatcher(context.RequestContext),
257 '19529cea-0471-4972-adaa-fee8694b7538',
258 )
259 self.lock_api.db.resource_lock_create.assert_not_called()
261 @ddt.data(constants.STATUS_DELETING,
262 constants.STATUS_ERROR_DELETING,
263 constants.STATUS_UNMANAGING,
264 constants.STATUS_MANAGE_ERROR_UNMANAGING,
265 constants.STATUS_UNMANAGE_ERROR,
266 constants.STATUS_UNMANAGED,
267 constants.STATUS_DELETED)
268 def test_create_lock_invalid_resource_status(self, status):
269 self.mock_object(self.lock_api.db, 'resource_lock_create',
270 mock.Mock(return_value='created_obj'))
271 self.mock_object(self.lock_api.db, 'share_get',
272 mock.Mock(return_value={'status': status}))
274 self.assertRaises(exception.InvalidInput,
275 self.lock_api.create,
276 self.ctxt,
277 resource_id='7dab6090-1dfd-4829-bbaf-602fcd1c8248',
278 resource_action='delete',
279 resource_type='share')
281 self.lock_api.db.resource_lock_create.assert_not_called()
283 def test_create_lock_invalid_resource_soft_deleted(self):
284 self.mock_object(self.lock_api.db, 'resource_lock_create',
285 mock.Mock(return_value='created_obj'))
286 self.mock_object(self.lock_api.db, 'share_get',
287 mock.Mock(return_value={'is_soft_deleted': True}))
289 self.assertRaises(exception.InvalidInput,
290 self.lock_api.create,
291 self.ctxt,
292 resource_id='0bbf0b62-cb29-4218-920b-3f62faa99ff8',
293 resource_action='delete',
294 resource_type='share')
296 self.lock_api.db.resource_lock_create.assert_not_called()
298 def test_create_lock(self):
299 self.mock_object(self.lock_api.db, 'resource_lock_create',
300 mock.Mock(return_value='created_obj'))
301 mock_share = {
302 'id': 'cacac01c-853d-47f3-afcb-da4484bd09a5',
303 'status': constants.STATUS_AVAILABLE,
304 'is_soft_deleted': False,
305 }
306 self.mock_object(self.lock_api.db, 'share_get',
307 mock.Mock(return_value=mock_share))
309 result = self.lock_api.create(
310 self.ctxt,
311 resource_id='cacac01c-853d-47f3-afcb-da4484bd09a5',
312 resource_action='delete',
313 resource_type='share',
314 )
316 self.assertEqual('created_obj', result)
317 db_create_arg = self.lock_api.db.resource_lock_create.call_args[0][1]
318 expected_create_arg = {
319 'resource_id': 'cacac01c-853d-47f3-afcb-da4484bd09a5',
320 'resource_action': 'delete',
321 'user_id': 'fakeuser',
322 'project_id': 'fakeproject',
323 'lock_context': 'user',
324 'lock_reason': None,
325 'resource_type': constants.SHARE_RESOURCE_TYPE
327 }
328 self.assertEqual(expected_create_arg, db_create_arg)
330 def test_create_access_show_lock(self):
331 self.mock_object(self.lock_api.db, 'resource_lock_create',
332 mock.Mock(return_value='created_obj'))
333 mock_access = {
334 'id': 'cacac01c-853d-47f3-afcb-da4484bd09a5',
335 'state': constants.STATUS_ACTIVE,
336 }
337 self.mock_object(self.lock_api.db, 'access_get',
338 mock.Mock(return_value=mock_access))
339 self.mock_object(self.lock_api.db, 'resource_lock_get_all',
340 mock.Mock(return_value=['', 0]))
341 self.mock_object(self.ctxt, 'elevated',
342 mock.Mock(return_value=self.ctxt))
344 result = self.lock_api.create(
345 self.ctxt,
346 resource_id='cacac01c-853d-47f3-afcb-da4484bd09a5',
347 resource_action=constants.RESOURCE_ACTION_SHOW,
348 resource_type=constants.SHARE_ACCESS_RESOURCE_TYPE,
349 )
351 self.assertEqual('created_obj', result)
352 db_create_arg = self.lock_api.db.resource_lock_create.call_args[0][1]
353 resource_id = 'cacac01c-853d-47f3-afcb-da4484bd09a5'
354 expected_create_arg = {
355 'resource_id': resource_id,
356 'resource_action': constants.RESOURCE_ACTION_SHOW,
357 'user_id': 'fakeuser',
358 'project_id': 'fakeproject',
359 'lock_context': 'user',
360 'lock_reason': None,
361 'resource_type': constants.SHARE_ACCESS_RESOURCE_TYPE
363 }
364 self.assertEqual(expected_create_arg, db_create_arg)
365 filters = {
366 'resource_id': resource_id,
367 'resource_action': constants.RESOURCE_ACTION_SHOW,
368 'all_projects': True
369 }
370 self.lock_api.db.resource_lock_get_all.assert_called_once_with(
371 self.ctxt, filters=filters)
373 def test_create_visibility_lock_lock_exists(self):
374 self.mock_object(self.lock_api.db, 'resource_lock_create',
375 mock.Mock(return_value='created_obj'))
376 self.mock_object(self.lock_api.db, 'resource_lock_get_all',
377 mock.Mock(return_value=['visibility_lock', 1]))
378 self.mock_object(self.ctxt, 'elevated',
379 mock.Mock(return_value=self.ctxt))
381 self.assertRaises(
382 exception.ResourceVisibilityLockExists,
383 self.lock_api.create,
384 self.ctxt,
385 resource_id='cacac01c-853d-47f3-afcb-da4484bd09a5',
386 resource_action=constants.RESOURCE_ACTION_SHOW,
387 resource_type=constants.SHARE_ACCESS_RESOURCE_TYPE,
388 )
390 resource_id = 'cacac01c-853d-47f3-afcb-da4484bd09a5'
391 filters = {
392 'resource_id': resource_id,
393 'resource_action': constants.RESOURCE_ACTION_SHOW,
394 'all_projects': True
395 }
396 self.lock_api.db.resource_lock_get_all.assert_called_once_with(
397 self.ctxt, filters=filters)
399 @ddt.data(True, False)
400 def test_update_lock_resource_not_allowed_with_policy_failure(
401 self, policy_fails):
402 lock = {'id': 'd767d3cd-1187-404a-a91f-8b172e0e768e'}
403 if policy_fails:
404 self.mock_object(
405 policy,
406 'check_policy',
407 mock.Mock(
408 side_effect=exception.PolicyNotAuthorized(
409 action='resource_lock:update'),
410 ),
411 )
412 self.mock_object(
413 self.lock_api,
414 '_check_allow_lock_manipulation',
415 mock.Mock(
416 side_effect=exception.NotAuthorized
417 ),
418 )
420 self.assertRaises(exception.NotAuthorized,
421 self.lock_api.update,
422 self.ctxt,
423 lock,
424 {'foo': 'bar'})
426 @ddt.data(constants.STATUS_DELETING,
427 constants.STATUS_ERROR_DELETING,
428 constants.STATUS_UNMANAGING,
429 constants.STATUS_MANAGE_ERROR_UNMANAGING,
430 constants.STATUS_UNMANAGE_ERROR,
431 constants.STATUS_UNMANAGED,
432 constants.STATUS_DELETED)
433 def test_update_invalid_resource_status(self, status):
434 lock = {
435 'id': 'd767d3cd-1187-404a-a91f-8b172e0e768e',
436 'resource_id': '266cf54f-f9cf-4d6c-94f3-7b67f00e0465',
437 'resource_action': 'something',
438 'resource_type': 'share',
439 }
440 self.mock_object(self.lock_api, '_check_allow_lock_manipulation')
441 self.mock_object(self.lock_api.db,
442 'share_get',
443 mock.Mock(return_value={'status': status}))
445 self.assertRaises(exception.InvalidInput,
446 self.lock_api.update,
447 self.ctxt,
448 lock,
449 {'resource_action': 'delete'})
451 self.lock_api.db.resource_lock_update.assert_not_called()
453 def test_update(self):
454 self.mock_object(self.lock_api, '_check_allow_lock_manipulation')
455 self.mock_object(self.lock_api.db, 'resource_lock_update',
456 mock.Mock(return_value='updated_obj'))
457 lock = {'id': 'd767d3cd-1187-404a-a91f-8b172e0e768e'}
459 result = self.lock_api.update(
460 self.ctxt,
461 lock,
462 {'foo': 'bar'},
463 )
465 self.assertEqual('updated_obj', result)
466 self.lock_api.db.resource_lock_update.assert_called_once_with(
467 utils.IsAMatcher(context.RequestContext),
468 'd767d3cd-1187-404a-a91f-8b172e0e768e',
469 {'foo': 'bar'},
470 )
472 @ddt.data(True, False)
473 def test_delete_not_allowed_with_policy_failure(self, policy_fails):
474 self.mock_object(self.lock_api.db, 'resource_lock_get', mock.Mock(
475 return_value={'id': 'd767d3cd-1187-404a-a91f-8b172e0e768e'}))
476 if policy_fails:
477 self.mock_object(
478 policy,
479 'check_policy',
480 mock.Mock(
481 side_effect=exception.PolicyNotAuthorized(
482 action='resource_lock:delete'),
483 ),
484 )
485 self.mock_object(
486 self.lock_api,
487 '_check_allow_lock_manipulation',
488 mock.Mock(
489 side_effect=exception.NotAuthorized
490 ),
491 )
493 self.assertRaises(exception.NotAuthorized,
494 self.lock_api.delete,
495 self.ctxt,
496 'd767d3cd-1187-404a-a91f-8b172e0e768e')
498 policy.check_policy.assert_called_once_with(
499 utils.IsAMatcher(context.RequestContext),
500 'resource_lock',
501 'delete',
502 {'id': 'd767d3cd-1187-404a-a91f-8b172e0e768e'},
503 )
504 self.assertEqual(not policy_fails,
505 self.lock_api._check_allow_lock_manipulation.called)
506 self.lock_api.db.resource_lock_delete.assert_not_called()
508 def test_delete(self):
509 self.mock_object(self.lock_api.db, 'resource_lock_get', mock.Mock(
510 return_value={'id': 'd767d3cd-1187-404a-a91f-8b172e0e768e'}))
511 self.mock_object(self.lock_api, '_check_allow_lock_manipulation')
513 result = self.lock_api.delete(self.ctxt,
514 'd767d3cd-1187-404a-a91f-8b172e0e768e')
515 self.assertIsNone(result)
516 self.lock_api.db.resource_lock_delete.assert_called_once_with(
517 utils.IsAMatcher(context.RequestContext),
518 'd767d3cd-1187-404a-a91f-8b172e0e768e'
519 )
521 def test_ensure_context_can_delete_lock_policy_fails(self):
522 lock = {'id': 'd767d3cd-1187-404a-a91f-8b172e0e768e'}
523 self.mock_object(
524 self.lock_api.db, 'resource_lock_get', mock.Mock(return_value=lock)
525 )
526 self.mock_object(
527 policy,
528 'check_policy',
529 mock.Mock(side_effect=exception.PolicyNotAuthorized(
530 action="resource_lock:delete")),
531 )
533 self.assertRaises(
534 exception.NotAuthorized,
535 self.lock_api.ensure_context_can_delete_lock,
536 self.ctxt,
537 'd767d3cd-1187-404a-a91f-8b172e0e768e')
539 self.lock_api.db.resource_lock_get.assert_called_once_with(
540 self.ctxt, 'd767d3cd-1187-404a-a91f-8b172e0e768e'
541 )
542 policy.check_policy.assert_called_once_with(
543 self.ctxt, 'resource_lock', 'delete', lock)
545 def test_ensure_context_can_delete_lock(self):
546 lock = {'id': 'd767d3cd-1187-404a-a91f-8b172e0e768e'}
547 self.mock_object(
548 self.lock_api.db, 'resource_lock_get', mock.Mock(return_value=lock)
549 )
550 self.mock_object(policy, 'check_policy')
551 self.mock_object(self.lock_api, '_check_allow_lock_manipulation')
553 self.lock_api.ensure_context_can_delete_lock(
554 self.ctxt,
555 'd767d3cd-1187-404a-a91f-8b172e0e768e')
557 self.lock_api.db.resource_lock_get.assert_called_once_with(
558 self.ctxt, 'd767d3cd-1187-404a-a91f-8b172e0e768e'
559 )
560 policy.check_policy.assert_called_once_with(
561 self.ctxt, 'resource_lock', 'delete', lock)