Merge "Add S-RBAC tests for manila"

This commit is contained in:
Zuul 2022-10-04 13:32:48 +00:00 committed by Gerrit Code Review
commit e344e37d62
9 changed files with 1850 additions and 0 deletions

View File

@ -0,0 +1,163 @@
# Copyright 2022 Red Hat, 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 tempest import clients
from tempest import config
from tempest.lib import auth
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from manila_tempest_tests.common import waiters
CONF = config.CONF
class ShareRbacBaseTests(object):
identity_version = 'v3'
protocols = ['nfs', 'cifs', 'glusterfs', 'hdfs', 'cephfs', 'maprfs']
@classmethod
def skip_checks(cls):
super(ShareRbacBaseTests, cls).skip_checks()
if not CONF.enforce_scope.manila:
raise cls.skipException(
"Tempest is not configured to enforce_scope for manila, "
"skipping RBAC tests. To enable these tests set "
"`tempest.conf [enforce_scope] manila=True`."
)
if not CONF.share.default_share_type_name:
raise cls.skipException("Secure rbac tests require a default "
"share type")
if not any(p in CONF.share.enable_protocols for p in cls.protocols):
message = "%s tests are disabled" % cls.protocol
raise cls.skipException(message)
@classmethod
def delete_resource(cls, client, **kwargs):
key_names = {
'st': 'share_type',
'sn': 'share_network',
}
key, resource_id = list(kwargs.items())[0]
key = key.split('_')[0]
resource_name = key_names[key] if key in key_names else key
del_action = getattr(client, 'delete_{}'.format(resource_name))
test_utils.call_and_ignore_notfound_exc(
del_action, resource_id)
test_utils.call_and_ignore_notfound_exc(
client.wait_for_resource_deletion, **kwargs)
@classmethod
def create_share(cls, client, share_type_id, size=None, name=None,
metadata=None):
kwargs = {}
name = name or data_utils.rand_name('share')
metadata = metadata or {}
kwargs.update({
'share_protocol': cls.protocol,
'size': size or CONF.share.share_size,
'name': name or data_utils.rand_name('share'),
'share_type_id': share_type_id,
'metadata': metadata,
})
share = client.create_share(**kwargs)['share']
waiters.wait_for_resource_status(client, share['id'], 'available')
cls.addClassResourceCleanup(
cls.delete_resource, client,
share_id=share['id'])
return share
@classmethod
def create_snapshot(cls, client, share_id, name=None):
name = name or data_utils.rand_name('snapshot')
snapshot = client.create_snapshot(share_id, name=name)['snapshot']
waiters.wait_for_resource_status(
client, snapshot['id'], 'available', resource_name='snapshot')
cls.addClassResourceCleanup(
cls.delete_resource, client, snapshot_id=snapshot['id'])
return snapshot
@classmethod
def create_share_network(cls, client, name=None):
name = name or data_utils.rand_name('share_network')
share_network = client.create_share_network(name=name)['share_network']
cls.addClassResourceCleanup(
cls.delete_resource, client, sn_id=share_network['id'])
return share_network
@classmethod
def get_share_type(cls):
return cls.shares_v2_client.get_default_share_type()['share_type']
def do_request(self, method, expected_status=200, client=None, **payload):
if not client:
client = self.client
if isinstance(expected_status, type(Exception)):
self.assertRaises(expected_status,
getattr(client, method),
**payload)
else:
response = getattr(client, method)(**payload)
self.assertEqual(response.response.status, expected_status)
return response
@classmethod
def setup_user_client(cls, client, project_id=None):
"""Set up project user with its own client.
This is useful for testing protection of resources in separate
projects.
NOTE(lkuchlan): Tempest creates 'project_member' and 'project_reader'
dynamic credentials in different projects. So this method is also
necessary for testing protection of resources in a specific project.
Returns a client object and the user's ID.
"""
projects_client = client.identity_v3.ProjectsClient()
users_client = client.identity_v3.UsersClient()
roles_client = client.identity_v3.RolesClient()
user_dict = {
'name': data_utils.rand_name('user'),
'password': data_utils.rand_password(),
}
user_id = users_client.create_user(
**user_dict)['user']['id']
cls.addClassResourceCleanup(users_client.delete_user, user_id)
if not project_id:
project_id = projects_client.create_project(
data_utils.rand_name())['project']['id']
cls.addClassResourceCleanup(
projects_client.delete_project,
project_id)
member_role_id = roles_client.list_roles(
name='member')['roles'][0]['id']
roles_client.create_user_role_on_project(
project_id, user_id, member_role_id)
creds = auth.KeystoneV3Credentials(
user_id=user_id,
password=user_dict['password'],
project_id=project_id)
auth_provider = clients.get_auth_provider(creds)
creds = auth_provider.fill_credentials()
client = clients.Manager(credentials=creds)
return client

View File

@ -0,0 +1,73 @@
# Copyright 2022 Red Hat, 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.
import abc
from tempest import config
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from testtools import testcase as tc
from manila_tempest_tests.tests.api import base
from manila_tempest_tests.tests.rbac import base as rbac_base
CONF = config.CONF
class ShareRbacSchedulerStatsTests(rbac_base.ShareRbacBaseTests,
metaclass=abc.ABCMeta):
@classmethod
def setup_clients(cls):
super(ShareRbacSchedulerStatsTests, cls).setup_clients()
cls.persona = getattr(cls, 'os_%s' % cls.credentials[0])
cls.client = cls.persona.share_v2.SharesV2Client()
@abc.abstractmethod
def test_list_storage_pools(self):
pass
class ProjectAdminTests(ShareRbacSchedulerStatsTests, base.BaseSharesTest):
credentials = ['project_admin']
@decorators.idempotent_id('1ec4d0f5-0d60-4bbc-88a4-57fa92f6f62f')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_list_storage_pools(self):
self.do_request(
'list_pools', expected_status=200)
class ProjectMemberTests(ShareRbacSchedulerStatsTests, base.BaseSharesTest):
credentials = ['project_member']
@decorators.idempotent_id('905aa5ea-eff9-4022-be41-df7a8593809d')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_list_storage_pools(self):
self.do_request(
'list_pools', expected_status=lib_exc.Forbidden)
class ProjectReaderTests(ShareRbacSchedulerStatsTests, base.BaseSharesTest):
credentials = ['project_reader']
@decorators.idempotent_id('faab12f9-ff51-458d-af47-362d872761e9')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_list_storage_pools(self):
self.do_request(
'list_pools', expected_status=lib_exc.Forbidden)

View File

@ -0,0 +1,291 @@
# Copyright 2022 Red Hat, 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.
import abc
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from testtools import testcase as tc
from manila_tempest_tests.tests.api import base
from manila_tempest_tests.tests.rbac import base as rbac_base
CONF = config.CONF
class ShareRbacShareNetworkTests(rbac_base.ShareRbacBaseTests,
metaclass=abc.ABCMeta):
@classmethod
def setup_clients(cls):
super(ShareRbacShareNetworkTests, cls).setup_clients()
cls.persona = getattr(cls, 'os_%s' % cls.credentials[0])
cls.client = cls.persona.share_v2.SharesV2Client()
@classmethod
def resource_setup(cls):
super(ShareRbacShareNetworkTests, cls).resource_setup()
cls.share_type = cls.get_share_type()
@abc.abstractmethod
def test_create_share_network(self):
pass
@abc.abstractmethod
def test_list_share_network(self):
pass
@abc.abstractmethod
def test_show_share_network(self):
pass
@abc.abstractmethod
def test_delete_share_network(self):
pass
@abc.abstractmethod
def test_update_share_network(self):
pass
class ProjectAdminTests(ShareRbacShareNetworkTests, base.BaseSharesTest):
credentials = ['project_admin', 'project_alt_member']
@classmethod
def setup_clients(cls):
super(ProjectAdminTests, cls).setup_clients()
project_member = cls.setup_user_client(
cls.persona, project_id=cls.persona.credentials.project_id)
cls.share_member_client = project_member.share_v2.SharesV2Client()
cls.alt_project_share_v2_client = (
cls.os_project_alt_member.share_v2.SharesV2Client())
@decorators.idempotent_id('358dd850-cd81-4b81-aefa-3dfcb7aa4551')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_create_share_network(self):
share_network = self.do_request(
'create_share_network', expected_status=200)['share_network']
self.addCleanup(
self.delete_resource, self.client, sn_id=share_network['id'])
@decorators.idempotent_id('deb20301-9d7c-4c08-b1f0-fc2c403ea708')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_list_share_network(self):
share_network = self.create_share_network(self.share_member_client)
alt_share_network = self.create_share_network(
self.alt_project_share_v2_client)
params = {"all_tenants": 1}
share_network_list = self.do_request(
'list_share_networks', expected_status=200,
params=params)['share_networks']
share_network_id_list = [
s['id'] for s in share_network_list
]
self.assertIn(share_network['id'], share_network_id_list)
self.assertIn(alt_share_network['id'], share_network_id_list)
@decorators.idempotent_id('43a3be84-d08b-4f17-89cf-02abda6df580')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_show_share_network(self):
share_network = self.create_share_network(self.share_member_client)
self.do_request(
'get_share_network', expected_status=200,
share_network_id=share_network['id'])
alt_share_network = self.create_share_network(
self.alt_project_share_v2_client)
self.do_request(
'get_share_network', expected_status=200,
share_network_id=alt_share_network['id'])
@decorators.idempotent_id('6c403ed6-b810-4794-8e9b-d57f173443a2')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_delete_share_network(self):
share_network = self.create_share_network(self.share_member_client)
self.do_request(
'delete_share_network', expected_status=202,
sn_id=share_network['id'])
alt_share_network = self.create_share_network(
self.alt_project_share_v2_client)
self.do_request(
'delete_share_network', expected_status=202,
sn_id=alt_share_network['id'])
@decorators.idempotent_id('abd2443d-3490-462a-8e51-73b6a8f48795')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_update_share_network(self):
share_network = self.create_share_network(self.share_member_client)
name = data_utils.rand_name("updated_share_network")
self.do_request(
'update_share_network', expected_status=200,
sn_id=share_network['id'], name=name)
alt_share_network = self.create_share_network(
self.alt_project_share_v2_client)
self.do_request(
'update_share_network', expected_status=200,
sn_id=alt_share_network['id'], name=name)
class ProjectMemberTests(ShareRbacShareNetworkTests, base.BaseSharesTest):
credentials = ['project_member', 'project_alt_member']
@classmethod
def setup_clients(cls):
super(ProjectMemberTests, cls).setup_clients()
cls.alt_project_share_v2_client = (
cls.os_project_alt_member.share_v2.SharesV2Client())
@decorators.idempotent_id('d051c749-3d1c-4485-86c5-6eb860b49cad')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_create_share_network(self):
share_network = self.do_request(
'create_share_network', expected_status=200)['share_network']
self.addCleanup(
self.delete_resource, self.client, sn_id=share_network['id'])
@decorators.idempotent_id('ac33cd51-1efe-4aaf-99ab-b510b7551571')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_list_share_network(self):
share_client = getattr(self, 'share_member_client', self.client)
share_network = self.create_share_network(share_client)
alt_share_network = self.create_share_network(
self.alt_project_share_v2_client)
share_network_list = self.do_request(
'list_share_networks', expected_status=200)['share_networks']
share_network_id_list = [
s['id'] for s in share_network_list
]
self.assertIn(share_network['id'], share_network_id_list)
self.assertNotIn(alt_share_network['id'], share_network_id_list)
@decorators.idempotent_id('dc3f8f95-f8c5-4030-93dd-e4c56e40b477')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_show_share_network(self):
share_client = getattr(self, 'share_member_client', self.client)
share_network = self.create_share_network(share_client)
self.do_request(
'get_share_network', expected_status=200,
share_network_id=share_network['id'])
alt_share_network = self.create_share_network(
self.alt_project_share_v2_client)
self.do_request(
'get_share_network', expected_status=lib_exc.NotFound,
share_network_id=alt_share_network['id'])
@decorators.idempotent_id('717977ab-f077-411a-9bdc-06c8ec9d4f8c')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_delete_share_network(self):
share_network = self.create_share_network(self.client)
self.do_request(
'delete_share_network', expected_status=202,
sn_id=share_network['id'])
alt_share_network = self.create_share_network(
self.alt_project_share_v2_client)
self.do_request(
'delete_share_network', expected_status=lib_exc.NotFound,
sn_id=alt_share_network['id'])
@decorators.idempotent_id('d1fce94c-b163-452d-bf79-13b6edf47e30')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_update_share_network(self):
share_network = self.create_share_network(self.client)
name = data_utils.rand_name("updated_share_network")
self.do_request(
'update_share_network', expected_status=200,
sn_id=share_network['id'], name=name)
alt_share_network = self.create_share_network(
self.alt_project_share_v2_client)
self.do_request(
'update_share_network', expected_status=lib_exc.NotFound,
sn_id=alt_share_network['id'], name=name)
class ProjectReaderTests(ProjectMemberTests):
"""Test suite for basic share network operations by reader user
In order to test certain share operations we must create a share network
resource for this. Since reader user is limited in resources creation, we
are forced to use admin credentials, so we can test other share
operations. In this class we use admin user to create a member user within
reader project. That way we can perform a reader actions on this resource.
"""
credentials = ['project_reader', 'project_admin', 'project_alt_member']
@classmethod
def setup_clients(cls):
super(ProjectReaderTests, cls).setup_clients()
project_member = cls.setup_user_client(
cls.os_project_admin,
project_id=cls.persona.credentials.project_id)
cls.share_member_client = project_member.share_v2.SharesV2Client()
cls.alt_project_share_v2_client = (
cls.os_project_alt_member.share_v2.SharesV2Client())
@decorators.idempotent_id('73dd9f09-7106-4fd5-a484-0eb986002e3b')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_create_share_network(self):
self.do_request(
'create_share_network', expected_status=lib_exc.Forbidden)
@decorators.idempotent_id('841e9e69-2a22-4572-9147-b233c8a842bc')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_list_share_network(self):
super(ProjectReaderTests, self).test_list_share_network()
@decorators.idempotent_id('c98893c8-cdc6-42af-a842-1ee9466904ae')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_show_share_network(self):
super(ProjectReaderTests, self).test_show_share_network()
@decorators.idempotent_id('f8f26bce-ff82-4472-a8dd-0f46c1757386')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_delete_share_network(self):
share_network = self.create_share_network(self.share_member_client)
self.do_request(
'delete_share_network', expected_status=lib_exc.Forbidden,
sn_id=share_network['id'])
alt_share_network = self.create_share_network(
self.alt_project_share_v2_client)
self.do_request(
'delete_share_network', expected_status=lib_exc.Forbidden,
sn_id=alt_share_network['id'])
@decorators.idempotent_id('67b745cd-e669-4872-bbb7-9307960fbd77')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_update_share_network(self):
share_network = self.create_share_network(self.share_member_client)
name = data_utils.rand_name("updated_share_network")
self.do_request(
'update_share_network', expected_status=lib_exc.Forbidden,
sn_id=share_network['id'], name=name)
alt_share_network = self.create_share_network(
self.alt_project_share_v2_client)
self.do_request(
'update_share_network', expected_status=lib_exc.Forbidden,
sn_id=alt_share_network['id'], name=name)

View File

@ -0,0 +1,190 @@
# Copyright 2022 Red Hat, 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.
import abc
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from testtools import testcase as tc
from manila_tempest_tests.tests.api import base
from manila_tempest_tests.tests.rbac import base as rbac_base
CONF = config.CONF
class ShareRbacShareTypesTests(rbac_base.ShareRbacBaseTests,
metaclass=abc.ABCMeta):
@classmethod
def setup_clients(cls):
super(ShareRbacShareTypesTests, cls).setup_clients()
cls.persona = getattr(cls, 'os_%s' % cls.credentials[0])
cls.client = cls.persona.share_v2.SharesV2Client()
@classmethod
def resource_setup(cls):
super(ShareRbacShareTypesTests, cls).resource_setup()
cls.share_type = cls.get_share_type()
def share_type_properties(self):
share_type = {}
share_type['extra_specs'] = {
'driver_handles_share_servers': CONF.share.multitenancy_enabled,
}
share_type['name'] = data_utils.rand_name("share-type")
return share_type
@abc.abstractmethod
def test_create_share_type(self):
pass
@abc.abstractmethod
def test_get_share_type(self):
pass
@abc.abstractmethod
def test_list_share_type(self):
pass
@abc.abstractmethod
def test_update_share_type(self):
pass
class ProjectAdminTests(ShareRbacShareTypesTests, base.BaseSharesTest):
credentials = ['project_admin']
@classmethod
def setup_clients(cls):
super(ProjectAdminTests, cls).setup_clients()
project_member = cls.setup_user_client(
cls.persona, project_id=cls.persona.credentials.project_id)
cls.share_member_client = project_member.share_v2.SharesV2Client()
@decorators.idempotent_id('b24bf137-352a-4ebd-b736-27518d32c1bd')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_create_share_type(self):
share_type = self.do_request(
'create_share_type', expected_status=200,
**self.share_type_properties())['share_type']
self.addCleanup(self.delete_resource, self.client,
st_id=share_type['id'])
@decorators.idempotent_id('741d69f3-b3fe-49cf-9e33-6b0696b353ec')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_get_share_type(self):
self.do_request(
'get_share_type', expected_status=200,
share_type_id=self.share_type['id'])
@decorators.idempotent_id('3f811ac6-a345-424f-863a-1a7a49ba0a32')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_list_share_type(self):
share_type_list = self.do_request(
'list_share_types', expected_status=200)['share_types']
share_type_id_list = [
st['id'] for st in share_type_list
]
self.assertIn(self.share_type['id'], share_type_id_list)
@decorators.idempotent_id('3bb9aaab-3c17-45be-a9b1-dd8b6942cb59')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_update_share_type(self):
share_type = self.client.create_share_type(
**self.share_type_properties())['share_type']
self.addCleanup(self.client.delete_share_type, share_type['id'])
name = data_utils.rand_name("updated_share_type")
self.do_request(
'update_share_type', expected_status=200,
share_type_id=share_type['id'], name=name)
class ProjectMemberTests(ShareRbacShareTypesTests, base.BaseSharesTest):
credentials = ['project_member']
@decorators.idempotent_id('270761cf-07b4-4fc7-96b5-4deb205adce3')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_create_share_type(self):
self.do_request(
'create_share_type', expected_status=lib_exc.Forbidden,
**self.share_type_properties())
@decorators.idempotent_id('d3f53218-d92f-489d-8e2e-985178e7fd02')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_get_share_type(self):
self.do_request(
'get_share_type', expected_status=200,
share_type_id=self.share_type['id'])
@decorators.idempotent_id('757c7ccd-e14e-4c1a-9172-998ae5eed1b8')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_list_share_type(self):
share_type_list = self.do_request(
'list_share_types', expected_status=200)['share_types']
share_type_id_list = [
st['id'] for st in share_type_list
]
self.assertIn(self.share_type['id'], share_type_id_list)
@decorators.idempotent_id('5210170c-b749-4645-a86f-7347c3ba3e99')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_update_share_type(self):
name = data_utils.rand_name("updated_share_type")
self.do_request(
'update_share_type', expected_status=lib_exc.Forbidden,
share_type_id=self.share_type['id'], name=name)
class ProjectReaderTests(ShareRbacShareTypesTests, base.BaseSharesTest):
credentials = ['project_reader', 'project_member']
@decorators.idempotent_id('f4c352c4-c12b-4722-9fe7-9a2ec639ee63')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_create_share_type(self):
self.do_request(
'create_share_type', expected_status=lib_exc.Forbidden,
**self.share_type_properties())
@decorators.idempotent_id('e9d9f244-7778-443b-aadc-bac9f2b687b7')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_get_share_type(self):
self.do_request(
'get_share_type', expected_status=200,
share_type_id=self.share_type['id'])
@decorators.idempotent_id('cf0e97f1-4853-4cf2-9e9a-041c6e57bab5')
@tc.attr(base.TAG_POSITIVE, base.TAG_API)
def test_list_share_type(self):
share_type_list = self.do_request(
'list_share_types', expected_status=200)['share_types']
share_type_id_list = [
st['id'] for st in share_type_list
]
self.assertIn(self.share_type['id'], share_type_id_list)
@decorators.idempotent_id('338d579b-ff91-4a30-af53-d0b317919efb')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API)
def test_update_share_type(self):
name = data_utils.rand_name("updated_share_type")
self.do_request(
'update_share_type', expected_status=lib_exc.Forbidden,
share_type_id=self.share_type['id'], name=name)

View File

@ -0,0 +1,686 @@
# Copyright 2022 Red Hat, 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.
import abc
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from testtools import testcase as tc
from manila_tempest_tests.common import waiters
from manila_tempest_tests.tests.api import base
from manila_tempest_tests.tests.rbac import base as rbac_base
CONF = config.CONF
class ShareRbacSharesTests(rbac_base.ShareRbacBaseTests,
metaclass=abc.ABCMeta):
@classmethod
def skip_checks(cls):
super(ShareRbacSharesTests, cls).skip_checks()
if cls.protocol not in CONF.share.enable_protocols:
message = "%s tests are disabled" % cls.protocol
raise cls.skipException(message)
@classmethod
def setup_clients(cls):
super(ShareRbacSharesTests, cls).setup_clients()
cls.persona = getattr(cls, 'os_%s' % cls.credentials[0])
cls.client = cls.persona.share_v2.SharesV2Client()
cls.alt_project_share_v2_client = (
cls.os_project_alt_member.share_v2.SharesV2Client())
@classmethod
def resource_setup(cls):
super(ShareRbacSharesTests, cls).resource_setup()
cls.share_type = cls.get_share_type()
def share(self, share_type_id, size=None):
share = {}
share['name'] = data_utils.rand_name('share')
share['size'] = size or CONF.share.share_size
share['share_type_id'] = share_type_id
share['share_protocol'] = self.protocol
return share
@abc.abstractmethod
def test_get_share(self):
pass
@abc.abstractmethod
def test_list_share(self):
pass
@abc.abstractmethod
def test_create_share(self):
pass
@abc.abstractmethod
def test_delete_share(self):
pass
@abc.abstractmethod
def test_force_delete_share(self):
pass
@abc.abstractmethod
def test_update_share(self):
pass
@abc.abstractmethod
def test_reset_share(self):
pass
@abc.abstractmethod
def test_shrink_share(self):
pass
@abc.abstractmethod
def test_extend_share(self):
pass
@abc.abstractmethod
def test_set_share_metadata(self):
pass
@abc.abstractmethod
def test_get_share_metadata(self):
pass
@abc.abstractmethod
def test_delete_share_metadata(self):
pass
class TestProjectAdminTestsNFS(ShareRbacSharesTests, base.BaseSharesTest):
credentials = ['project_admin', 'project_alt_member']
protocol = 'nfs'
@classmethod
def setup_clients(cls):
super(TestProjectAdminTestsNFS, cls).setup_clients()
project_member = cls.setup_user_client(
cls.persona, project_id=cls.persona.credentials.project_id)
cls.share_member_client = project_member.share_v2.SharesV2Client()
cls.alt_project_share_v2_client = (
cls.os_project_alt_member.share_v2.SharesV2Client())
@decorators.idempotent_id('14a52454-cba0-4973-926a-28e924ae2e63')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_get_share(self):
share = self.create_share(
self.share_member_client, self.share_type['id'])
self.do_request(
'get_share', expected_status=200, share_id=share['id'])
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'get_share', expected_status=200,
share_id=alt_share['id'])
@decorators.idempotent_id('5f8c06e6-5b80-45f8-aefb-1b55617d1bd1')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_list_share(self):
share = self.create_share(
self.share_member_client, self.share_type['id'])
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
params = {"all_tenants": 1}
share_list = self.do_request(
'list_shares', expected_status=200, params=params)['shares']
share_id_list = [
s['id'] for s in share_list
]
self.assertIn(share['id'], share_id_list)
self.assertIn(alt_share['id'], share_id_list)
@decorators.idempotent_id('34b84af3-a9ea-4c19-8414-e4e44648099c')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_create_share(self):
share = self.do_request(
'create_share', expected_status=200,
**self.share(self.share_type['id']))['share']
waiters.wait_for_resource_status(self.client,
share['id'], 'available')
self.addCleanup(self.delete_resource, self.client,
share_id=share['id'])
@decorators.idempotent_id('44f2eae6-44d4-4962-a94a-d2717b74728f')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_delete_share(self):
share = self.create_share(
self.share_member_client, self.share_type['id'])
self.do_request(
'delete_share', expected_status=202, share_id=share['id'])
self.client.wait_for_resource_deletion(share_id=share['id'])
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'delete_share', expected_status=202,
share_id=alt_share['id'])
self.client.wait_for_resource_deletion(share_id=alt_share['id'])
@decorators.idempotent_id('2e915a27-488d-4e33-b2f8-37758ef11653')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_force_delete_share(self):
share = self.create_share(
self.share_member_client, self.share_type['id'])
self.do_request(
'force_delete', expected_status=202, s_id=share['id'])
self.client.wait_for_resource_deletion(share_id=share['id'])
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'force_delete', expected_status=202,
s_id=alt_share['id'])
self.client.wait_for_resource_deletion(share_id=alt_share['id'])
@decorators.idempotent_id('5c2bda4c-0179-4af9-b18c-430a7d31f962')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_update_share(self):
share = self.create_share(
self.share_member_client, self.share_type['id'])
name = data_utils.rand_name("updated_share")
self.do_request(
'update_share', expected_status=200,
share_id=share['id'], name=name)
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
name = data_utils.rand_name("updated_share")
self.do_request(
'update_share', expected_status=200,
share_id=alt_share['id'], name=name)
@decorators.idempotent_id('44fb7049-8fc0-4584-9ff1-7527395d2ec5')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_reset_share(self):
share = self.create_share(
self.share_member_client, self.share_type['id'])
self.do_request(
'reset_state', expected_status=202, s_id=share['id'],
status="error")
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'reset_state', expected_status=202,
s_id=alt_share['id'], status="error")
@decorators.idempotent_id('cc49ae58-6696-4030-a029-a66bae2efa96')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_shrink_share(self):
share = self.create_share(
self.share_member_client, self.share_type['id'],
size=CONF.share.share_size + 1)
self.do_request(
'shrink_share', expected_status=202, share_id=share['id'],
new_size=CONF.share.share_size)
waiters.wait_for_resource_status(self.client, share['id'], 'available')
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'],
size=CONF.share.share_size + 1)
self.do_request(
'shrink_share', expected_status=202,
share_id=alt_share['id'], new_size=CONF.share.share_size)
waiters.wait_for_resource_status(
self.alt_project_share_v2_client, alt_share['id'], 'available')
@decorators.idempotent_id('2cfa04e5-16cc-43e4-b892-c1a11b0a2f2d')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_extend_share(self):
share = self.create_share(
self.share_member_client, self.share_type['id'])
self.do_request(
'extend_share', expected_status=202, share_id=share['id'],
new_size=CONF.share.share_size + 1)
waiters.wait_for_resource_status(self.client, share['id'], 'available')
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'extend_share', expected_status=202,
share_id=alt_share['id'], new_size=CONF.share.share_size + 1)
waiters.wait_for_resource_status(
self.alt_project_share_v2_client, alt_share['id'], 'available')
@decorators.idempotent_id('d6014579-d772-441a-a9b1-01b1e87caeaa')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_set_share_metadata(self):
share = self.create_share(
self.share_member_client, self.share_type['id'])
self.do_request(
'set_metadata', expected_status=200, share_id=share['id'],
metadata={'key': 'value'})
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'set_metadata', expected_status=200,
share_id=alt_share['id'], metadata={'key': 'value'})
@decorators.idempotent_id('2d91e97e-d0e5-4112-8b22-60cd4659586c')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_get_share_metadata(self):
metadata = {'key': 'value'}
share = self.create_share(
self.share_member_client, self.share_type['id'],
metadata=metadata)
self.do_request(
'get_metadata', expected_status=200, share_id=share['id'])
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'],
metadata=metadata)
self.do_request(
'get_metadata', expected_status=200,
share_id=alt_share['id'])
@decorators.idempotent_id('4cd807d6-bac4-4d0f-a207-c84dfe77f032')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_delete_share_metadata(self):
share = self.create_share(
self.share_member_client, self.share_type['id'],
metadata={'key': 'value'})
self.do_request(
'delete_metadata', expected_status=200, share_id=share['id'],
key='key')
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'],
metadata={'key': 'value'})
self.do_request(
'delete_metadata', expected_status=200,
share_id=alt_share['id'], key='key')
class TestProjectMemberTestsNFS(ShareRbacSharesTests, base.BaseSharesTest):
credentials = ['project_member', 'project_alt_member']
protocol = 'nfs'
@classmethod
def setup_clients(cls):
super(TestProjectMemberTestsNFS, cls).setup_clients()
cls.alt_project_share_v2_client = (
cls.os_project_alt_member.share_v2.SharesV2Client())
@decorators.idempotent_id('75b9fd40-ae63-4caf-9c93-0fe24b2ce904')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_get_share(self):
share_client = getattr(self, 'share_member_client', self.client)
share = self.create_share(share_client, self.share_type['id'])
self.do_request(
'get_share', expected_status=200, share_id=share['id'])
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'get_share', expected_status=lib_exc.NotFound,
share_id=alt_share['id'])
@decorators.idempotent_id('92fd157a-f357-4a08-9fc6-9e77a55b89a8')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_list_share(self):
share_client = getattr(self, 'share_member_client', self.client)
share = self.create_share(share_client, self.share_type['id'])
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
# We expect this key to be ignored since project_member isn't an admin
params = {"all_tenants": 1}
share_list = self.do_request(
'list_shares', expected_status=200, params=params)['shares']
share_id_list = [
s['id'] for s in share_list
]
self.assertIn(share['id'], share_id_list)
self.assertNotIn(alt_share['id'], share_id_list)
@decorators.idempotent_id('7a6eef6b-bf8e-4cb3-a39c-6dc7fbe115ab')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_create_share(self):
share = self.do_request(
'create_share', expected_status=200,
**self.share(self.share_type['id']))['share']
waiters.wait_for_resource_status(self.client,
share['id'], 'available')
self.addCleanup(self.client.wait_for_resource_deletion,
share_id=share['id'])
self.addCleanup(self.client.delete_share, share['id'])
@decorators.idempotent_id('6c546ed7-ebfd-4ac5-a626-d333a25a9e66')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_delete_share(self):
share = self.create_share(self.client, self.share_type['id'])
self.do_request(
'delete_share', expected_status=202, share_id=share['id'])
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'delete_share', expected_status=lib_exc.NotFound,
share_id=alt_share['id'])
@decorators.idempotent_id('2349d2b0-6314-4018-85e5-696f8d1ca94a')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_force_delete_share(self):
share_client = getattr(self, 'share_member_client', self.client)
share = self.create_share(share_client, self.share_type['id'])
self.do_request(
'force_delete', expected_status=lib_exc.Forbidden,
s_id=share['id'])
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'force_delete', expected_status=lib_exc.Forbidden,
s_id=alt_share['id'])
@decorators.idempotent_id('20d6360d-5cea-4305-be36-7e1429007598')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_update_share(self):
share = self.create_share(self.client, self.share_type['id'])
name = data_utils.rand_name("rename_share")
self.do_request(
'update_share', expected_status=200, share_id=share['id'],
name=name)
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
alt_name = data_utils.rand_name("rename_share")
self.do_request(
'update_share', expected_status=lib_exc.NotFound,
share_id=alt_share['id'], name=alt_name)
@decorators.idempotent_id('483cbaef-a53d-433a-9259-f2ecc209f405')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_reset_share(self):
share_client = getattr(self, 'share_member_client', self.client)
share = self.create_share(share_client, self.share_type['id'])
self.do_request(
'reset_state', expected_status=lib_exc.Forbidden,
s_id=share['id'], status="error")
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'reset_state', expected_status=lib_exc.NotFound,
s_id=alt_share['id'], status="error")
@decorators.idempotent_id('56a07567-d0a9-460a-9267-fcd82306a371')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_shrink_share(self):
share = self.create_share(self.client, self.share_type['id'], size=2)
self.do_request(
'shrink_share', expected_status=202,
share_id=share['id'], new_size=CONF.share.share_size)
waiters.wait_for_resource_status(self.client, share['id'], 'available')
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'], size=2)
self.do_request(
'shrink_share', expected_status=lib_exc.NotFound,
share_id=alt_share['id'], new_size=CONF.share.share_size)
@decorators.idempotent_id('c09e6a72-5b99-4be6-8ffe-8ecaad0be990')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_extend_share(self):
share = self.create_share(self.client, self.share_type['id'])
self.do_request(
'extend_share', expected_status=202,
share_id=share['id'], new_size=CONF.share.share_size + 1)
waiters.wait_for_resource_status(self.client, share['id'], 'available')
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'extend_share', expected_status=lib_exc.NotFound,
share_id=alt_share['id'], new_size=CONF.share.share_size + 1)
@decorators.idempotent_id('f1c03630-987c-4f19-938d-4a0ef6529177')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_set_share_metadata(self):
share = self.create_share(
self.client, self.share_type['id'])
self.do_request(
'set_metadata', expected_status=200, share_id=share['id'],
metadata={'key': 'value'})
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'set_metadata', expected_status=lib_exc.Forbidden,
share_id=alt_share['id'], metadata={'key': 'value'})
@decorators.idempotent_id('a69a2b85-3374-4621-83a9-89937ddb520b')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_get_share_metadata(self):
metadata = {'key': 'value'}
share_client = getattr(self, 'share_member_client', self.client)
share = self.create_share(share_client, self.share_type['id'],
metadata=metadata)
self.do_request(
'get_metadata', expected_status=200, share_id=share['id'])
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'],
metadata=metadata)
self.do_request(
'get_metadata', expected_status=lib_exc.Forbidden,
share_id=alt_share['id'])
@decorators.idempotent_id('bea5518a-338e-494d-9034-1d03658ed58b')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_delete_share_metadata(self):
share = self.create_share(
self.client, self.share_type['id'], metadata={'key': 'value'})
self.do_request(
'delete_metadata', expected_status=200, share_id=share['id'],
key='key')
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'],
metadata={'key': 'value'})
self.do_request(
'delete_metadata', expected_status=lib_exc.Forbidden,
share_id=alt_share['id'], key='key')
class TestProjectReaderTestsNFS(TestProjectMemberTestsNFS):
"""Test suite for basic share operations by reader user
In order to test certain share operations we must create a share resource
for this. Since reader user is limited in resources creation, we are forced
to use admin credentials, so we can test other share operations.
In this class we use admin user to create a member user within reader
project. That way we can perform a reader actions on this resource.
"""
credentials = ['project_reader', 'project_admin', 'project_alt_member']
@classmethod
def setup_clients(cls):
super(TestProjectReaderTestsNFS, cls).setup_clients()
project_member = cls.setup_user_client(
cls.os_project_admin,
project_id=cls.persona.credentials.project_id)
cls.share_member_client = project_member.share_v2.SharesV2Client()
@decorators.idempotent_id('dc439eaf-c885-4002-be8f-4c488beeca81')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_get_share(self):
super(TestProjectReaderTestsNFS, self).test_get_share()
@decorators.idempotent_id('1fbb1078-4386-4b52-aa88-e6be4a286791')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_list_share(self):
super(TestProjectReaderTestsNFS, self).test_list_share()
@decorators.idempotent_id('350ba4c9-def9-4865-824a-de1ddff5dcf9')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_create_share(self):
self.do_request(
'create_share', expected_status=lib_exc.Forbidden,
**self.share(self.share_type['id']))
@decorators.idempotent_id('eb92b142-fd8d-47e3-99fe-944cce747ad7')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_delete_share(self):
share = self.create_share(
self.share_member_client, self.share_type['id'])
self.do_request(
'delete_share', expected_status=lib_exc.Forbidden,
share_id=share['id'])
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'delete_share', expected_status=lib_exc.NotFound,
share_id=alt_share['id'])
@decorators.idempotent_id('cb040955-5897-409f-aea0-84b6ae16b77e')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_force_delete_share(self):
super(TestProjectReaderTestsNFS, self).test_force_delete_share()
@decorators.idempotent_id('3184269a-11ca-4484-8a4d-b855a6e1800f')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_update_share(self):
share = self.create_share(
self.share_member_client, self.share_type['id'])
name = data_utils.rand_name("rename_share")
self.do_request(
'update_share', expected_status=lib_exc.Forbidden,
share_id=share['id'], name=name)
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
alt_name = data_utils.rand_name("rename_share")
self.do_request(
'update_share', expected_status=lib_exc.Forbidden,
share_id=alt_share['id'], name=alt_name)
@decorators.idempotent_id('e5ae5b56-38c0-44ec-b8e0-4bc2a5c1d28a')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_reset_share(self):
super(TestProjectReaderTestsNFS, self).test_reset_share()
@decorators.idempotent_id('f85818b1-b93a-4b89-8aa4-b099e582be7c')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_shrink_share(self):
share = self.create_share(
self.share_member_client, self.share_type['id'],
size=CONF.share.share_size + 1)
self.do_request(
'shrink_share', expected_status=lib_exc.Forbidden,
share_id=share['id'], new_size=CONF.share.share_size)
waiters.wait_for_resource_status(self.client, share['id'], 'available')
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'],
size=CONF.share.share_size + 1)
self.do_request(
'shrink_share', expected_status=lib_exc.NotFound,
share_id=alt_share['id'], new_size=CONF.share.share_size)
@decorators.idempotent_id('0b57aedb-6b68-498f-814e-173c47e6c307')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_extend_share(self):
share = self.create_share(
self.share_member_client, self.share_type['id'])
self.do_request(
'extend_share', expected_status=lib_exc.Forbidden,
share_id=share['id'], new_size=CONF.share.share_size + 1)
waiters.wait_for_resource_status(self.client, share['id'], 'available')
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'extend_share', expected_status=lib_exc.NotFound,
share_id=alt_share['id'], new_size=CONF.share.share_size + 1)
@decorators.idempotent_id('3def3f4e-33fc-4726-8818-6cffbc2cab51')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_set_share_metadata(self):
share = self.create_share(
self.share_member_client, self.share_type['id'])
self.do_request(
'set_metadata', expected_status=lib_exc.Forbidden,
share_id=share['id'], metadata={'key': 'value'})
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'])
self.do_request(
'set_metadata', expected_status=lib_exc.Forbidden,
share_id=alt_share['id'], metadata={'key': 'value'})
@decorators.idempotent_id('28cacc77-556f-4707-ba2b-5ef3e56d6ef9')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_get_share_metadata(self):
super(TestProjectReaderTestsNFS, self).test_get_share_metadata()
@decorators.idempotent_id('55486589-a4ef-44f2-b489-96bc29dcd243')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_delete_share_metadata(self):
share = self.create_share(
self.share_member_client, self.share_type['id'],
metadata={'key': 'value'})
self.do_request(
'delete_metadata', expected_status=lib_exc.Forbidden,
share_id=share['id'], key='key')
alt_share = self.create_share(
self.alt_project_share_v2_client, self.share_type['id'],
metadata={'key': 'value'})
self.do_request(
'delete_metadata', expected_status=lib_exc.Forbidden,
share_id=alt_share['id'], key='key')
class TestProjectAdminTestsCEPHFS(TestProjectAdminTestsNFS):
protocol = 'cephfs'
class TestProjectMemberTestsCEPHFS(TestProjectMemberTestsNFS):
protocol = 'cephfs'
class TestProjectReaderTestsCEPHFS(TestProjectReaderTestsNFS):
protocol = 'cephfs'
class TestProjectAdminTestsCIFS(TestProjectAdminTestsNFS):
protocol = 'cifs'
class TestProjectMemberTestsCIFS(TestProjectMemberTestsNFS):
protocol = 'cifs'
class TestProjectReaderTestsCIFS(TestProjectReaderTestsNFS):
protocol = 'cifs'

View File

@ -0,0 +1,430 @@
# Copyright 2022 Red Hat, 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.
import abc
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
from testtools import testcase as tc
from manila_tempest_tests.common import waiters
from manila_tempest_tests.tests.api import base
from manila_tempest_tests.tests.rbac import base as rbac_base
CONF = config.CONF
class ShareRbacSnapshotsTests(rbac_base.ShareRbacBaseTests,
metaclass=abc.ABCMeta):
@classmethod
def skip_checks(cls):
super(ShareRbacSnapshotsTests, cls).skip_checks()
if cls.protocol not in CONF.share.enable_protocols:
message = "%s tests are disabled" % cls.protocol
raise cls.skipException(message)
@classmethod
def setup_clients(cls):
super(ShareRbacSnapshotsTests, cls).setup_clients()
cls.persona = getattr(cls, 'os_%s' % cls.credentials[0])
cls.client = cls.persona.share_v2.SharesV2Client()
cls.alt_project_share_v2_client = (
cls.os_project_alt_member.share_v2.SharesV2Client())
@abc.abstractmethod
def test_get_snapshot(self):
pass
@abc.abstractmethod
def test_list_snapshot(self):
pass
@abc.abstractmethod
def test_create_snapshot(self):
pass
@abc.abstractmethod
def test_delete_snapshot(self):
pass
@abc.abstractmethod
def test_force_delete_snapshot(self):
pass
@abc.abstractmethod
def test_rename_snapshot(self):
pass
@abc.abstractmethod
def test_reset_snapshot(self):
pass
class TestProjectAdminTestsNFS(ShareRbacSnapshotsTests, base.BaseSharesTest):
credentials = ['project_admin', 'project_alt_member']
protocol = 'nfs'
@classmethod
def setup_clients(cls):
super(TestProjectAdminTestsNFS, cls).setup_clients()
project_member = cls.setup_user_client(
cls.persona, project_id=cls.persona.credentials.project_id)
cls.share_member_client = project_member.share_v2.SharesV2Client()
@classmethod
def resource_setup(cls):
super(TestProjectAdminTestsNFS, cls).resource_setup()
share_type = cls.get_share_type()
cls.share = cls.create_share(cls.client, share_type['id'])
cls.alt_share = cls.create_share(
cls.alt_project_share_v2_client, share_type['id'])
@decorators.idempotent_id('e55b1a01-0fcb-42aa-8cc4-b041fc75f1e4')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_get_snapshot(self):
snapshot = self.create_snapshot(
self.share_member_client, self.share['id'])
self.do_request(
'get_snapshot', expected_status=200, snapshot_id=snapshot['id'])
alt_snapshot = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
self.do_request(
'get_snapshot', expected_status=200,
snapshot_id=alt_snapshot['id'])
@decorators.idempotent_id('3b209017-f5ad-4daa-8932-582a75975bbe')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_list_snapshot(self):
snap = self.create_snapshot(
self.share_member_client, self.share['id'])
alt_snap = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
params = {"all_tenants": 1}
snapshot_list = self.do_request(
'list_snapshots', expected_status=200, params=params)['snapshots']
snapshot_id_list = [
s['id'] for s in snapshot_list
]
self.assertIn(snap['id'], snapshot_id_list)
self.assertIn(alt_snap['id'], snapshot_id_list)
@decorators.idempotent_id('2b90d3e9-ec71-468a-86e9-e8955139ad48')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_create_snapshot(self):
snapshot = self.do_request(
'create_snapshot', expected_status=202,
share_id=self.share['id'])['snapshot']
waiters.wait_for_resource_status(
self.client, snapshot['id'], 'available', resource_name='snapshot')
self.addCleanup(self.delete_resource, self.client,
snapshot_id=snapshot['id'])
@decorators.idempotent_id('6de91ee0-d27e-409a-957b-75489d4e7291')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_delete_snapshot(self):
snap = self.create_snapshot(
self.share_member_client, self.share['id'])
self.do_request(
'delete_snapshot', expected_status=202, snap_id=snap['id'])
self.client.wait_for_resource_deletion(snapshot_id=snap['id'])
alt_snap = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
self.do_request(
'delete_snapshot', expected_status=202,
snap_id=alt_snap['id'])
self.client.wait_for_resource_deletion(snapshot_id=alt_snap['id'])
@decorators.idempotent_id('3ac10dfb-3445-4052-855a-a17056d16a9c')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_force_delete_snapshot(self):
snap = self.create_snapshot(
self.share_member_client, self.share['id'])
self.do_request(
'force_delete', expected_status=202, s_id=snap['id'],
s_type='snapshots')
self.client.wait_for_resource_deletion(snapshot_id=snap['id'])
alt_snap = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
self.do_request(
'force_delete', expected_status=202,
s_id=alt_snap['id'], s_type='snapshots')
self.client.wait_for_resource_deletion(snapshot_id=alt_snap['id'])
@decorators.idempotent_id('513c8fef-9597-4e6c-a811-fb89b456d457')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_rename_snapshot(self):
snap = self.create_snapshot(
self.share_member_client, self.share['id'])
name = data_utils.rand_name("updated_snapshot")
self.do_request(
'rename_snapshot', expected_status=200, snapshot_id=snap['id'],
name=name)
alt_snap = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
self.do_request(
'rename_snapshot', expected_status=200,
snapshot_id=alt_snap['id'], name=name)
@decorators.idempotent_id('a5e99bfb-8767-4680-9e39-bde767e4b8f8')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_reset_snapshot(self):
snap = self.create_snapshot(
self.share_member_client, self.share['id'])
self.do_request(
'snapshot_reset_state', expected_status=202,
snapshot_id=snap['id'], status='error')
alt_snap = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
self.do_request(
'snapshot_reset_state', expected_status=202,
snapshot_id=alt_snap['id'], status='error')
class TestProjectMemberTestsNFS(ShareRbacSnapshotsTests, base.BaseSharesTest):
credentials = ['project_member', 'project_alt_member']
protocol = 'nfs'
@classmethod
def resource_setup(cls):
super(TestProjectMemberTestsNFS, cls).resource_setup()
share_type = cls.get_share_type()
share_client = getattr(cls, 'share_member_client', cls.client)
cls.share = cls.create_share(share_client, share_type['id'])
cls.alt_share = cls.create_share(
cls.alt_project_share_v2_client, share_type['id'])
@decorators.idempotent_id('4ba65029-5c8b-4e96-940a-094d9f662cf6')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_get_snapshot(self):
share_client = getattr(self, 'share_member_client', self.client)
snapshot = self.create_snapshot(share_client, self.share['id'])
self.do_request(
'get_snapshot', expected_status=200, snapshot_id=snapshot['id'])
alt_snapshot = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
self.do_request(
'get_snapshot', expected_status=lib_exc.NotFound,
snapshot_id=alt_snapshot['id'])
@decorators.idempotent_id('0dcc1f68-86e2-432e-ad50-51c3cb78b986')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_list_snapshot(self):
share_client = getattr(self, 'share_member_client', self.client)
snap = self.create_snapshot(share_client, self.share['id'])
alt_snap = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
# We expect this key to be ignored since project_member isn't an admin
params = {"all_tenants": 1}
snapshot_list = self.do_request(
'list_snapshots', expected_status=200, params=params)['snapshots']
snapshot_id_list = [
s['id'] for s in snapshot_list
]
self.assertIn(snap['id'], snapshot_id_list)
self.assertNotIn(alt_snap['id'], snapshot_id_list)
@decorators.idempotent_id('d880b3f0-9027-4141-b28a-13e797919af7')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_create_snapshot(self):
snapshot = self.do_request(
'create_snapshot', expected_status=202,
share_id=self.share['id'])['snapshot']
waiters.wait_for_resource_status(
self.client, snapshot['id'], 'available', resource_name='snapshot')
self.addCleanup(self.delete_resource, self.client,
snapshot_id=snapshot['id'])
@decorators.idempotent_id('e3fdd270-971f-4478-9e64-9bd11166bab6')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_delete_snapshot(self):
snap = self.create_snapshot(self.client, self.share['id'])
self.do_request(
'delete_snapshot', expected_status=202, snap_id=snap['id'])
self.client.wait_for_resource_deletion(snapshot_id=snap['id'])
alt_snap = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
self.do_request(
'delete_snapshot', expected_status=lib_exc.NotFound,
snap_id=alt_snap['id'])
@decorators.idempotent_id('a93d6946-1d86-40a1-af01-90e843f8f575')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_force_delete_snapshot(self):
share_client = getattr(self, 'share_member_client', self.client)
snap = self.create_snapshot(share_client, self.share['id'])
self.do_request(
'force_delete', expected_status=lib_exc.Forbidden, s_id=snap['id'],
s_type='snapshots')
alt_snap = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
self.do_request(
'force_delete', expected_status=lib_exc.Forbidden,
s_id=alt_snap['id'], s_type='snapshots')
@decorators.idempotent_id('6da7bf79-25ab-4475-a5e0-1046781e9bc7')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_rename_snapshot(self):
snap = self.create_snapshot(self.client, self.share['id'])
name = data_utils.rand_name("updated_snapshot")
self.do_request(
'rename_snapshot', expected_status=200, snapshot_id=snap['id'],
name=name)
alt_snap = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
self.do_request(
'rename_snapshot', expected_status=lib_exc.NotFound,
snapshot_id=alt_snap['id'], name=name)
@decorators.idempotent_id('22ba2e2e-6788-4075-9e92-af140d3b1238')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_reset_snapshot(self):
share_client = getattr(self, 'share_member_client', self.client)
snap = self.create_snapshot(share_client, self.share['id'])
self.do_request(
'snapshot_reset_state', expected_status=lib_exc.Forbidden,
snapshot_id=snap['id'], status='error')
alt_snap = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
self.do_request(
'snapshot_reset_state', expected_status=lib_exc.Forbidden,
snapshot_id=alt_snap['id'], status='error')
class TestProjectReaderTestsNFS(TestProjectMemberTestsNFS):
"""Test suite for basic share snapshot operations by reader user
In order to test certain share operations we must create a share snapshot
resource for this. Since reader user is limited in resources creation, we
are forced to use admin credentials, so we can test other share
operations. In this class we use admin user to create a member user within
reader project. That way we can perform a reader actions on this resource.
"""
credentials = ['project_reader', 'project_admin', 'project_alt_member']
@classmethod
def setup_clients(cls):
super(TestProjectReaderTestsNFS, cls).setup_clients()
project_member = cls.setup_user_client(
cls.os_project_admin,
project_id=cls.persona.credentials.project_id)
cls.share_member_client = project_member.share_v2.SharesV2Client()
@classmethod
def resource_setup(cls):
super(TestProjectReaderTestsNFS, cls).resource_setup()
share_type = cls.get_share_type()
cls.share = cls.create_share(cls.share_member_client, share_type['id'])
cls.alt_share = cls.create_share(
cls.alt_project_share_v2_client, share_type['id'])
@decorators.idempotent_id('46a09178-0264-4f56-9a5f-9a0583e72e4d')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_get_snapshot(self):
super(TestProjectReaderTestsNFS, self).test_get_snapshot()
@decorators.idempotent_id('fef4285a-a489-4fec-97af-763c2e33282e')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_list_snapshot(self):
super(TestProjectReaderTestsNFS, self).test_list_snapshot()
@decorators.idempotent_id('17a80156-8cd6-420e-8ffe-97103edef4c3')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_create_snapshot(self):
self.do_request(
'create_snapshot', expected_status=lib_exc.Forbidden,
share_id=self.share['id'])
@decorators.idempotent_id('b0ca5483-ebdb-484c-a975-525e4d7deca2')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_delete_snapshot(self):
snap = self.create_snapshot(self.share_member_client, self.share['id'])
self.do_request(
'delete_snapshot', expected_status=lib_exc.Forbidden,
snap_id=snap['id'])
alt_snap = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
self.do_request(
'delete_snapshot', expected_status=lib_exc.NotFound,
snap_id=alt_snap['id'])
@decorators.idempotent_id('ed0af390-e3d0-432b-9147-c0d569181b92')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_force_delete_snapshot(self):
super(TestProjectReaderTestsNFS, self).test_force_delete_snapshot()
@decorators.idempotent_id('21db863f-c2a4-4d07-b435-2a000255ea3b')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_rename_snapshot(self):
snap = self.create_snapshot(self.share_member_client, self.share['id'])
name = data_utils.rand_name("updated_snapshot")
self.do_request(
'rename_snapshot', expected_status=lib_exc.Forbidden,
snapshot_id=snap['id'], name=name)
alt_snap = self.create_snapshot(
self.alt_project_share_v2_client, self.alt_share['id'])
self.do_request(
'rename_snapshot', expected_status=lib_exc.NotFound,
snapshot_id=alt_snap['id'], name=name)
@decorators.idempotent_id('b8c9c9a4-3b2a-4b1c-80d8-2ec87d708111')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_reset_snapshot(self):
super(TestProjectReaderTestsNFS, self).test_reset_snapshot()
class TestProjectAdminTestsCEPHFS(TestProjectAdminTestsNFS):
protocol = 'cephfs'
class TestProjectMemberTestsCEPHFS(TestProjectMemberTestsNFS):
protocol = 'cephfs'
class TestProjectReaderTestsCEPHFS(TestProjectReaderTestsNFS):
protocol = 'cephfs'
class TestProjectAdminTestsCIFS(TestProjectAdminTestsNFS):
protocol = 'cifs'
class TestProjectMemberTestsCIFS(TestProjectMemberTestsNFS):
protocol = 'cifs'
class TestProjectReaderTestsCIFS(TestProjectReaderTestsNFS):
protocol = 'cifs'

View File

@ -638,6 +638,22 @@
# read-only access rules not supported
enable_ro_access_level_for_protocols: ""
- job:
name: manila-tempest-plugin-dummy-no-dhss-rbac
parent: manila-tempest-plugin-dummy-no-dhss
description: |
This job runs the devstack with scope checks enabled,
on the Dummy driver with DHSS=False.
vars:
tempest_test_regex: 'manila_tempest_tests.tests.rbac'
devstack_localrc:
MANILA_ENFORCE_SCOPE: true
devstack_local_conf:
test-config:
"$TEMPEST_CONFIG":
share:
default_share_type_name: default
- job:
name: manila-tempest-plugin-lvm-fips-base
parent: manila-tempest-plugin-lvm-base

View File

@ -11,6 +11,7 @@
- manila-tempest-plugin-lvm-yoga
- manila-tempest-plugin-lvm-xena
- manila-tempest-plugin-lvm-wallaby
- manila-tempest-plugin-dummy-no-dhss-rbac
- manila-tempest-plugin-container:
voting: false
- manila-tempest-plugin-glusterfs-nfs: