openstack/manila
Code Issues Proposed changes
a13ff5d5a7
manila/manila/tests/api/v1/test_share_manage.py
haixin 1b5771ef15 Fix logic that determines a share exists before manage 
The share "manage" API checks whether
an existing/known share is being imported by
matching the export path provided to existing
shares.

This lookup does not consider the
fact that shares may have multiple export
locations, because it relies on an old/deprecated
"export_location" property on shares which
was added to provide backwards compatibility
to the API that presented only one export
location per share.

Further, it's possible to get a
"ERROR: Invalid share: Share already exists"
exception even when no such share exists in the
database.

Fix the lookup by using the "export_location_path"
based lookup which is faster, since it performs
a meaningful join on the export locations table;
and remove the parameters "protocol"
and "share_type_id" - these things make no
difference when there's a duplicated export
location. We'll consider "host" as a lookup
parameter since we can't be sure that export
locations are unique in a deployment - but they
ought to be unique for a given host.

Closes-Bug: #1848608
Closes-Bug: #1893718
Change-Id: I1d1aef0c2b48764789b43b91b258def6464b389f
Co-Authored-By: Goutham Pacha Ravi <gouthampravi@gmail.com>
Signed-off-by: Goutham Pacha Ravi <gouthampravi@gmail.com>
2020-10-19 15:25:07 +08:00

264 lines
11 KiB
Python
Raw Blame History

# Copyright 2015 Mirantis inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from unittest import mock
import ddt
import webob
from manila.api import common
from manila.api.v1 import share_manage
from manila.db import api as db_api
from manila import exception
from manila import policy
from manila.share import api as share_api
from manila.share import share_types
from manila import test
from manila.tests.api import fakes
from manila import utils
def get_fake_manage_body(export_path='/fake', service_host='fake@host#POOL',
protocol='fake', share_type='fake', **kwargs):
fake_share = {
'export_path': export_path,
'service_host': service_host,
'protocol': protocol,
'share_type': share_type,
}
fake_share.update(kwargs)
return {'share': fake_share}
@ddt.ddt
class ShareManageTest(test.TestCase):
"""Share Manage Test."""
def setUp(self):
super(ShareManageTest, self).setUp()
self.controller = share_manage.ShareManageController()
self.resource_name = self.controller.resource_name
self.request = fakes.HTTPRequest.blank('/share/manage',
use_admin_context=True)
self.context = self.request.environ['manila.context']
self.mock_policy_check = self.mock_object(
policy, 'check_policy', mock.Mock(return_value=True))
self.mock_object(
common, 'validate_public_share_policy',
mock.Mock(side_effect=lambda *args, **kwargs: args[1]))
@ddt.data({},
{'shares': {}},
{'share': get_fake_manage_body('', None, None)},
{'share': get_fake_manage_body(
export_path={'not_path': '/fake'})})
def test_share_manage_invalid_body(self, body):
self.assertRaises(webob.exc.HTTPUnprocessableEntity,
self.controller.create,
self.request,
body)
self.mock_policy_check.assert_called_once_with(
self.context, self.resource_name, 'manage')
def test_share_manage_service_not_found(self):
body = get_fake_manage_body()
self.mock_object(db_api, 'service_get_by_host_and_topic', mock.Mock(
side_effect=exception.ServiceNotFound(service_id='fake')))
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.create,
self.request,
body)
self.mock_policy_check.assert_called_once_with(
self.context, self.resource_name, 'manage')
def test_share_manage_share_type_not_found(self):
body = get_fake_manage_body()
self.mock_object(db_api, 'service_get_by_host_and_topic', mock.Mock())
self.mock_object(utils, 'service_is_up', mock.Mock(return_value=True))
self.mock_object(db_api, 'share_type_get_by_name', mock.Mock(
side_effect=exception.ShareTypeNotFoundByName(
share_type_name='fake')))
self.assertRaises(webob.exc.HTTPNotFound,
self.controller.create,
self.request,
body)
self.mock_policy_check.assert_called_once_with(
self.context, self.resource_name, 'manage')
def _setup_manage_mocks(self, service_is_up=True):
self.mock_object(db_api, 'service_get_by_host_and_topic', mock.Mock(
return_value={'host': 'fake'}))
self.mock_object(share_types, 'get_share_type_by_name_or_id',
mock.Mock(return_value={'id': 'fake'}))
self.mock_object(utils, 'service_is_up', mock.Mock(
return_value=service_is_up))
@ddt.data({'service_is_up': False, 'service_host': 'fake@host#POOL'},
{'service_is_up': True, 'service_host': 'fake@host'})
def test_share_manage_bad_request(self, settings):
body = get_fake_manage_body(service_host=settings.pop('service_host'))
self._setup_manage_mocks(**settings)
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.create,
self.request,
body)
self.mock_policy_check.assert_called_once_with(
self.context, self.resource_name, 'manage')
def test_share_manage_duplicate_share(self):
body = get_fake_manage_body()
exc = exception.InvalidShare(reason="fake")
self._setup_manage_mocks()
self.mock_object(share_api.API, 'manage', mock.Mock(side_effect=exc))
self.assertRaises(webob.exc.HTTPConflict,
self.controller.create,
self.request,
body)
self.mock_policy_check.assert_called_once_with(
self.context, self.resource_name, 'manage')
def test_share_manage_forbidden_manage(self):
body = get_fake_manage_body()
self._setup_manage_mocks()
error = mock.Mock(side_effect=exception.PolicyNotAuthorized(action=''))
self.mock_object(share_api.API, 'manage', error)
self.assertRaises(webob.exc.HTTPForbidden,
self.controller.create,
self.request,
body)
self.mock_policy_check.assert_called_once_with(
self.context, self.resource_name, 'manage')
def test_share_manage_forbidden_validate_service_host(self):
body = get_fake_manage_body()
self._setup_manage_mocks()
error = mock.Mock(side_effect=exception.PolicyNotAuthorized(action=''))
self.mock_object(utils, 'service_is_up', mock.Mock(side_effect=error))
self.assertRaises(webob.exc.HTTPForbidden,
self.controller.create,
self.request,
body)
self.mock_policy_check.assert_called_once_with(
self.context, self.resource_name, 'manage')
def test_share_manage_invalid_input(self):
body = get_fake_manage_body()
self._setup_manage_mocks()
error = mock.Mock(side_effect=exception.InvalidInput(message="",
reason="fake"))
self.mock_object(share_api.API, 'manage', mock.Mock(side_effect=error))
self.assertRaises(webob.exc.HTTPBadRequest,
self.controller.create,
self.request,
body)
self.mock_policy_check.assert_called_once_with(
self.context, self.resource_name, 'manage')
def test_share_manage_invalid_share_server(self):
body = get_fake_manage_body()
self._setup_manage_mocks()
error = mock.Mock(
side_effect=exception.InvalidShareServer(reason="")
)
self.mock_object(share_api.API, 'manage', mock.Mock(side_effect=error))
self.assertRaises(webob.exc.HTTPConflict,
self.controller.create,
self.request,
body)
self.mock_policy_check.assert_called_once_with(
self.context, self.resource_name, 'manage')
@ddt.data(
get_fake_manage_body(name='foo', description='bar'),
get_fake_manage_body(display_name='foo', description='bar'),
get_fake_manage_body(name='foo', display_description='bar'),
get_fake_manage_body(display_name='foo', display_description='bar'),
get_fake_manage_body(display_name='foo', display_description='bar',
driver_options=dict(volume_id='quuz')),
get_fake_manage_body(display_name='foo', display_description='bar',
export_path={'path': '/fake'}),
)
def test_share_manage(self, data):
self._setup_manage_mocks()
return_share = {'share_type_id': '', 'id': 'fake'}
self.mock_object(
share_api.API, 'manage', mock.Mock(return_value=return_share))
share = {
'host': data['share']['service_host'],
'export_location_path': data['share']['export_path'],
'share_proto': data['share']['protocol'].upper(),
'share_type_id': 'fake',
'display_name': 'foo',
'display_description': 'bar',
}
data['share']['is_public'] = 'foo'
driver_options = data['share'].get('driver_options', {})
if isinstance(share['export_location_path'], dict):
share['export_location_path'] = (
share['export_location_path']['path']
)
actual_result = self.controller.create(self.request, data)
share_api.API.manage.assert_called_once_with(
mock.ANY, share, driver_options)
self.assertIsNotNone(actual_result)
self.mock_policy_check.assert_called_once_with(
self.context, self.resource_name, 'manage')
def test_share_manage_allow_dhss_true(self):
self._setup_manage_mocks()
data = get_fake_manage_body(name='foo', description='bar')
return_share = {'share_type_id': '', 'id': 'fake'}
self.mock_object(
share_api.API, 'manage', mock.Mock(return_value=return_share))
share = {
'host': data['share']['service_host'],
'export_location_path': data['share']['export_path'],
'share_proto': data['share']['protocol'].upper(),
'share_type_id': 'fake',
'display_name': 'foo',
'display_description': 'bar',
'share_server_id': 'fake'
}
data['share']['share_server_id'] = 'fake'
driver_options = data['share'].get('driver_options', {})
self.controller._manage(self.request, data, allow_dhss_true=True)
share_api.API.manage.assert_called_once_with(
self.context, share, driver_options
)
self.mock_policy_check.assert_called_once_with(
self.context, self.resource_name, 'manage')
def test_wrong_permissions(self):
body = get_fake_manage_body()
fake_req = fakes.HTTPRequest.blank(
'/share/manage', use_admin_context=False)
self.assertRaises(webob.exc.HTTPForbidden,
self.controller.create,
fake_req, body)
self.mock_policy_check.assert_called_once_with(
fake_req.environ['manila.context'], self.resource_name, 'manage')
View Git Blame Copy Permalink