openstack/cyborg
Code Issues Proposed changes

Add policies tests for existing device_profile APIs

Browse Source 
This patch added policies tests for existing device_profile APIs

Change-Id: I99f2f88c157f7e412b887ac8fc1abb19b735291a
Story: 2007024
Task: 40313
This commit is contained in:
Yumeng Bao
2020-09-03 20:34:17 +08:00
parent 21c43211d3
commit 27d5a7de39
4 changed files with 244 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
cyborg/tests/unit/api/controllers/v2/test_device_profiles.py
View File

@@ -71,21 +71,6 @@ class TestDeviceProfileController(v2_test.APITestV2):
for in_dp, out_dp in zip(self.fake_dp_objs, out_dps):
self._validate_dp(in_dp, out_dp)
def test_create_with_non_admin(self):
value = {"is_admin": False, "roles": "user", "is_admin_project": False}
ct = self.gen_context(value)
headers = self.gen_headers(ct)
dp = [self.fake_dps[0]]
dp[0]['created_at'] = str(dp[0]['created_at'])
exc = None
try:
self.post_json(self.DP_URL, dp, headers=headers)
except Exception as e:
exc = e
# Cyborg does not raise different exception when policy check failed
# now, improve this case with assertRaises later.
self.assertIn("Bad response: 403 Forbidden", exc.args[0])
@mock.patch('cyborg.conductor.rpcapi.ConductorAPI.device_profile_create')
def test_create(self, mock_cond_dp):
dp = [self.fake_dps[0]]
@@ -109,18 +94,3 @@ class TestDeviceProfileController(v2_test.APITestV2):
url = self.DP_URL + "/mydp"
response = self.delete(url, headers=self.headers)
self.assertEqual(http_client.NO_CONTENT, response.status_int)
def test_delete_with_non_default(self):
value = {"is_admin": False, "roles": "user", "is_admin_project": False}
ct = self.gen_context(value)
headers = self.gen_headers(ct)
dp = self.fake_dp_objs[0]
url = self.DP_URL + '/%s'
exc = None
try:
self.delete(url % dp['uuid'], headers=headers)
except Exception as e:
exc = e
# Cyborg does not raise different exception when policy check failed
# now, improve this case with assertRaises later.
self.assertIn("Bad response: 403 Forbidden", exc.args[0])

0
cyborg/tests/unit/policies/__init__.py Normal file
View File

90
cyborg/tests/unit/policies/base.py Normal file
View File

@@ -0,0 +1,90 @@
# Copyright 2020 ZTE Corporation.
# 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 oslo_log import log as logging
from oslo_utils.fixture import uuidsentinel as uuids
from cyborg import context as cyborg_context
from cyborg.tests.unit.api.controllers.v2 import base as v2_test
from cyborg.tests.unit import policy_fixture
LOG = logging.getLogger(__name__)
class BasePolicyTest(v2_test.APITestV2):
def setUp(self):
super(BasePolicyTest, self).setUp()
self.policy = self.useFixture(policy_fixture.PolicyFixture())
self.admin_project_id = uuids.admin_project_id
self.project_id = uuids.project_id
self.foo_project_id = uuids.foo_project_id
self.project_id_other = uuids.project_id_other
# legacy default role: "default:admin_or_owner"
self.legacy_admin_context = cyborg_context.RequestContext(
user_id="legacy_admin", project_id=self.admin_project_id,
roles='admin')
self.legacy_owner_context = cyborg_context.RequestContext(
user_id="legacy_owner", project_id=self.admin_project_id,
roles='member')
# system scoped users
self.system_admin_context = cyborg_context.RequestContext(
user_id="sys_admin",
roles='admin', system_scope='all')
self.system_member_context = cyborg_context.RequestContext(
user_id="sys_member",
roles='member', system_scope='all')
self.system_reader_context = cyborg_context.RequestContext(
user_id="sys_reader", roles='reader', system_scope='all')
self.system_foo_context = cyborg_context.RequestContext(
user_id="sys_foo", roles='foo', system_scope='all')
# project scoped users
self.project_admin_context = cyborg_context.RequestContext(
user_id="project_admin", project_id=self.project_id,
roles='admin')
self.project_member_context = cyborg_context.RequestContext(
user_id="project_member", project_id=self.project_id,
roles='member')
self.project_reader_context = cyborg_context.RequestContext(
user_id="project_reader", project_id=self.project_id,
roles='reader')
self.project_foo_context = cyborg_context.RequestContext(
user_id="project_foo", project_id=self.project_id,
roles='foo')
self.other_project_member_context = cyborg_context.RequestContext(
user_id="other_project_member",
project_id=self.project_id_other,
roles='member')
self.all_contexts = [
self.legacy_admin_context, self.legacy_owner_context,
self.system_admin_context, self.system_member_context,
self.system_reader_context, self.system_foo_context,
self.project_admin_context, self.project_member_context,
self.project_reader_context, self.other_project_member_context,
self.project_foo_context,
]

154
cyborg/tests/unit/policies/test_device_profiles.py Normal file
View File

@@ -0,0 +1,154 @@
# Copyright 2020 ZTE Corporation.
# 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 oslo_log import log as logging
from oslo_serialization import jsonutils
from six.moves import http_client
from unittest import mock
from cyborg.api.controllers.v2 import device_profiles
from cyborg.tests.unit import fake_device_profile
from cyborg.tests.unit.policies import base
LOG = logging.getLogger(__name__)
DP_URL = '/device_profiles'
class DeviceProfilePolicyTest(base.BasePolicyTest):
"""Test device_profile APIs policies with all possible contexts.
This class defines the set of context with different roles
which are allowed and not allowed to pass the policy checks.
With those set of context, it will call the API operation and
verify the expected behaviour.
"""
def setUp(self):
super(DeviceProfilePolicyTest, self).setUp()
self.controller = device_profiles.DeviceProfilesController()
self.fake_dp_objs = fake_device_profile.get_obj_devprofs()
self.fake_dps = fake_device_profile.get_api_devprofs()
# check both legacy and new policies for create APIs
self.create_authorized_contexts = [
self.legacy_admin_context, # legacy: admin
self.system_admin_context, # new policy: system_admin
self.project_admin_context
]
self.create_unauthorized_contexts = list(
set(self.all_contexts) - set(self.create_authorized_contexts))
# check both legacy and new policies for delete APIs
self.delete_authorized_contexts = [
# legacy: admin_or_owner
self.legacy_admin_context,
# NOTE(yumeng) although the legacy rule is admin_or_owner,
# cyborg does not "actually" support owner check for
# device profile, so we just uncomment legacy_owner_context here.
# If later we need support owner policy, we should recheck here.
# self.legacy_owner_context,
self.system_admin_context, # new policy: system_admin
self.project_admin_context
]
self.delete_unauthorized_contexts = list(
set(self.all_contexts) - set(self.delete_authorized_contexts))
def _validate_links(self, links, dp_uuid):
has_self_link = False
for link in links:
if link['rel'] == 'self':
has_self_link = True
url = link['href']
components = url.split('/')
self.assertEqual(components[-1], dp_uuid)
self.assertTrue(has_self_link)
def _validate_dp(self, in_dp, out_dp):
self.assertEqual(in_dp['name'], out_dp['name'])
self.assertEqual(in_dp['uuid'], out_dp['uuid'])
self.assertEqual(in_dp['groups'], out_dp['groups'])
# Check that the link is properly set up
self._validate_links(out_dp['links'], in_dp['uuid'])
@mock.patch('cyborg.conductor.rpcapi.ConductorAPI.device_profile_create')
def test_create_device_profile_success(self, mock_cond_dp):
mock_cond_dp.return_value = self.fake_dp_objs[0]
dp = [self.fake_dps[0]]
dp[0]['created_at'] = str(dp[0]['created_at'])
for context in self.create_authorized_contexts:
headers = self.gen_headers(context)
response = self.post_json(DP_URL, dp, headers=headers)
out_dp = jsonutils.loads(response.controller_output)
self.assertEqual(http_client.CREATED, response.status_int)
self._validate_dp(dp[0], out_dp)
def test_create_device_profile_forbidden(self):
dp = [self.fake_dps[0]]
dp[0]['created_at'] = str(dp[0]['created_at'])
for context in self.create_unauthorized_contexts:
headers = self.gen_headers(context)
try:
self.post_json(DP_URL, dp, headers=headers)
except Exception as e:
exc = e
self.assertIn("Bad response: 403 Forbidden", exc.args[0])
@mock.patch('cyborg.conductor.rpcapi.ConductorAPI.device_profile_delete')
@mock.patch('cyborg.objects.DeviceProfile.get_by_name')
@mock.patch('cyborg.objects.DeviceProfile.get_by_uuid')
def test_delete_device_profile_success(self, mock_dp_uuid,
mock_dp_name, mock_cond_del):
for context in self.delete_authorized_contexts:
headers = self.gen_headers(context)
# Delete by UUID
url = DP_URL + "/5d2c0797-c3cd-4f4b-b0d0-2cc5e99ef66e"
response = self.delete(url, headers=headers)
self.assertEqual(http_client.NO_CONTENT, response.status_int)
# Delete by name
url = DP_URL + "/mydp"
response = self.delete(url, headers=headers)
self.assertEqual(http_client.NO_CONTENT, response.status_int)
def test_delete_device_profile_forbidden(self):
dp = self.fake_dp_objs[0]
url = DP_URL + '/%s'
exc = None
for context in self.delete_unauthorized_contexts:
headers = self.gen_headers(context)
try:
self.delete(url % dp['uuid'], headers=headers)
except Exception as e:
exc = e
self.assertIn("Bad response: 403 Forbidden", exc.args[0])
class DeviceProfileScopeTypePolicyTest(DeviceProfilePolicyTest):
"""Test device_profile APIs policies with system scope enabled.
This class set the cyborg.conf [oslo_policy] enforce_scope to True
so that we can switch on the scope checking on oslo policy side.
It defines the set of context with scoped token
which are allowed and not allowed to pass the policy checks.
With those set of context, it will run the API operation and
verify the expected behaviour.
"""
def setUp(self):
super(DeviceProfileScopeTypePolicyTest, self).setUp()