Add Additional RBAC Test Coverage for Shipyard

This commit does the following:
- Add test coverage for:
  - Actions API
  - Airflow Monitoring API
  - Log Retrieval API
- Add tox.ini
- Add hacking checks
- Fix pep8 issues

Future work needed to fix some of the Actions API RBAC tests

Change-Id: I6e17ffa3ecc3c8a181790bdb79ad6b29fe127114
This commit is contained in:
Rick Bartra 2018-08-20 17:26:18 -04:00
parent e7807b4caf
commit 9f473f288a
19 changed files with 647 additions and 80 deletions

View File

@ -24,7 +24,7 @@ ServiceAvailableGroup = [
] ]
shipyard_group = cfg.OptGroup(name='shipyard', shipyard_group = cfg.OptGroup(name='shipyard',
title='Shipyard service options') title='Shipyard service options')
ShipyardGroup = [ ShipyardGroup = [
cfg.StrOpt('endpoint_type', cfg.StrOpt('endpoint_type',
@ -36,10 +36,9 @@ ShipyardGroup = [
help="Catalog type of the Shipyard service"), help="Catalog type of the Shipyard service"),
] ]
def get_opt_lists(self, conf): def get_opt_lists(self, conf):
""" """Get a list of options for sample config generation"""
Get a list of options for sample config generation
"""
return [ return [
(service_available_group, ServiceAvailableGroup), (service_available_group, ServiceAvailableGroup),
(shipyard_group, ShipyardGroup) (shipyard_group, ShipyardGroup)

View File

@ -0,0 +1,209 @@
# Copyright 2013 IBM Corp.
# Copyright 2017 AT&T 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.
import os
import re
import pycodestyle
PYTHON_CLIENTS = ['cinder', 'glance', 'keystone', 'nova', 'swift', 'neutron',
'ironic', 'heat', 'sahara']
PYTHON_CLIENT_RE = re.compile('import (%s)client' % '|'.join(PYTHON_CLIENTS))
TEST_DEFINITION = re.compile(r'^\s*def test.*')
SETUP_TEARDOWN_CLASS_DEFINITION = re.compile(r'^\s+def (setUp|tearDown)Class')
SCENARIO_DECORATOR = re.compile(r'\s*@.*services\((.*)\)')
VI_HEADER_RE = re.compile(r"^#\s+vim?:.+")
RAND_NAME_HYPHEN_RE = re.compile(r".*rand_name\(.+[\-\_][\"\']\)")
MUTABLE_DEFAULT_ARGS = re.compile(r"^\s*def .+\((.+=\{\}|.+=\[\])")
TESTTOOLS_SKIP_DECORATOR = re.compile(r'\s*@testtools\.skip\((.*)\)')
CLASS = re.compile(r"^class .+")
RBAC_CLASS_NAME_RE = re.compile(r'class .+RbacTest')
RULE_VALIDATION_DECORATOR = re.compile(
r'\s*@rbac_rule_validation.action\(.*')
IDEMPOTENT_ID_DECORATOR = re.compile(r'\s*@decorators\.idempotent_id\((.*)\)')
have_rbac_decorator = False
def import_no_clients_in_api_tests(physical_line, filename):
"""Check for client imports from airship_tempest_plugin/tests/api
T102: Cannot import OpenStack python clients
"""
if "airship_tempest_plugin/tests/api" in filename:
res = PYTHON_CLIENT_RE.match(physical_line)
if res:
return (physical_line.find(res.group(1)),
("T102: python clients import not allowed "
"in airship_tempest_plugin/tests/api/* or "
"airship_tempest_plugin/tests/scenario/* tests"))
def no_setup_teardown_class_for_tests(physical_line, filename):
"""Check that tests do not use setUpClass/tearDownClass
T105: Tests cannot use setUpClass/tearDownClass
"""
if pycodestyle.noqa(physical_line):
return
if SETUP_TEARDOWN_CLASS_DEFINITION.match(physical_line):
return (physical_line.find('def'),
"T105: (setUp|tearDown)Class can not be used in tests")
def no_vi_headers(physical_line, line_number, lines):
"""Check for vi editor configuration in source files.
By default vi modelines can only appear in the first or
last 5 lines of a source file.
T106
"""
# NOTE(gilliard): line_number is 1-indexed
if line_number <= 5 or line_number > len(lines) - 5:
if VI_HEADER_RE.match(physical_line):
return 0, "T106: Don't put vi configuration in source files"
def service_tags_not_in_module_path(physical_line, filename):
"""Check that a service tag isn't in the module path
A service tag should only be added if the service name isn't already in
the module path.
T107
"""
matches = SCENARIO_DECORATOR.match(physical_line)
if matches:
services = matches.group(1).split(',')
for service in services:
service_name = service.strip().strip("'")
modulepath = os.path.split(filename)[0]
if service_name in modulepath:
return (physical_line.find(service_name),
"T107: service tag should not be in path")
def no_hyphen_at_end_of_rand_name(logical_line, filename):
"""Check no hyphen at the end of rand_name() argument
T108
"""
msg = "T108: hyphen should not be specified at the end of rand_name()"
if RAND_NAME_HYPHEN_RE.match(logical_line):
return 0, msg
def no_mutable_default_args(logical_line):
"""Check that mutable object isn't used as default argument
N322: Method's default argument shouldn't be mutable
"""
msg = "N322: Method's default argument shouldn't be mutable!"
if MUTABLE_DEFAULT_ARGS.match(logical_line):
yield (0, msg)
def no_testtools_skip_decorator(logical_line):
"""Check that methods do not have the testtools.skip decorator
T109
"""
if TESTTOOLS_SKIP_DECORATOR.match(logical_line):
yield (0, "T109: Cannot use testtools.skip decorator; instead use "
"decorators.skip_because from tempest.lib")
def use_rand_uuid_instead_of_uuid4(logical_line, filename):
"""Check that tests use data_utils.rand_uuid() instead of uuid.uuid4()
T113
"""
if 'uuid.uuid4()' not in logical_line:
return
msg = ("T113: Tests should use data_utils.rand_uuid()/rand_uuid_hex() "
"instead of uuid.uuid4()/uuid.uuid4().hex")
yield (0, msg)
def no_rbac_rule_validation_decorator(physical_line, filename):
"""Check that each test has the ``rbac_rule_validation.action`` decorator.
Checks whether the test function has "@rbac_rule_validation.action"
above it; otherwise checks that it has "@decorators.idempotent_id" above
it and "@rbac_rule_validation.action" above that.
Assumes that ``rbac_rule_validation.action`` decorator is either the first
or second decorator above the test function; otherwise this check fails.
P100
"""
global have_rbac_decorator
if ("airship_tempest_plugin/tests/api" in filename or
"airship_tempest_plugin/tests/scenario" in filename):
if RULE_VALIDATION_DECORATOR.match(physical_line):
have_rbac_decorator = True
return
if TEST_DEFINITION.match(physical_line):
if not have_rbac_decorator:
return (0, "Must use rbac_rule_validation.action "
"decorator for API and scenario tests")
have_rbac_decorator = False
def no_rbac_suffix_in_test_filename(filename):
"""Check that RBAC filenames end with "_rbac" suffix.
P101
"""
if "airship_tempest_plugin/tests/api" in filename:
if filename.endswith('rbac_base.py'):
return
if not filename.endswith('_rbac.py'):
return 0, "RBAC test filenames must end in _rbac suffix"
def no_rbac_test_suffix_in_test_class_name(physical_line, filename):
"""Check that RBAC class names end with "RbacTest"
P102
"""
if "airship_tempest_plugin/tests/api" in filename:
if filename.endswith('rbac_base.py'):
return
if CLASS.match(physical_line):
if not RBAC_CLASS_NAME_RE.match(physical_line):
return 0, "RBAC test class names must end in 'RbacTest'"
def no_client_alias_in_test_cases(logical_line, filename):
"""Check that test cases don't use "self.client" to define a client.
P103
"""
if "airship_tempest_plugin/tests/api" in filename:
if "self.client" in logical_line or "cls.client" in logical_line:
return 0, "Do not use 'self.client' as a service client alias"
def factory(register):
register(import_no_clients_in_api_tests)
register(no_setup_teardown_class_for_tests)
register(no_vi_headers)
register(no_hyphen_at_end_of_rand_name)
register(no_mutable_default_args)
register(no_testtools_skip_decorator)
register(use_rand_uuid_instead_of_uuid4)
register(service_tags_not_in_module_path)
register(no_rbac_rule_validation_decorator)
register(no_rbac_suffix_in_test_filename)
register(no_rbac_test_suffix_in_test_class_name)

View File

@ -34,7 +34,9 @@ class AirshipRbacPlugin(plugins.TempestPlugin):
config.register_opt_group(conf, project_config.service_available_group, config.register_opt_group(conf, project_config.service_available_group,
project_config.ServiceAvailableGroup) project_config.ServiceAvailableGroup)
config.register_opt_group(conf, project_config.shipyard_group, config.register_opt_group(conf, project_config.shipyard_group,
project_config.ShipyardGroup) project_config.ShipyardGroup)
def get_opt_lists(self): def get_opt_lists(self):
return [ return [
(project_config.service_available_group.name, (project_config.service_available_group.name,

View File

@ -19,16 +19,56 @@ http://airship-shipyard.readthedocs.io/en/latest/API.html#action-api
""" """
from oslo_serialization import jsonutils as json from oslo_serialization import jsonutils as json
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client from tempest.lib.common import rest_client
# NOTE(rb560u): The following will need to be rewritten in the future if
# functional testing is desired:
# - 'def post_actions`
# This initial implementation is just to meet the first use case which is RBAC
# testing. For RBAC testing, we only need to hit the API endpoint and check
# role permission to that API.
class ActionsClient(rest_client.RestClient): class ActionsClient(rest_client.RestClient):
api_version = "v1.0" api_version = "v1.0"
def get_actions(self): def list_actions(self):
resp, body = self.get('actions') resp, body = self.get('actions')
self.expected_success(200, resp.status) self.expected_success(200, resp.status)
body = json.loads(body) body = json.loads(body)
return rest_client.ResponseBody(resp, body) return rest_client.ResponseBody(resp, body)
def create_action(self):
url = "actions"
post_body = json.dumps({})
resp, body = self.post(url, post_body)
self.expected_success(201, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def get_action(self):
resp, body = self.get('actions/1')
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def get_action_validation(self):
resp, body = self.get('actions/1/validationdetails/1')
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def get_action_step(self):
resp, body = self.get('actions/1/steps/1')
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def invoke_action_control(self):
url = "actions/1/pause"
post_body = json.dumps({})
resp, body = self.post(url, post_body)
self.expected_success(202, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)

View File

@ -19,7 +19,6 @@ http://airship-shipyard.readthedocs.io/en/latest/API.html#airflow-monitoring-api
""" """
from oslo_serialization import jsonutils as json from oslo_serialization import jsonutils as json
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client from tempest.lib.common import rest_client
@ -27,8 +26,14 @@ from tempest.lib.common import rest_client
class AirflowMonitoringClient(rest_client.RestClient): class AirflowMonitoringClient(rest_client.RestClient):
api_version = "v1.0" api_version = "v1.0"
def get_workflows(self): def list_workflows(self):
resp, body = self.get('workflows') resp, body = self.get('workflows')
self.expected_success(200, resp.status) self.expected_success(200, resp.status)
body = json.loads(body) body = json.loads(body)
return rest_client.ResponseBody(resp, body) return rest_client.ResponseBody(resp, body)
def get_workflow(self):
resp, body = self.get('workflows/1')
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)

View File

@ -19,7 +19,6 @@ http://airship-shipyard.readthedocs.io/en/latest/API.html#document-staging-api
""" """
from oslo_serialization import jsonutils as json from oslo_serialization import jsonutils as json
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client from tempest.lib.common import rest_client
@ -36,13 +35,13 @@ from tempest.lib.common import rest_client
class DocumentStagingClient(rest_client.RestClient): class DocumentStagingClient(rest_client.RestClient):
api_version = "v1.0" api_version = "v1.0"
def get_configdocs(self): def get_configdocs_status(self):
resp, body = self.get('configdocs') resp, body = self.get('configdocs')
self.expected_success(200, resp.status) self.expected_success(200, resp.status)
body = json.loads(body) body = json.loads(body)
return rest_client.ResponseBody(resp, body) return rest_client.ResponseBody(resp, body)
def post_configdocs(self): def create_configdocs(self):
url = "configdocs/1" url = "configdocs/1"
post_body = json.dumps({}) post_body = json.dumps({})
resp, body = self.post(url, post_body) resp, body = self.post(url, post_body)
@ -50,7 +49,7 @@ class DocumentStagingClient(rest_client.RestClient):
body = json.loads(body) body = json.loads(body)
return rest_client.ResponseBody(resp, body) return rest_client.ResponseBody(resp, body)
def get_configdocs_within_collection(self): def get_configdocs(self):
resp, body = self.get('configdocs/1') resp, body = self.get('configdocs/1')
self.expected_success(200, resp.status) self.expected_success(200, resp.status)
body = json.loads(body) body = json.loads(body)
@ -62,7 +61,7 @@ class DocumentStagingClient(rest_client.RestClient):
body = json.loads(body) body = json.loads(body)
return rest_client.ResponseBody(resp, body) return rest_client.ResponseBody(resp, body)
def post_commitconfigdocs(self): def commit_configdocs(self):
post_body = json.dumps({}) post_body = json.dumps({})
resp, body = self.post("commitconfigdocs", post_body) resp, body = self.post("commitconfigdocs", post_body)
self.expected_success(200, resp.status) self.expected_success(200, resp.status)

View File

@ -0,0 +1,33 @@
# Copyright 2018 AT&T Corp
# 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.
#
"""
http://airship-shipyard.readthedocs.io/en/latest/API.html#airflow-monitoring-api
"""
from oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
class LogRetrievalClient(rest_client.RestClient):
api_version = "v1.0"
def get_action_step_logs(self):
resp, body = self.get('actions/1/steps/1/logs')
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)

View File

@ -1,28 +1,54 @@
shipyard: shipyard:
get_actions: workflow_orchestrator:list_actions:
- admin - admin
- admin_ucp - admin_ucp
- admin_ucp_viewer - admin_ucp_viewer
get_configdocs: workflow_orchestrator:create_action:
- admin
- admin_ucp
workflow_orchestrator:get_action:
- admin - admin
- admin_ucp - admin_ucp
- admin_ucp_viewer - admin_ucp_viewer
get_workflows: workflow_orchestrator:get_action_validation:
- admin - admin
- admin_ucp - admin_ucp
- admin_ucp_viewer - admin_ucp_viewer
post_configdocs: workflow_orchestrator:get_action_step:
- admin
- admin_ucp
get_configdocs_within_collection:
- admin - admin
- admin_ucp - admin_ucp
- admin_ucp_viewer - admin_ucp_viewer
get_renderedconfigdocs: workflow_orchestrator:invoke_action_control:
- admin
- admin_ucp
workflow_orchestrator:get_action_step_logs:
- admin - admin
- admin_ucp - admin_ucp
- admin_ucp_viewer - admin_ucp_viewer
post_commitconfigdocs: workflow_orchestrator:get_configdocs:
- admin
- admin_ucp
- admin_ucp_viewer
workflow_orchestrator:create_configdocs:
- admin
- admin_ucp
workflow_orchestrator:get_configdocs_status:
- admin
- admin_ucp
- admin_ucp_viewer
workflow_orchestrator:get_renderedconfigdocs:
- admin
- admin_ucp
- admin_ucp_viewer
workflow_orchestrator:commit_configdocs:
- admin
- admin_ucp
- admin_ucp_viewer
workflow_orchestrator:list_workflows:
- admin
- admin_ucp
- admin_ucp_viewer
workflow_orchestrator:get_workflow:
- admin - admin
- admin_ucp - admin_ucp
- admin_ucp_viewer - admin_ucp_viewer

View File

@ -14,17 +14,21 @@
# under the License. # under the License.
# #
from airship_tempest_plugin.services.shipyard.json.actions_client import ActionsClient from airship_tempest_plugin.services.shipyard.json.actions_client \
from airship_tempest_plugin.services.shipyard.json.document_staging_client import DocumentStagingClient import ActionsClient
from airship_tempest_plugin.services.shipyard.json.airflow_monitoring_client import AirflowMonitoringClient from airship_tempest_plugin.services.shipyard.json.airflow_monitoring_client \
import AirflowMonitoringClient
from airship_tempest_plugin.services.shipyard.json.document_staging_client \
import DocumentStagingClient
from airship_tempest_plugin.services.shipyard.json.log_retrieval_client \
import LogRetrievalClient
from tempest import config from tempest import config
from tempest import test from tempest import test
from patrole_tempest_plugin import rbac_utils
CONF = config.CONF CONF = config.CONF
class BaseShipyardTest(test.BaseTestCase): class BaseShipyardTest(test.BaseTestCase):
"""Base class for Shipyard tests.""" """Base class for Shipyard tests."""
credentials = ['primary', 'admin'] credentials = ['primary', 'admin']
@ -33,7 +37,8 @@ class BaseShipyardTest(test.BaseTestCase):
def skip_checks(cls): def skip_checks(cls):
super(BaseShipyardTest, cls).skip_checks() super(BaseShipyardTest, cls).skip_checks()
if not CONF.service_available.shipyard: if not CONF.service_available.shipyard:
raise cls.skipException("Shipyard is not enabled in the deployment") raise cls.skipException("Shipyard is not enabled in "
"the deployment")
@classmethod @classmethod
def setup_clients(cls): def setup_clients(cls):
@ -55,3 +60,8 @@ class BaseShipyardTest(test.BaseTestCase):
CONF.shipyard.catalog_type, CONF.shipyard.catalog_type,
CONF.identity.region, CONF.identity.region,
CONF.shipyard.endpoint_type) CONF.shipyard.endpoint_type)
cls.shipyard_log_retrieval_client = LogRetrievalClient(
cls.auth_provider,
CONF.shipyard.catalog_type,
CONF.identity.region,
CONF.shipyard.endpoint_type)

View File

@ -14,7 +14,6 @@
# under the License. # under the License.
# #
from airship_tempest_plugin.services.shipyard.json.actions_client import ActionsClient
from airship_tempest_plugin.tests.api.shipyard import base from airship_tempest_plugin.tests.api.shipyard import base
from tempest import config from tempest import config
@ -23,6 +22,7 @@ from patrole_tempest_plugin import rbac_utils
CONF = config.CONF CONF = config.CONF
class BaseShipyardRbacTest(rbac_utils.RbacUtilsMixin, class BaseShipyardRbacTest(rbac_utils.RbacUtilsMixin,
base.BaseShipyardTest): base.BaseShipyardTest):
"""Base class for Shipyard RBAC tests.""" """Base class for Shipyard RBAC tests."""

View File

@ -18,18 +18,78 @@ from airship_tempest_plugin.tests.api.shipyard.rbac import rbac_base
from patrole_tempest_plugin import rbac_rule_validation from patrole_tempest_plugin import rbac_rule_validation
from tempest.common import utils
from tempest.lib import decorators from tempest.lib import decorators
from tempest.lib.common.utils import data_utils from tempest.lib import exceptions
from tempest.lib.common.utils import test_utils
from tempest.api.identity import base
class ActionsRbacTest(rbac_base.BaseShipyardRbacTest): class ActionsRbacTest(rbac_base.BaseShipyardRbacTest):
@rbac_rule_validation.action(service="shipyard", @rbac_rule_validation.action(
rules=["get_actions"]) service="shipyard",
rules=["workflow_orchestrator:list_actions"])
@decorators.idempotent_id('183dd007-8a97-4070-afc3-9318401ebad7') @decorators.idempotent_id('183dd007-8a97-4070-afc3-9318401ebad7')
def test_get_actions(self): def test_list_actions(self):
with self.rbac_utils.override_role(self): with self.rbac_utils.override_role(self):
self.shipyard_actions_client.get_actions() self.shipyard_actions_client.list_actions()
@rbac_rule_validation.action(
service="shipyard",
rules=["workflow_orchestrator:create_action"])
@decorators.idempotent_id('fff43c6f-b6ed-44dd-b47b-02c45d7bdb8c')
def test_create_action(self):
with self.rbac_utils.override_role(self):
# As this is a RBAC test, we only care about whether the role has
# permission or not. Role permission is checked prior to validating
# the post body, therefore we will ignore a BadRequest exception
try:
self.shipyard_actions_client.create_action()
except exceptions.BadRequest:
pass
@rbac_rule_validation.action(
service="shipyard",
rules=["workflow_orchestrator:get_action"])
@decorators.idempotent_id('68e2f10f-0676-41bb-8f47-bc695e1aa536')
def test_get_action(self):
with self.rbac_utils.override_role(self):
# As this is a RBAC test, we only care about whether the role has
# permission or not. Role permission is checked prior to validating
# the post body, therefore we will ignore a NotFound exception
try:
self.shipyard_actions_client.get_action()
except exceptions.NotFound:
pass
''' NEEDS REWORK AS SHIPYARD NOT DOING POLICY ENFORCEMENT FIRST
@rbac_rule_validation.action(
service="shipyard",
rules=["workflow_orchestrator:get_action_validation"])
@decorators.idempotent_id('a5156dcd-2674-4295-aa6a-d8db1bd4cf4b')
def test_get_action_validation(self):
with self.rbac_utils.override_role(self):
self.shipyard_actions_client.get_action_validation()
'''
@rbac_rule_validation.action(
service="shipyard",
rules=["workflow_orchestrator:get_action_step"])
@decorators.idempotent_id('6243d2ff-f88e-41cf-8169-140a551834a4')
def test_get_action_step(self):
with self.rbac_utils.override_role(self):
# As this is a RBAC test, we only care about whether the role has
# permission or not. Role permission is checked prior to validating
# the post body, therefore we will ignore a NotFound exception
try:
self.shipyard_actions_client.get_action_step()
except exceptions.NotFound:
pass
''' NEEDS REWORK AS SHIPYARD NOT DOING POLICY ENFORCEMENT FIRST
@rbac_rule_validation.action(
service="shipyard",
rules=["workflow_orchestrator:invoke_action_control"])
@decorators.idempotent_id('4f6b6564-ff1d-463a-aee8-ed2d51e2a286')
def test_invoke_action_control(self):
with self.rbac_utils.override_role(self):
self.shipyard_actions_client.invoke_action_control()
'''

View File

@ -18,18 +18,30 @@ from airship_tempest_plugin.tests.api.shipyard.rbac import rbac_base
from patrole_tempest_plugin import rbac_rule_validation from patrole_tempest_plugin import rbac_rule_validation
from tempest.common import utils
from tempest.lib import decorators from tempest.lib import decorators
from tempest.lib.common.utils import data_utils from tempest.lib import exceptions
from tempest.lib.common.utils import test_utils
from tempest.api.identity import base
class AirflowMonitoringRbacTest(rbac_base.BaseShipyardRbacTest): class AirflowMonitoringRbacTest(rbac_base.BaseShipyardRbacTest):
@rbac_rule_validation.action(service="shipyard", @rbac_rule_validation.action(
rules=["get_configdocs"]) service="shipyard",
@decorators.idempotent_id('0ab53b15-bce9-494f-9a11-34dd2c44d699') rules=["workflow_orchestrator:list_workflows"])
def test_get_workflows(self): @decorators.idempotent_id('fc75a269-04cb-4a8d-a627-907f72081b8a')
def test_list_workflows(self):
with self.rbac_utils.override_role(self): with self.rbac_utils.override_role(self):
self.shipyard_airflow_monitoring_client.get_workflows() self.shipyard_airflow_monitoring_client.list_workflows()
@rbac_rule_validation.action(
service="shipyard",
rules=["workflow_orchestrator:get_workflow"])
@decorators.idempotent_id('1679c5fa-571a-4af8-8f14-ca0c0a49761b')
def test_get_workflow(self):
with self.rbac_utils.override_role(self):
# As this is a RBAC test, we only care about whether the role has
# permission or not. Role permission is checked prior to validating
# the post body, therefore we will ignore a BadRequest exception
try:
self.shipyard_airflow_monitoring_client.get_workflow()
except exceptions.BadRequest:
pass

View File

@ -18,71 +18,81 @@ from airship_tempest_plugin.tests.api.shipyard.rbac import rbac_base
from patrole_tempest_plugin import rbac_rule_validation from patrole_tempest_plugin import rbac_rule_validation
from tempest.common import utils
from tempest.lib import decorators from tempest.lib import decorators
from tempest.lib import exceptions from tempest.lib import exceptions
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.api.identity import base
class DocumentStagingRbacTest(rbac_base.BaseShipyardRbacTest): class DocumentStagingRbacTest(rbac_base.BaseShipyardRbacTest):
@rbac_rule_validation.action(service="shipyard", @rbac_rule_validation.action(
rules=["get_configdocs"]) service="shipyard",
rules=["workflow_orchestrator:get_configdocs_status"])
@decorators.idempotent_id('0ab53b15-bce9-494f-9a11-34dd2c44d699') @decorators.idempotent_id('0ab53b15-bce9-494f-9a11-34dd2c44d699')
def test_get_configdocs(self): def test_get_configdocs_status(self):
with self.rbac_utils.override_role(self):
self.shipyard_document_staging_client.get_configdocs()
@rbac_rule_validation.action(service="shipyard",
rules=["post_configdocs"])
@decorators.idempotent_id('1a0daf92-9dba-470c-a317-66b41c0b3df7')
def test_post_configdocs(self):
with self.rbac_utils.override_role(self): with self.rbac_utils.override_role(self):
# As this is a RBAC test, we only care about whether the role has # As this is a RBAC test, we only care about whether the role has
# permission or not. Role permission is checked prior to validating # permission or not. Role permission is checked prior to validating
# the post body, therefore we will ignore a BadRequest exception # the request body, therefore we will ignore a ValueError exception
try: try:
self.shipyard_document_staging_client.post_configdocs() self.shipyard_document_staging_client.get_configdocs_status()
except exceptions.BadRequest: except ValueError:
pass pass
@rbac_rule_validation.action(service="shipyard", @rbac_rule_validation.action(
rules=["get_configdocs_within_collection"]) service="shipyard",
@decorators.idempotent_id('d64cfa75-3bbe-4688-8849-db5a54ce98ea') rules=["workflow_orchestrator:create_configdocs"])
def test_get_configdocs_within_collection(self): @decorators.idempotent_id('1a0daf92-9dba-470c-a317-66b41c0b3df7')
def test_create_configdocs(self):
with self.rbac_utils.override_role(self): with self.rbac_utils.override_role(self):
# As this is a RBAC test, we only care about whether the role has # As this is a RBAC test, we only care about whether the role has
# permission or not. Role permission is checked prior to validating # permission or not. Role permission is checked prior to validating
# the post body, therefore we will ignore a NotFound exception # the request body, therefore we will ignore a BadRequest exception
# and Conflict exception
try: try:
self.shipyard_document_staging_client.get_configdocs_within_collection() self.shipyard_document_staging_client.create_configdocs()
except (exceptions.BadRequest, exceptions.Conflict):
pass
@rbac_rule_validation.action(
service="shipyard",
rules=["workflow_orchestrator:get_configdocs"])
@decorators.idempotent_id('d64cfa75-3bbe-4688-8849-db5a54ce98ea')
def test_get_configdocs(self):
with self.rbac_utils.override_role(self):
# As this is a RBAC test, we only care about whether the role has
# permission or not. Role permission is checked prior to validating
# the request body, therefore we will ignore a NotFound exception
try:
self.shipyard_document_staging_client.get_configdocs()
except exceptions.NotFound: except exceptions.NotFound:
pass pass
@rbac_rule_validation.action(service="shipyard", @rbac_rule_validation.action(
rules=["get_renderedconfigdocs"]) service="shipyard",
@decorators.idempotent_id('0ab53b15-bce9-494f-9a11-34dd2c44d699') rules=["workflow_orchestrator:get_renderedconfigdocs"])
@decorators.idempotent_id('76e81d8d-4e06-42f8-9c9d-082020674994')
def test_get_renderedconfigdocs(self): def test_get_renderedconfigdocs(self):
with self.rbac_utils.override_role(self): with self.rbac_utils.override_role(self):
# As this is a RBAC test, we only care about whether the role has # As this is a RBAC test, we only care about whether the role has
# permission or not. Role permission is checked prior to validating # permission or not. Role permission is checked prior to validating
# the post body, therefore we will ignore a NotFound exception # the request body, therefore we will ignore a NotFound exception
# and ServerFault exception
try: try:
self.shipyard_document_staging_client.get_renderedconfigdocs() self.shipyard_document_staging_client.get_renderedconfigdocs()
except exceptions.NotFound: except (exceptions.NotFound, exceptions.ServerFault):
pass pass
@rbac_rule_validation.action(service="shipyard", @rbac_rule_validation.action(
rules=["post_commitconfigdocs"]) service="shipyard",
rules=["workflow_orchestrator:commit_configdocs"])
@decorators.idempotent_id('200d1cbf-ca11-4b92-9cfd-6cd2a90bc919') @decorators.idempotent_id('200d1cbf-ca11-4b92-9cfd-6cd2a90bc919')
def test_post_commitconfigdocs(self): def test_commit_configdocs(self):
with self.rbac_utils.override_role(self): with self.rbac_utils.override_role(self):
# As this is a RBAC test, we only care about whether the role has # As this is a RBAC test, we only care about whether the role has
# permission or not. Role permission is checked prior to validating # permission or not. Role permission is checked prior to validating
# the post body, therefore we will ignore a Conflict exception # the request body, therefore we will ignore a Conflict exception
# and BadRequest exception
try: try:
self.shipyard_document_staging_client.post_commitconfigdocs() self.shipyard_document_staging_client.commit_configdocs()
except exceptions.Conflict: except (exceptions.Conflict, exceptions.BadRequest):
pass pass

View File

@ -0,0 +1,39 @@
# Copyright 2018 AT&T Corp
# 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 airship_tempest_plugin.tests.api.shipyard.rbac import rbac_base
from patrole_tempest_plugin import rbac_rule_validation
from tempest.lib import decorators
from tempest.lib import exceptions
class LogRetrievalRbacTest(rbac_base.BaseShipyardRbacTest):
@rbac_rule_validation.action(
service="shipyard",
rules=["workflow_orchestrator:get_action_step_logs"])
@decorators.idempotent_id('5fd2c572-a226-482d-bdce-70d3ffcd7495')
def test_get_action_step_logs(self):
with self.rbac_utils.override_role(self):
# As this is a RBAC test, we only care about whether the role has
# permission or not. Role permission is checked prior to validating
# the post body, therefore we will ignore a BadRequest exception
try:
self.shipyard_log_retrieval_client.get_action_step_logs()
except exceptions.BadRequest:
pass

9
requirements.txt Normal file
View File

@ -0,0 +1,9 @@
# The order of packages is significant, because pip processes them in the order
# of appearance. Changing the order has an impact on the overall integration
# process, which may cause wedges in the gate later.
pbr!=2.1.0,>=2.0.0 # Apache-2.0
oslo.log>=3.36.0 # Apache-2.0
oslo.config>=5.2.0 # Apache-2.0
oslo.policy>=1.30.0 # Apache-2.0
tempest>=17.1.0 # Apache-2.0
stevedore>=1.20.0 # Apache-2.0

10
test-requirements.txt Normal file
View File

@ -0,0 +1,10 @@
# The order of packages is significant, because pip processes them in the order
# of appearance. Changing the order has an impact on the overall integration
# process, which may cause wedges in the gate later.
hacking>=1.1.0,<1.2.0 # Apache-2.0
fixtures>=3.0.0 # Apache-2.0/BSD
mock>=2.0.0 # BSD
coverage!=4.4,>=4.0 # Apache-2.0
nose>=1.3.7 # LGPL
nosexcover>=1.0.10 # BSD
oslotest>=3.2.0 # Apache-2.0

104
tox.ini Normal file
View File

@ -0,0 +1,104 @@
[tox]
minversion = 1.6
envlist = pep8,py35,py27
skipsdist = True
[testenv]
usedevelop = True
install_command = pip install -c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt} {opts} {packages}
setenv =
VIRTUAL_ENV={envdir}
OS_TEST_PATH=./airship_tempest_plugin/tests/unit
LANGUAGE=en_US
LC_ALL=en_US.utf-8
PYTHONWARNINGS=default::DeprecationWarning
passenv = OS_STDOUT_CAPTURE OS_STDERR_CAPTURE OS_TEST_TIMEOUT OS_TEST_LOCK_PATH http_proxy HTTP_PROXY https_proxy HTTPS_PROXY no_proxy NO_PROXY
whitelist_externals = find
deps = -r{toxinidir}/requirements.txt
-r{toxinidir}/test-requirements.txt
commands =
find . -type f -name "*.pyc" -delete
stestr --test-path ./airship_tempest_plugin/tests/unit run {posargs}
[testenv:pep8]
basepython = python3
commands = flake8 {posargs}
check-uuid --package airship_tempest_plugin.tests.api
[testenv:uuidgen]
basepython = python3
commands = check-uuid --package airship_tempest_plugin.tests.api --fix
[testenv:venv]
basepython = python3
commands = {posargs}
[testenv:cover]
basepython = python3
commands = rm -rf *.pyc
rm -rf cover
rm -f .coverage
nosetests {posargs}
setenv = VIRTUAL_ENV={envdir}
NOSE_WITH_COVERAGE=1
NOSE_COVER_BRANCHES=1
NOSE_COVER_PACKAGE=airship_tempest_plugin
NOSE_COVER_HTML=1
NOSE_COVER_HTML_DIR={toxinidir}/cover
NOSE_WHERE=airship_tempest_plugin/tests/unit
whitelist_externals = nosetests
rm
[testenv:docs]
basepython = python3
deps =
-c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt}
-r{toxinidir}/requirements.txt
-r{toxinidir}/doc/requirements.txt
commands =
rm -rf doc/build
sphinx-build -W -b html doc/source doc/build/html
whitelist_externals = rm
[testenv:releasenotes]
basepython = python3
deps =
-c{env:UPPER_CONSTRAINTS_FILE:https://git.openstack.org/cgit/openstack/requirements/plain/upper-constraints.txt}
-r{toxinidir}/requirements.txt
-r{toxinidir}/doc/requirements.txt
commands =
rm -rf releasenotes/build
sphinx-build -a -E -W -d releasenotes/build/doctrees -b html releasenotes/source releasenotes/build/html
whitelist_externals = rm
[testenv:debug]
basepython = python3
commands = oslo_debug_helper -t airship_tempest_plugin/tests {posargs}
[flake8]
# [H106] Don't put vim configuration in source files.
# [H203] Use assertIs(Not)None to check for None.
# [H204] Use assert(Not)Equal to check for equality.
# [H205] Use assert(Greater|Less)(Equal) for comparison.
# [H210] Require 'autospec', 'spec', or 'spec_set' in mock.patch/mock.patch.object calls
# [H904] Delay string interpolations at logging calls.
enable-extensions = H106,H203,H204,H205,H210,H904
show-source = True
# E123, E125 skipped as they are invalid PEP-8.
#
# H405 is another one that is good as a guideline, but sometimes
# multiline doc strings just don't have a natural summary
# line. Rejecting code for this reason is wrong.
ignore = E123,E125,H405
builtins = _
exclude=.venv,.git,.tox,dist,doc,*lib/python*,*egg,build
[hacking]
local-check-factory = airship_tempest_plugin.hacking.checks.factory
[testenv:lower-constraints]
basepython = python3
deps =
-c{toxinidir}/lower-constraints.txt
-r{toxinidir}/test-requirements.txt
-r{toxinidir}/requirements.txt