Coverage for manila/tests/api/v2/test_share_accesses.py: 100%
126 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright (c) 2018 Huawei 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.
16from unittest import mock
18import copy
19import ddt
20from webob import exc
22from manila.api.v2 import share_accesses
23from manila.common import constants
24from manila import exception
25from manila import policy
26from manila import test
27from manila.tests.api import fakes
28from manila.tests import db_utils
29from oslo_utils import uuidutils
32@ddt.ddt
33class ShareAccessesAPITest(test.TestCase):
35 def _get_index_request(self, share_id=None, filters='', version="2.45",
36 use_admin_context=True):
37 share_id = share_id or self.share['id']
38 req = fakes.HTTPRequest.blank(
39 '/v2/share-access-rules?share_id=%s' % share_id + filters,
40 version=version, use_admin_context=use_admin_context)
41 return req
43 def _get_show_request(self, access_id=None, version="2.45",
44 use_admin_context=True):
45 access_id = access_id or self.access['id']
46 req = fakes.HTTPRequest.blank(
47 '/v2/share-access-rules/%s' % access_id,
48 version=version, use_admin_context=use_admin_context)
49 return req
51 def setUp(self):
52 super(ShareAccessesAPITest, self).setUp()
53 self.controller = (
54 share_accesses.ShareAccessesController())
55 self.resource_name = self.controller.resource_name
56 self.mock_policy_check = self.mock_object(
57 policy, 'check_policy', mock.Mock(return_value=True))
58 self.share = db_utils.create_share()
59 self.access = db_utils.create_share_access(
60 id=uuidutils.generate_uuid(),
61 share_id=self.share['id'],
62 )
63 db_utils.create_share_access(
64 id=uuidutils.generate_uuid(),
65 share_id=self.share['id'],
66 metadata={'k1': 'v1'}
67 )
69 @ddt.data({'role': 'admin', 'version': '2.45',
70 'filters': '&metadata=%7B%27k1%27%3A+%27v1%27%7D'},
71 {'role': 'user', 'version': '2.45', 'filters': ''})
72 @ddt.unpack
73 def test_list_and_show(self, role, version, filters):
74 summary_keys = ['id', 'access_level', 'access_to',
75 'access_type', 'state', 'metadata']
77 self._test_list_and_show(role, filters, version, summary_keys)
79 def _test_list_and_show(self, role, filters, version, summary_keys):
81 req = self._get_index_request(
82 filters=filters, version=version,
83 use_admin_context=(role == 'admin'))
84 index_result = self.controller.index(req)
86 self.assertIn('access_list', index_result)
87 self.assertEqual(1, len(index_result))
89 access_count = 1 if filters else 2
90 self.assertEqual(access_count, len(index_result['access_list']))
92 for index_access in index_result['access_list']:
93 self.assertIn('id', index_access)
94 req = self._get_show_request(
95 index_access['id'], version=version,
96 use_admin_context=(role == 'admin'))
97 show_result = self.controller.show(req, index_access['id'])
98 self.assertIn('access', show_result)
99 self.assertEqual(1, len(show_result))
101 show_el = show_result['access']
103 # Ensure keys common to index & show results have matching values
104 for key in summary_keys:
105 self.assertEqual(index_access[key], show_el[key])
107 @ddt.data(True, False)
108 def test_list_accesses_restricted(self, restricted):
109 req = self._get_index_request(version='2.82')
110 rule_list = [{
111 'access_to': '0.0.0.0/0',
112 'id': 'fakeid',
113 'access_key': 'fake_key'
114 }]
115 self.mock_object(
116 self.controller.share_api, 'access_get_all',
117 mock.Mock(return_value=rule_list))
118 self.mock_object(
119 self.controller, '_is_rule_restricted',
120 mock.Mock(return_value=restricted))
122 index_result = self.controller.index(req)
124 self.assertIn('access_list', index_result)
125 self.controller._is_rule_restricted.assert_called_once_with(
126 req.environ['manila.context'], rule_list[0]['id'])
127 if restricted:
128 for access in index_result['access_list']:
129 self.assertEqual('******', access['access_key'])
130 self.assertEqual('******', access['access_to'])
132 @ddt.data(True, False)
133 def test_show_restricted(self, restricted):
134 req = self._get_show_request(
135 version='2.82', use_admin_context=False)
136 self.mock_object(
137 self.controller, '_is_rule_restricted',
138 mock.Mock(return_value=restricted))
140 show_result = self.controller.show(req, self.access['id'])
142 expected_access_to = (
143 '******' if restricted else self.access['access_to'])
145 self.assertEqual(
146 expected_access_to, show_result['access']['access_to'])
148 @ddt.data(True, False)
149 def test__is_rule_restricted(self, is_rule_restricted):
150 req = self._get_show_request(
151 version='2.82', use_admin_context=False)
152 context = req.environ['manila.context']
153 fake_lock = {
154 'lock_context': 'user',
155 'user_id': 'fake',
156 'project_id': 'fake',
157 'resource_id': 'fake',
158 'resource_action': constants.RESOURCE_ACTION_DELETE,
159 'lock_reason': 'fake reason',
160 }
161 lock = fake_lock if is_rule_restricted else {}
162 locks = [lock]
164 self.mock_object(
165 self.controller.resource_locks_api, 'get_all',
166 mock.Mock(return_value=(locks, len(locks))))
167 self.mock_object(
168 self.controller.resource_locks_api, 'access_is_restricted',
169 mock.Mock(return_value=is_rule_restricted))
171 result_rule_restricted = self.controller._is_rule_restricted(
172 context, self.access['id'])
174 self.assertEqual(
175 is_rule_restricted, result_rule_restricted)
177 def test_list_accesses_share_not_found(self):
178 self.assertRaises(
179 exc.HTTPBadRequest,
180 self.controller.index,
181 self._get_index_request(share_id='inexistent_share_id'))
183 def test_list_accesses_share_req_share_id_not_exist(self):
184 req = fakes.HTTPRequest.blank('/v2/share-access-rules?',
185 version="2.45")
186 self.assertRaises(exc.HTTPBadRequest, self.controller.index, req)
188 def test_show_access_not_authorized(self):
189 share = db_utils.create_share(
190 project_id='c3c5ec1ccc4640d0af1914cbf11f05ad',
191 is_public=False)
192 access = db_utils.create_access(
193 id='76699c6b-f3da-47d7-b468-364f1347ba04',
194 share_id=share['id'])
195 req = fakes.HTTPRequest.blank(
196 '/v2/share-access-rules/%s' % access['id'],
197 version="2.45")
198 self.mock_object(
199 policy, 'check_policy',
200 mock.Mock(side_effect=[None, None, exception.NotAuthorized]))
202 self.assertRaises(exception.NotAuthorized,
203 self.controller.show,
204 req,
205 access['id'])
206 policy.check_policy.assert_has_calls([
207 mock.call(req.environ['manila.context'],
208 'share_access_rule', 'get'),
209 mock.call(req.environ['manila.context'],
210 'share', 'access_get'),
211 mock.call(req.environ['manila.context'],
212 'share', 'get', mock.ANY, do_raise=False)])
213 policy_check_call_args_list = policy.check_policy.call_args_list[2][0]
214 share_being_checked = policy_check_call_args_list[3]
215 self.assertEqual('c3c5ec1ccc4640d0af1914cbf11f05ad',
216 share_being_checked['project_id'])
217 self.assertIs(False, share_being_checked['is_public'])
219 def test_show_access_not_found(self):
220 req = self._get_show_request('inexistent_id')
221 print(req.environ)
222 self.assertRaises(
223 exc.HTTPNotFound,
224 self.controller.show,
225 req, 'inexistent_id')
227 @ddt.data('1.0', '2.0', '2.8', '2.44')
228 def test_list_with_unsupported_version(self, version):
229 self.assertRaises(
230 exception.VersionNotFoundForAPIMethod,
231 self.controller.index,
232 self._get_index_request(version=version))
234 @ddt.data('1.0', '2.0', '2.44')
235 def test_show_with_unsupported_version(self, version):
236 self.assertRaises(
237 exception.VersionNotFoundForAPIMethod,
238 self.controller.show,
239 self._get_show_request(version=version),
240 self.access['id'])
242 def _get_update_request(self, access_id=None):
243 access_id = access_id or self.access['id']
244 req = fakes.HTTPRequest.blank(
245 '/v2/share-access-rules/%s' % access_id, version="2.88",
246 experimental=True)
247 return req
249 def test_update_access_level(self):
250 update_share_access = copy.deepcopy(self.access)
251 update_share_access.update({'access_level': 'ro'})
252 self.mock_object(
253 self.controller.share_api, 'update_access',
254 mock.Mock(return_value=update_share_access))
256 body = {'update_access': {'access_level': 'ro'}}
257 url = self._get_update_request()
258 ret = self.controller.update(url, self.access['id'], body=body)
259 self.assertEqual(update_share_access['access_level'],
260 ret['access']['access_level'])
262 def test_update_access_level_invalid_access_level(self):
263 body = {'access': {'access_level': 'fake_access'}}
264 self.assertRaises(
265 exc.HTTPBadRequest,
266 self.controller.update,
267 self._get_update_request(), self.access['id'],
268 body=body)
270 def test_update_access_level_invalid_update_request(self):
271 body = {'access': {'access_key': 'xxxx'}}
272 self.assertRaises(
273 exc.HTTPBadRequest,
274 self.controller.update,
275 self._get_update_request(), self.access['id'],
276 body=body)