Enhance validation decorator with error code
Rather than wrapping each API call in a try/except block in the event
that a 404 rather than a 403 is thrown due to lack of permissions, the
decorator rbac_rule_validation.action is enhanced with the a new kwarg
called 'expected_error_code'.
The decorator now handles any error code that is passed in, and raises
the Forbidden exception even if the error code is not 403.
This commit also refactors the tests that use the try/except block to
handle non-403 error codes.
Change-Id: I9d263e21110fb9f988a6d4d781bf54e81f485f14
Implements: blueprint enhance-validation-decorator-with-error-code
diff --git a/patrole_tempest_plugin/rbac_exceptions.py b/patrole_tempest_plugin/rbac_exceptions.py
index 967342b..aa3135e 100644
--- a/patrole_tempest_plugin/rbac_exceptions.py
+++ b/patrole_tempest_plugin/rbac_exceptions.py
@@ -34,3 +34,7 @@
class RbacParsingException (exceptions.TempestException):
message = "Attempted to test an invalid policy file or action"
+
+
+class RbacInvalidErrorCode (exceptions.TempestException):
+ message = "Unsupported error code passed in test"
diff --git a/patrole_tempest_plugin/rbac_rule_validation.py b/patrole_tempest_plugin/rbac_rule_validation.py
index a6490e7..33ee666 100644
--- a/patrole_tempest_plugin/rbac_rule_validation.py
+++ b/patrole_tempest_plugin/rbac_rule_validation.py
@@ -26,7 +26,7 @@
LOG = logging.getLogger(__name__)
-def action(service, rule):
+def action(service, rule, expected_error_code=403):
def decorator(func):
def wrapper(*args, **kwargs):
try:
@@ -43,6 +43,8 @@
authority = rbac_auth.RbacAuthority(tenant_id, user_id, service)
allowed = authority.get_permission(rule, CONF.rbac.rbac_test_role)
+ expected_exception, irregular_msg = _get_exception_type(
+ expected_error_code)
try:
func(*args)
@@ -52,7 +54,7 @@
raise exceptions.NotFound(
"%s RbacInvalidService was: %s" %
(msg, e))
- except exceptions.Forbidden as e:
+ except expected_exception as e:
if allowed:
msg = ("Role %s was not allowed to perform %s." %
(CONF.rbac.rbac_test_role, rule))
@@ -60,6 +62,8 @@
raise exceptions.Forbidden(
"%s exception was: %s" %
(msg, e))
+ if irregular_msg:
+ LOG.warning(irregular_msg.format(rule, service))
except rbac_exceptions.RbacActionFailed as e:
if allowed:
msg = ("Role %s was not allowed to perform %s." %
@@ -80,3 +84,23 @@
switchToRbacRole=False)
return wrapper
return decorator
+
+
+def _get_exception_type(expected_error_code):
+ expected_exception = None
+ irregular_msg = None
+ supported_error_codes = [403, 404]
+ if expected_error_code == 403:
+ expected_exception = exceptions.Forbidden
+ elif expected_error_code == 404:
+ expected_exception = exceptions.NotFound
+ irregular_msg = ("NotFound exception was caught for policy action "
+ "{0}. The service {1} throws a 404 instead of a 403, "
+ "which is irregular.")
+ else:
+ msg = ("Please pass an expected error code. Currently "
+ "supported codes: {0}".format(str(supported_error_codes)))
+ LOG.error(msg)
+ raise rbac_exceptions.RbacInvalidErrorCode()
+
+ return expected_exception, irregular_msg
diff --git a/patrole_tempest_plugin/tests/api/image/v2/test_images_member_rbac.py b/patrole_tempest_plugin/tests/api/image/v2/test_images_member_rbac.py
index bbd5a79..7d99d55 100644
--- a/patrole_tempest_plugin/tests/api/image/v2/test_images_member_rbac.py
+++ b/patrole_tempest_plugin/tests/api/image/v2/test_images_member_rbac.py
@@ -16,9 +16,7 @@
from oslo_log import log as logging
from tempest import config
from tempest.lib import decorators
-from tempest.lib import exceptions
-from patrole_tempest_plugin import rbac_exceptions
from patrole_tempest_plugin import rbac_rule_validation
from patrole_tempest_plugin.tests.api.image import rbac_base as base
@@ -79,7 +77,8 @@
self.alt_tenant_id)
@rbac_rule_validation.action(service="glance",
- rule="get_member")
+ rule="get_member",
+ expected_error_code=404)
@decorators.idempotent_id('c01fd308-6484-11e6-881e-080027d0d606')
def test_show_image_member(self):
@@ -87,24 +86,16 @@
RBAC test for the glance get_member policy
"""
- try:
- image_id = self.create_image()['id']
- self.image_member_client.create_image_member(
- image_id,
- member=self.alt_tenant_id)
+ image_id = self.create_image()['id']
+ self.image_member_client.create_image_member(
+ image_id,
+ member=self.alt_tenant_id)
- # Toggle role and get image member
- self.rbac_utils.switch_role(self, switchToRbacRole=True)
- self.image_member_client.show_image_member(
- image_id,
- self.alt_tenant_id)
- except exceptions.NotFound as e:
- '''If the role doesn't have access to an image, a 404 exception is
- thrown when the roles tries to show an image member'''
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the image and tries to show "
- "image members")
- raise rbac_exceptions.RbacActionFailed(e)
+ # Toggle role and get image member
+ self.rbac_utils.switch_role(self, switchToRbacRole=True)
+ self.image_member_client.show_image_member(
+ image_id,
+ self.alt_tenant_id)
@rbac_rule_validation.action(service="glance",
rule="modify_member")
diff --git a/patrole_tempest_plugin/tests/api/network/test_floating_ips_rbac.py b/patrole_tempest_plugin/tests/api/network/test_floating_ips_rbac.py
index d186f38..bd562ec 100644
--- a/patrole_tempest_plugin/tests/api/network/test_floating_ips_rbac.py
+++ b/patrole_tempest_plugin/tests/api/network/test_floating_ips_rbac.py
@@ -19,9 +19,7 @@
from tempest import config
from tempest.lib.common.utils import test_utils
from tempest.lib import decorators
-from tempest.lib import exceptions
-from patrole_tempest_plugin import rbac_exceptions
from patrole_tempest_plugin import rbac_rule_validation
from patrole_tempest_plugin.tests.api.network import rbac_base as base
@@ -108,7 +106,9 @@
self.floating_ips_client.update_floatingip(
floating_ip['id'], port_id=None)
- @rbac_rule_validation.action(service="neutron", rule="get_floatingip")
+ @rbac_rule_validation.action(service="neutron",
+ rule="get_floatingip",
+ expected_error_code=404)
@decorators.idempotent_id('f8846fd0-c976-48fe-a148-105303931b32')
def test_show_floating_ip(self):
"""Show floating IP.
@@ -117,18 +117,12 @@
"""
floating_ip = self._create_floatingip()
self.rbac_utils.switch_role(self, switchToRbacRole=True)
-
- try:
- # Show floating IP
- self.floating_ips_client.show_floatingip(floating_ip['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ # Show floating IP
+ self.floating_ips_client.show_floatingip(floating_ip['id'])
@rbac_rule_validation.action(service="neutron",
- rule="delete_floatingip")
+ rule="delete_floatingip",
+ expected_error_code=404)
@decorators.idempotent_id('2611b068-30d4-4241-a78f-1b801a14db7e')
def test_delete_floating_ip(self):
"""Delete floating IP.
@@ -137,13 +131,5 @@
"""
floating_ip = self._create_floatingip()
self.rbac_utils.switch_role(self, switchToRbacRole=True)
-
- try:
- # Delete the floating IP
- self.floating_ips_client.delete_floatingip(floating_ip['id'])
-
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ # Delete the floating IP
+ self.floating_ips_client.delete_floatingip(floating_ip['id'])
diff --git a/patrole_tempest_plugin/tests/api/network/test_metering_label_rules_rbac.py b/patrole_tempest_plugin/tests/api/network/test_metering_label_rules_rbac.py
index 21c2f96..10a20f4 100644
--- a/patrole_tempest_plugin/tests/api/network/test_metering_label_rules_rbac.py
+++ b/patrole_tempest_plugin/tests/api/network/test_metering_label_rules_rbac.py
@@ -18,10 +18,8 @@
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import decorators
-from tempest.lib import exceptions
from tempest import test
-from patrole_tempest_plugin import rbac_exceptions
from patrole_tempest_plugin import rbac_rule_validation
from patrole_tempest_plugin.tests.api.network import rbac_base as base
@@ -81,7 +79,8 @@
self._create_metering_label_rule(self.label)
@rbac_rule_validation.action(service="neutron",
- rule="get_metering_label_rule")
+ rule="get_metering_label_rule",
+ expected_error_code=404)
@decorators.idempotent_id('e21b40c3-d44d-412f-84ea-836ca8603bcb')
def test_show_metering_label_rule(self):
"""Show metering label rule.
@@ -90,17 +89,12 @@
"""
label_rule = self._create_metering_label_rule(self.label)
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.metering_label_rules_client.show_metering_label_rule(
- label_rule['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.metering_label_rules_client.show_metering_label_rule(
+ label_rule['id'])
@rbac_rule_validation.action(service="neutron",
- rule="delete_metering_label_rule")
+ rule="delete_metering_label_rule",
+ expected_error_code=404)
@decorators.idempotent_id('e3adc88c-05c0-43a7-8e32-63947ae4890e')
def test_delete_metering_label_rule(self):
"""Delete metering label rule.
@@ -109,11 +103,5 @@
"""
label_rule = self._create_metering_label_rule(self.label)
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.metering_label_rules_client.delete_metering_label_rule(
- label_rule['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.metering_label_rules_client.delete_metering_label_rule(
+ label_rule['id'])
diff --git a/patrole_tempest_plugin/tests/api/network/test_metering_labels_rbac.py b/patrole_tempest_plugin/tests/api/network/test_metering_labels_rbac.py
index 09e298c..eb17df4 100644
--- a/patrole_tempest_plugin/tests/api/network/test_metering_labels_rbac.py
+++ b/patrole_tempest_plugin/tests/api/network/test_metering_labels_rbac.py
@@ -17,10 +17,8 @@
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import decorators
-from tempest.lib import exceptions
from tempest import test
-from patrole_tempest_plugin import rbac_exceptions
from patrole_tempest_plugin import rbac_rule_validation
from patrole_tempest_plugin.tests.api.network import rbac_base as base
@@ -63,7 +61,8 @@
self._create_metering_label()
@rbac_rule_validation.action(service="neutron",
- rule="get_metering_label")
+ rule="get_metering_label",
+ expected_error_code=404)
@decorators.idempotent_id('c57f6636-c702-4755-8eac-5e73bc1f7d14')
def test_show_metering_label(self):
"""Show metering label.
@@ -72,16 +71,11 @@
"""
label = self._create_metering_label()
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.metering_labels_client.show_metering_label(label['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.metering_labels_client.show_metering_label(label['id'])
@rbac_rule_validation.action(service="neutron",
- rule="delete_metering_label")
+ rule="delete_metering_label",
+ expected_error_code=404)
@decorators.idempotent_id('1621ccfe-2e3f-4d16-98aa-b620f9d00404')
def test_delete_metering_label(self):
"""Delete metering label.
@@ -90,10 +84,4 @@
"""
label = self._create_metering_label()
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.metering_labels_client.delete_metering_label(label['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.metering_labels_client.delete_metering_label(label['id'])
diff --git a/patrole_tempest_plugin/tests/api/network/test_ports_rbac.py b/patrole_tempest_plugin/tests/api/network/test_ports_rbac.py
index 518c71b..093459e 100644
--- a/patrole_tempest_plugin/tests/api/network/test_ports_rbac.py
+++ b/patrole_tempest_plugin/tests/api/network/test_ports_rbac.py
@@ -21,9 +21,7 @@
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import decorators
-from tempest.lib import exceptions
-from patrole_tempest_plugin import rbac_exceptions
from patrole_tempest_plugin import rbac_rule_validation
from patrole_tempest_plugin.tests.api.network import rbac_base as base
@@ -143,61 +141,43 @@
self.rbac_utils.switch_role(self, switchToRbacRole=True)
self._create_port(**post_body)
- @rbac_rule_validation.action(service="neutron", rule="get_port")
+ @rbac_rule_validation.action(service="neutron",
+ rule="get_port",
+ expected_error_code=404)
@decorators.idempotent_id('a9d41cb8-78a2-4b97-985c-44e4064416f4')
def test_show_port(self):
-
- try:
- self.rbac_utils.switch_role(self, switchToRbacRole=True)
-
- self.ports_client.show_port(self.admin_port['id'])
-
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.rbac_utils.switch_role(self, switchToRbacRole=True)
+ self.ports_client.show_port(self.admin_port['id'])
@rbac_rule_validation.action(service="neutron",
- rule="get_port:binding:vif_type")
+ rule="get_port:binding:vif_type",
+ expected_error_code=404)
@decorators.idempotent_id('125aff0b-8fed-4f8e-8410-338616594b06')
def test_show_port_binding_vif_type(self):
# Verify specific fields of a port
fields = ['binding:vif_type']
- try:
- self.rbac_utils.switch_role(self, switchToRbacRole=True)
- self.ports_client.show_port(self.admin_port['id'],
- fields=fields)
-
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.rbac_utils.switch_role(self, switchToRbacRole=True)
+ self.ports_client.show_port(self.admin_port['id'],
+ fields=fields)
@rbac_rule_validation.action(service="neutron",
- rule="get_port:binding:vif_details")
+ rule="get_port:binding:vif_details",
+ expected_error_code=404)
@decorators.idempotent_id('e42bfd77-fcce-45ee-9728-3424300f0d6f')
def test_show_port_binding_vif_details(self):
# Verify specific fields of a port
fields = ['binding:vif_details']
- try:
- self.rbac_utils.switch_role(self, switchToRbacRole=True)
- self.ports_client.show_port(self.admin_port['id'],
- fields=fields)
-
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.rbac_utils.switch_role(self, switchToRbacRole=True)
+ self.ports_client.show_port(self.admin_port['id'],
+ fields=fields)
@rbac_rule_validation.action(service="neutron",
- rule="get_port:binding:host_id")
+ rule="get_port:binding:host_id",
+ expected_error_code=404)
@decorators.idempotent_id('8e61bcdc-6f81-443c-833e-44410266551e')
def test_show_port_binding_host_id(self):
@@ -207,19 +187,13 @@
'binding:host_id': data_utils.rand_name('host-id')}
port = self._create_port(**post_body)
- try:
- self.rbac_utils.switch_role(self, switchToRbacRole=True)
- self.ports_client.show_port(port['id'],
- fields=fields)
-
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.rbac_utils.switch_role(self, switchToRbacRole=True)
+ self.ports_client.show_port(port['id'],
+ fields=fields)
@rbac_rule_validation.action(service="neutron",
- rule="get_port:binding:profile")
+ rule="get_port:binding:profile",
+ expected_error_code=404)
@decorators.idempotent_id('d497cea9-c4ad-42e0-acc9-8d257d6b01fc')
def test_show_port_binding_profile(self):
@@ -230,16 +204,9 @@
'binding:profile': binding_profile}
port = self._create_port(**post_body)
- try:
- self.rbac_utils.switch_role(self, switchToRbacRole=True)
- self.ports_client.show_port(port['id'],
- fields=fields)
-
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.rbac_utils.switch_role(self, switchToRbacRole=True)
+ self.ports_client.show_port(port['id'],
+ fields=fields)
@rbac_rule_validation.action(service="neutron",
rule="update_port")
@@ -337,17 +304,11 @@
allowed_address_pairs=address_pairs)
@rbac_rule_validation.action(service="neutron",
- rule="delete_port")
+ rule="delete_port",
+ expected_error_code=404)
@decorators.idempotent_id('1cf8e582-bc09-46cb-b32a-82bf991ad56f')
def test_delete_port(self):
- try:
- port = self._create_port(network_id=self.admin_network['id'])
- self.rbac_utils.switch_role(self, switchToRbacRole=True)
- self.ports_client.delete_port(port['id'])
-
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ port = self._create_port(network_id=self.admin_network['id'])
+ self.rbac_utils.switch_role(self, switchToRbacRole=True)
+ self.ports_client.delete_port(port['id'])
diff --git a/patrole_tempest_plugin/tests/api/network/test_routers_rbac.py b/patrole_tempest_plugin/tests/api/network/test_routers_rbac.py
index a227e5c..203384e 100644
--- a/patrole_tempest_plugin/tests/api/network/test_routers_rbac.py
+++ b/patrole_tempest_plugin/tests/api/network/test_routers_rbac.py
@@ -21,10 +21,8 @@
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import decorators
-from tempest.lib import exceptions
from tempest import test
-from patrole_tempest_plugin import rbac_exceptions
from patrole_tempest_plugin import rbac_rule_validation
from patrole_tempest_plugin.tests.api.network import rbac_base as base
@@ -111,7 +109,9 @@
self.addCleanup(self.routers_client.delete_router,
router['router']['id'])
- @rbac_rule_validation.action(service="neutron", rule="get_router")
+ @rbac_rule_validation.action(service="neutron",
+ rule="get_router",
+ expected_error_code=404)
@decorators.idempotent_id('bfbdbcff-f115-4d3e-8cd5-6ada33fd0e21')
def test_show_router(self):
"""Get Router
@@ -119,13 +119,7 @@
RBAC test for the neutron get_router policy
"""
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.routers_client.show_router(self.admin_router['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.routers_client.show_router(self.admin_router['id'])
@rbac_rule_validation.action(
service="neutron", rule="update_router")
@@ -211,7 +205,8 @@
external_gateway_info=None)
@rbac_rule_validation.action(service="neutron",
- rule="delete_router")
+ rule="delete_router",
+ expected_error_code=404)
@decorators.idempotent_id('c0634dd5-0467-48f7-a4ae-1014d8edb2a7')
def test_delete_router(self):
"""Delete Router
@@ -220,16 +215,11 @@
"""
router = self.create_router()
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.routers_client.delete_router(router['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.routers_client.delete_router(router['id'])
@rbac_rule_validation.action(service="neutron",
- rule="add_router_interface")
+ rule="add_router_interface",
+ expected_error_code=404)
@decorators.idempotent_id('a0627778-d68d-4913-881b-e345360cca19')
def test_add_router_interfaces(self):
"""Add Router Interface
@@ -241,22 +231,17 @@
router = self.create_router()
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.routers_client.add_router_interface(
- router['id'], subnet_id=subnet['id'])
- self.addCleanup(
- test_utils.call_and_ignore_notfound_exc,
- self.routers_client.remove_router_interface,
- router['id'],
- subnet_id=subnet['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.routers_client.add_router_interface(
+ router['id'], subnet_id=subnet['id'])
+ self.addCleanup(
+ test_utils.call_and_ignore_notfound_exc,
+ self.routers_client.remove_router_interface,
+ router['id'],
+ subnet_id=subnet['id'])
@rbac_rule_validation.action(service="neutron",
- rule="remove_router_interface")
+ rule="remove_router_interface",
+ expected_error_code=404)
@decorators.idempotent_id('ff2593a4-2bff-4c27-97d3-dd3702b27dfb')
def test_remove_router_interfaces(self):
"""Remove Router Interface
@@ -276,12 +261,6 @@
subnet_id=subnet['id'])
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.routers_client.remove_router_interface(
- router['id'],
- subnet_id=subnet['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.routers_client.remove_router_interface(
+ router['id'],
+ subnet_id=subnet['id'])
diff --git a/patrole_tempest_plugin/tests/api/network/test_security_groups_rbac.py b/patrole_tempest_plugin/tests/api/network/test_security_groups_rbac.py
index cf76836..289c669 100644
--- a/patrole_tempest_plugin/tests/api/network/test_security_groups_rbac.py
+++ b/patrole_tempest_plugin/tests/api/network/test_security_groups_rbac.py
@@ -18,9 +18,7 @@
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import decorators
-from tempest.lib import exceptions
-from patrole_tempest_plugin import rbac_exceptions
from patrole_tempest_plugin import rbac_rule_validation
from patrole_tempest_plugin.tests.api.network import rbac_base as base
@@ -80,53 +78,40 @@
self._create_security_group()
@rbac_rule_validation.action(service="neutron",
- rule="get_security_group")
+ rule="get_security_group",
+ expected_error_code=404)
@decorators.idempotent_id('56335e77-aef2-4b54-86c7-7f772034b585')
def test_show_security_groups(self):
- try:
- self.rbac_utils.switch_role(self, switchToRbacRole=True)
- self.security_groups_client.show_security_group(
- self.secgroup['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.rbac_utils.switch_role(self, switchToRbacRole=True)
+ self.security_groups_client.show_security_group(
+ self.secgroup['id'])
@rbac_rule_validation.action(service="neutron",
- rule="delete_security_group")
+ rule="delete_security_group",
+ expected_error_code=404)
@decorators.idempotent_id('0b1330fd-dd28-40f3-ad73-966052e4b3de')
def test_delete_security_group(self):
# Create a security group
secgroup_id = self._create_security_group()['id']
+
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.security_groups_client.delete_security_group(secgroup_id)
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.security_groups_client.delete_security_group(secgroup_id)
@rbac_rule_validation.action(service="neutron",
- rule="update_security_group")
+ rule="update_security_group",
+ expected_error_code=404)
@decorators.idempotent_id('56c5e4dc-f8aa-11e6-bc64-92361f002671')
def test_update_security_group(self):
# Create a security group
secgroup_id = self._create_security_group()['id']
+
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.security_groups_client.update_security_group(
- secgroup_id,
- description="test description")
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.security_groups_client.update_security_group(
+ secgroup_id,
+ description="test description")
@rbac_rule_validation.action(service="neutron",
rule="get_security_groups")
@@ -145,38 +130,26 @@
self._create_security_group_rule()
@rbac_rule_validation.action(service="neutron",
- rule="delete_security_group_rule")
+ rule="delete_security_group_rule",
+ expected_error_code=404)
@decorators.idempotent_id('2262539e-b7d9-438c-acf9-a5ce0613be28')
def test_delete_security_group_rule(self):
sec_group_rule = self._create_security_group_rule()
-
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.security_group_rules_client.delete_security_group_rule(
- sec_group_rule['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.security_group_rules_client.delete_security_group_rule(
+ sec_group_rule['id'])
@rbac_rule_validation.action(service="neutron",
- rule="get_security_group_rule")
+ rule="get_security_group_rule",
+ expected_error_code=404)
@decorators.idempotent_id('84b4038c-261e-4a94-90d5-c885739ab0d5')
def test_show_security_group_rule(self):
sec_group_rule = self._create_security_group_rule()
-
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.security_group_rules_client.show_security_group_rule(
- sec_group_rule['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.security_group_rules_client.show_security_group_rule(
+ sec_group_rule['id'])
@rbac_rule_validation.action(service="neutron",
rule="get_security_group_rules")
diff --git a/patrole_tempest_plugin/tests/api/network/test_subnetpools_rbac.py b/patrole_tempest_plugin/tests/api/network/test_subnetpools_rbac.py
index 35ad335..3667212 100644
--- a/patrole_tempest_plugin/tests/api/network/test_subnetpools_rbac.py
+++ b/patrole_tempest_plugin/tests/api/network/test_subnetpools_rbac.py
@@ -18,10 +18,8 @@
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import decorators
-from tempest.lib import exceptions
from tempest import test
-from patrole_tempest_plugin import rbac_exceptions
from patrole_tempest_plugin import rbac_rule_validation
from patrole_tempest_plugin.tests.api.network import rbac_base as base
@@ -79,7 +77,8 @@
self._create_subnetpool(shared=True)
@rbac_rule_validation.action(service="neutron",
- rule="get_subnetpool")
+ rule="get_subnetpool",
+ expected_error_code=404)
@decorators.idempotent_id('4f5aee26-0507-4b6d-b44c-3128a25094d2')
def test_show_subnetpool(self):
"""Show subnetpool.
@@ -88,13 +87,7 @@
"""
subnetpool = self._create_subnetpool()
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.subnetpools_client.show_subnetpool(subnetpool['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.subnetpools_client.show_subnetpool(subnetpool['id'])
@rbac_rule_validation.action(service="neutron",
rule="update_subnetpool")
@@ -110,7 +103,8 @@
min_prefixlen=24)
@rbac_rule_validation.action(service="neutron",
- rule="delete_subnetpool")
+ rule="delete_subnetpool",
+ expected_error_code=404)
@decorators.idempotent_id('50f5944e-43e5-457b-ab50-fb48a73f0d3e')
def test_delete_subnetpool(self):
"""Delete subnetpool.
@@ -119,10 +113,4 @@
"""
subnetpool = self._create_subnetpool()
self.rbac_utils.switch_role(self, switchToRbacRole=True)
- try:
- self.subnetpools_client.delete_subnetpool(subnetpool['id'])
- except exceptions.NotFound as e:
- LOG.info("NotFound exception caught. Exception is thrown when "
- "role doesn't have access to the endpoint."
- "This is irregular and should be fixed.")
- raise rbac_exceptions.RbacActionFailed(e)
+ self.subnetpools_client.delete_subnetpool(subnetpool['id'])
diff --git a/patrole_tempest_plugin/tests/unit/test_rbac_rule_validation.py b/patrole_tempest_plugin/tests/unit/test_rbac_rule_validation.py
index eeb06cc..9fa0d98 100644
--- a/patrole_tempest_plugin/tests/unit/test_rbac_rule_validation.py
+++ b/patrole_tempest_plugin/tests/unit/test_rbac_rule_validation.py
@@ -19,10 +19,13 @@
from patrole_tempest_plugin import rbac_exceptions
from patrole_tempest_plugin import rbac_rule_validation as rbac_rv
+from tempest import config
from tempest.lib import exceptions
from tempest import test
from tempest.tests import base
+CONF = config.CONF
+
class RBACRuleValidationTest(base.TestCase):
@@ -36,6 +39,10 @@
self.mock_args.auth_provider.credentials.user_id = \
mock.sentinel.user_id
+ CONF.set_override('rbac_test_role', 'Member', group='rbac',
+ enforce_type=True)
+ self.addCleanup(CONF.clear_override, 'rbac_test_role', group='rbac')
+
@mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
def test_RBAC_rv_happy_path(self, mock_auth):
decorator = rbac_rv.action("", "")
@@ -44,27 +51,58 @@
wrapper((self.mock_args))
self.assertTrue(mock_function.called)
+ @mock.patch.object(rbac_rv, 'LOG', autospec=True)
@mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
- def test_RBAC_rv_forbidden(self, mock_auth):
- decorator = rbac_rv.action("", "")
+ def test_RBAC_rv_forbidden(self, mock_auth, mock_log):
+ decorator = rbac_rv.action(mock.sentinel.service, mock.sentinel.action)
mock_function = mock.Mock()
mock_function.side_effect = exceptions.Forbidden
wrapper = decorator(mock_function)
- self.assertRaises(exceptions.Forbidden, wrapper, self.mock_args)
+ e = self.assertRaises(exceptions.Forbidden, wrapper, self.mock_args)
+ self.assertIn(
+ "Role Member was not allowed to perform sentinel.action.",
+ e.__str__())
+ mock_log.error.assert_called_once_with("Role Member was not allowed to"
+ " perform sentinel.action.")
+ @mock.patch.object(rbac_rv, 'LOG', autospec=True)
@mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
- def test_RBAC_rv_rbac_action_failed(self, mock_auth):
- decorator = rbac_rv.action("", "")
+ def test_expect_not_found_but_raises_forbidden(self, mock_auth, mock_log):
+ decorator = rbac_rv.action(mock.sentinel.service,
+ mock.sentinel.action,
+ 404)
+ mock_function = mock.Mock()
+ mock_function.side_effect = exceptions.NotFound
+ wrapper = decorator(mock_function)
+
+ e = self.assertRaises(exceptions.Forbidden, wrapper, self.mock_args)
+ self.assertIn(
+ "Role Member was not allowed to perform sentinel.action.",
+ e.__str__())
+ mock_log.error.assert_called_once_with("Role Member was not allowed to"
+ " perform sentinel.action.")
+
+ @mock.patch.object(rbac_rv, 'LOG', autospec=True)
+ @mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
+ def test_RBAC_rv_rbac_action_failed(self, mock_auth, mock_log):
+ decorator = rbac_rv.action(mock.sentinel.service, mock.sentinel.action)
mock_function = mock.Mock()
mock_function.side_effect = rbac_exceptions.RbacActionFailed
-
wrapper = decorator(mock_function)
- self.assertRaises(exceptions.Forbidden, wrapper, self.mock_args)
+ e = self.assertRaises(exceptions.Forbidden, wrapper, self.mock_args)
+ self.assertIn(
+ "Role Member was not allowed to perform sentinel.action.",
+ e.__str__())
+
+ mock_log.error.assert_called_once_with("Role Member was not allowed to"
+ " perform sentinel.action.")
+
+ @mock.patch.object(rbac_rv, 'LOG', autospec=True)
@mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
- def test_RBAC_rv_not_allowed(self, mock_auth):
- decorator = rbac_rv.action("", "")
+ def test_RBAC_rv_not_allowed(self, mock_auth, mock_log):
+ decorator = rbac_rv.action(mock.sentinel.service, mock.sentinel.action)
mock_function = mock.Mock()
wrapper = decorator(mock_function)
@@ -73,8 +111,13 @@
mock_permission.get_permission.return_value = False
mock_auth.return_value = mock_permission
- self.assertRaises(rbac_exceptions.RbacOverPermission, wrapper,
- self.mock_args)
+ e = self.assertRaises(rbac_exceptions.RbacOverPermission, wrapper,
+ self.mock_args)
+ self.assertIn(("OverPermission: Role Member was allowed to perform "
+ "sentinel.action"), e.__str__())
+
+ mock_log.error.assert_called_once_with(
+ "Role Member was allowed to perform sentinel.action")
@mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
def test_RBAC_rv_forbidden_not_allowed(self, mock_auth):
@@ -90,6 +133,29 @@
self.assertIsNone(wrapper(self.mock_args))
+ @mock.patch.object(rbac_rv, 'LOG', autospec=True)
+ @mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
+ def test_expect_not_found_and_not_allowed(self, mock_auth, mock_log):
+ decorator = rbac_rv.action(mock.sentinel.service,
+ mock.sentinel.action,
+ 404)
+
+ mock_function = mock.Mock()
+ mock_function.side_effect = exceptions.NotFound
+ wrapper = decorator(mock_function)
+
+ mock_permission = mock.Mock()
+ mock_permission.get_permission.return_value = False
+ mock_auth.return_value = mock_permission
+
+ self.assertIsNone(wrapper(self.mock_args))
+
+ mock_log.warning.assert_called_once_with(
+ 'NotFound exception was caught for policy action sentinel.action. '
+ 'The service sentinel.service throws a 404 instead of a 403, '
+ 'which is irregular.')
+ mock_log.error.assert_not_called()
+
@mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
def test_RBAC_rv_rbac_action_failed_not_allowed(self, mock_auth):
decorator = rbac_rv.action("", "")
@@ -122,3 +188,37 @@
mock_rbac_policy_parser.RbacPolicyParser.assert_called_once_with(
mock.sentinel.tenant_id, mock.sentinel.user_id,
mock.sentinel.service)
+
+ @mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
+ def test_get_exception_type_404(self, mock_auth):
+ expected_exception = exceptions.NotFound
+ expected_irregular_msg = ("NotFound exception was caught for policy "
+ "action {0}. The service {1} throws a 404 "
+ "instead of a 403, which is irregular.")
+
+ actual_exception, actual_irregular_msg = \
+ rbac_rv._get_exception_type(404)
+
+ self.assertEqual(expected_exception, actual_exception)
+ self.assertEqual(expected_irregular_msg, actual_irregular_msg)
+
+ @mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
+ def test_get_exception_type_403(self, mock_auth):
+ expected_exception = exceptions.Forbidden
+ expected_irregular_msg = None
+
+ actual_exception, actual_irregular_msg = \
+ rbac_rv._get_exception_type(403)
+
+ self.assertEqual(expected_exception, actual_exception)
+ self.assertEqual(expected_irregular_msg, actual_irregular_msg)
+
+ @mock.patch.object(rbac_rv, 'LOG', autospec=True)
+ @mock.patch('patrole_tempest_plugin.rbac_auth.RbacAuthority')
+ def test_exception_thrown_when_type_is_not_int(self, mock_auth, mock_log):
+ self.assertRaises(rbac_exceptions.RbacInvalidErrorCode,
+ rbac_rv._get_exception_type, "403")
+
+ mock_log.error.assert_called_once_with("Please pass an expected error "
+ "code. Currently supported "
+ "codes: [403, 404]")