openstack
/
manila-tempest-plugin
Code Issues Proposed changes

Merge "Add access rules restriction tests"

This commit is contained in:
Zuul 2024-03-13 16:16:42 +00:00 committed by Gerrit Code Review
parent cbab9fe0c7 d10819b89b
commit 1687dd6ef6
5 changed files with 207 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
manila_tempest_tests/config.py
View File

@ -40,7 +40,7 @@ ShareGroup = [
"This value is only used to validate the versions "
"response from Manila."),
cfg.StrOpt("max_api_microversion",
default="2.81",
default="2.82",
help="The maximum api microversion is configured to be the "
"value of the latest microversion supported by Manila."),
cfg.StrOpt("region",

12
manila_tempest_tests/services/share/v2/json/shares_client.py
View File

@ -819,7 +819,8 @@ class SharesV2Client(shares_client.SharesClient):
def create_access_rule(self, share_id, access_type="ip",
access_to="0.0.0.0", access_level=None,
version=LATEST_MICROVERSION, metadata=None,
action_name=None):
action_name=None, lock_visibility=False,
lock_deletion=False):
post_body = {
self._get_access_action_name(version, 'os-allow_access'): {
"access_type": access_type,
@ -829,6 +830,10 @@ class SharesV2Client(shares_client.SharesClient):
}
if metadata is not None:
post_body['allow_access']['metadata'] = metadata
if lock_visibility:
post_body['allow_access']['lock_visibility'] = True
if lock_deletion:
post_body['allow_access']['lock_deletion'] = True
body = json.dumps(post_body)
resp, body = self.post(
"shares/%s/action" % share_id, body, version=version,
@ -872,12 +877,15 @@ class SharesV2Client(shares_client.SharesClient):
return rest_client.ResponseBody(resp, body)
def delete_access_rule(self, share_id, rule_id,
version=LATEST_MICROVERSION, action_name=None):
version=LATEST_MICROVERSION, action_name=None,
unrestrict=False):
post_body = {
self._get_access_action_name(version, 'os-deny_access'): {
"access_id": rule_id,
}
}
if unrestrict:
post_body['deny_access']['unrestrict'] = True
body = json.dumps(post_body)
resp, body = self.post(
"shares/%s/action" % share_id, body, version=version)

14
manila_tempest_tests/tests/api/base.py
View File

@ -1045,7 +1045,8 @@ class BaseSharesTest(test.BaseTestCase):
def allow_access(self, share_id, client=None, access_type=None,
access_level='rw', access_to=None, metadata=None,
version=LATEST_MICROVERSION, status='active',
raise_rule_in_error_state=True, cleanup=True):
raise_rule_in_error_state=True, lock_visibility=False,
lock_deletion=False, cleanup=True):
client = client or self.shares_v2_client
a_type, a_to = utils.get_access_rule_data_from_config(
@ -1058,8 +1059,15 @@ class BaseSharesTest(test.BaseTestCase):
'access_to': access_to,
'access_level': access_level
}
delete_kwargs = (
{'unrestrict': True} if lock_deletion else {}
)
if client is self.shares_v2_client:
kwargs.update({'metadata': metadata, 'version': version})
if lock_visibility:
kwargs.update({'lock_visibility': True})
if lock_deletion:
kwargs.update({'lock_deletion': True})
rule = client.create_access_rule(share_id, **kwargs)['access']
waiters.wait_for_resource_status(
@ -1070,7 +1078,9 @@ class BaseSharesTest(test.BaseTestCase):
self.addCleanup(
client.wait_for_resource_deletion, rule_id=rule['id'],
share_id=share_id, version=version)
self.addCleanup(client.delete_access_rule, share_id, rule['id'])
self.addCleanup(
client.delete_access_rule, share_id, rule['id'],
**delete_kwargs)
return rule

91
manila_tempest_tests/tests/api/test_rules.py
View File

@ -28,6 +28,7 @@ from manila_tempest_tests import utils
CONF = config.CONF
LATEST_MICROVERSION = CONF.share.max_api_microversion
RESTRICTED_RULES_VERSION = '2.82'
def _create_delete_ro_access_rule(self, version):
@ -444,6 +445,10 @@ class ShareRulesTest(base.BaseSharesMixedTest):
cls.share_type = cls.create_share_type()
cls.share_type_id = cls.share_type['id']
cls.share = cls.create_share(share_type_id=cls.share_type_id)
cls.user_project = cls.os_admin.projects_client.show_project(
cls.shares_v2_client.project_id)['project']
cls.new_user = cls.create_user_and_get_client(
project=cls.user_project)
@decorators.idempotent_id('c52e95cc-d6ea-4d02-9b52-cd7c1913dfff')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
@ -519,6 +524,92 @@ class ShareRulesTest(base.BaseSharesMixedTest):
msg = "expected id lists %s times in rule list" % (len(gen))
self.assertEqual(1, len(gen), msg)
@decorators.idempotent_id('3bca373e-e54f-49e1-8789-99a383cf4df3')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
@utils.skip_if_microversion_not_supported(RESTRICTED_RULES_VERSION)
@ddt.data(
*itertools.product(utils.deduplicate(
["2.81", CONF.share.max_api_microversion]), (True, False))
)
@ddt.unpack
def test_list_restricted_rules_from_other_user(
self, older_version, lock_visibility):
# create rule
self.allow_access(
self.share["id"], client=self.shares_v2_client,
access_type=self.access_type, access_to=self.access_to,
lock_visibility=lock_visibility, lock_deletion=True)
rule = (
self.shares_v2_client.list_access_rules(
self.share["id"])["access_list"][0])
rules = self.new_user.shares_v2_client.list_access_rules(
self.share['id'])['access_list']
rules_get_lower_version = (
self.new_user.shares_v2_client.list_access_rules(
self.share['id'], version=older_version)['access_list'])
expected_access_to = '******' if lock_visibility else self.access_to
expected_access_key = (
'******' if lock_visibility else rule['access_key'])
# verify values
rule_latest_rules_api = [r for r in rules if r['id'] == rule['id']][0]
rule_lower_version_rules_api = [r for r in rules_get_lower_version
if r['id'] == rule['id']][0]
self.assertEqual(
expected_access_to, rule_latest_rules_api["access_to"])
self.assertEqual(
expected_access_to,
rule_lower_version_rules_api['access_to'])
if self.access_type == 'cephx':
self.assertEqual(expected_access_key,
rule_latest_rules_api['access_key'])
self.assertEqual(
expected_access_key,
rule_lower_version_rules_api['access_key'])
@decorators.idempotent_id('4829265a-eb32-400d-91a0-be06ce31a2ef')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_admin_listing_restricted_rules(self):
utils.check_skip_if_microversion_not_supported(
RESTRICTED_RULES_VERSION)
# create rule
self.allow_access(
self.share["id"], client=self.shares_v2_client,
access_type=self.access_type, access_to=self.access_to,
lock_visibility=True)
rules = self.admin_shares_v2_client.list_access_rules(
self.share["id"])['access_list']
# ensure admin can see rules even if locked
self.assertEqual(self.access_to, rules[0]["access_to"])
if self.access_type == 'cephx':
self.assertIsNotNone(rules[0]['access_key'])
self.assertFalse(rules[0]['access_key'] == '******')
else:
self.assertIsNone(rules[0]['access_key'])
@decorators.idempotent_id('00202c6c-b4c7-4fa6-933a-562fbffde405')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_admin_delete_restricted_rules(self):
utils.check_skip_if_microversion_not_supported(
RESTRICTED_RULES_VERSION)
# create rule
rule = self.allow_access(
self.share["id"], client=self.shares_v2_client,
access_type=self.access_type, access_to=self.access_to,
lock_visibility=True, lock_deletion=True, cleanup=False)
self.admin_shares_v2_client.delete_access_rule(
self.share['id'], rule['id'], unrestrict=True)
self.shares_v2_client.wait_for_resource_deletion(
rule_id=rule['id'], share_id=self.share['id'])
@decorators.idempotent_id('b77bcbda-9754-48f0-9be6-79341ad1af64')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
@ddt.data(*utils.deduplicate(['1.0', '2.9', '2.27', '2.28',

93
manila_tempest_tests/tests/api/test_rules_negative.py
View File

@ -28,6 +28,7 @@ from manila_tempest_tests import utils
CONF = config.CONF
LATEST_MICROVERSION = CONF.share.max_api_microversion
RESTRICTED_RULES_VERSION = '2.82'
@ddt.ddt
@ -60,6 +61,11 @@ class ShareIpRulesForNFSNegativeTest(base.BaseSharesMixedTest):
# create snapshot
cls.snap = cls.create_snapshot_wait_for_active(cls.share["id"])
cls.user_project = cls.os_admin.projects_client.show_project(
cls.shares_v2_client.project_id)['project']
cls.new_user = cls.create_user_and_get_client(
project=cls.user_project)
@decorators.idempotent_id('16781b45-d2bb-4891-aa97-c28c0769d5bd')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
@ddt.data('1.2.3.256',
@ -170,6 +176,93 @@ class ShareIpRulesForNFSNegativeTest(base.BaseSharesMixedTest):
self.admin_client.create_access_rule,
share["id"], access_type, access_to)
@decorators.idempotent_id('478d3c84-b0ea-41c8-a860-e87f182d991c')
@tc.attr(base.TAG_POSITIVE, base.TAG_API_WITH_BACKEND)
def test_deny_access_unrestrict_other_user_rule(self):
utils.check_skip_if_microversion_not_supported(
RESTRICTED_RULES_VERSION)
access_type, access_to = utils.get_access_rule_data_from_config(
self.protocol)
# create rule
rule = self.allow_access(
self.share["id"], client=self.shares_v2_client,
access_type=access_type, access_to=access_to,
lock_visibility=True, lock_deletion=True)
self.assertRaises(
lib_exc.Forbidden,
self.new_user.shares_v2_client.delete_access_rule,
self.share['id'],
rule['id'],
unrestrict=True
)
@decorators.idempotent_id('c107b0b7-7a3e-4114-af64-ca8fe6e836c9')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
@ddt.data(True, False)
def test_deny_access_without_unrestrict_as_owner_user(self, same_user):
utils.check_skip_if_microversion_not_supported(
RESTRICTED_RULES_VERSION)
access_type, access_to = utils.get_access_rule_data_from_config(
self.protocol)
# create rule
rule = self.allow_access(
self.share["id"], client=self.shares_v2_client,
access_type=access_type, access_to=access_to,
lock_visibility=True, lock_deletion=True)
client = (
self.shares_v2_client
if same_user else self.new_user.shares_v2_client
)
self.assertRaises(
lib_exc.Forbidden,
client.delete_access_rule,
self.share['id'],
rule['id'])
self.assertRaises(
lib_exc.Forbidden,
client.delete_access_rule,
self.share['id'],
rule['id'],
version='2.81'
)
@decorators.idempotent_id('f5b9e7c9-7e6b-4918-a1c4-e03c8d82c46a')
@tc.attr(base.TAG_NEGATIVE, base.TAG_API_WITH_BACKEND)
def test_allow_access_multiple_visibility_locks_not_allowed(self):
utils.check_skip_if_microversion_not_supported(
RESTRICTED_RULES_VERSION)
access_type, access_to = utils.get_access_rule_data_from_config(
self.protocol)
# create rule
rule = self.allow_access(
self.share["id"], client=self.shares_v2_client,
access_type=access_type, access_to=access_to,
lock_visibility=True, lock_deletion=True)
self.assertRaises(
lib_exc.Conflict,
self.shares_v2_client.create_resource_lock,
rule['id'],
"access_rule",
resource_action="show",
lock_reason="locked for testing"
)
self.assertRaises(
lib_exc.Conflict,
self.new_user.shares_v2_client.create_resource_lock,
rule['id'],
"access_rule",
resource_action="show",
lock_reason="locked for testing"
)
@ddt.ddt
class ShareIpRulesForCIFSNegativeTest(ShareIpRulesForNFSNegativeTest):