openstack/neutron
Code Issues Proposed changes

Merge "Add policy enforcer for "tags" service plugin" into stable/2023.2

Browse Source
This commit is contained in:
Zuul 2024-07-12 10:27:01 +00:00 committed by Gerrit Code Review
commit 528e6fd012
23 changed files with 1234 additions and 50 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
neutron/conf/policies/floatingip.py
View File

@ -19,6 +19,21 @@ from neutron.conf.policies import base
COLLECTION_PATH = '/floatingips'
RESOURCE_PATH = '/floatingips/{id}'
TAGS_PATH = RESOURCE_PATH + '/tags'
TAG_PATH = RESOURCE_PATH + '/tags/{tag_id}'
ACTION_GET_TAGS = [
{'method': 'GET', 'path': TAGS_PATH},
{'method': 'GET', 'path': TAG_PATH},
]
ACTION_PUT_TAGS = [
{'method': 'PUT', 'path': TAGS_PATH},
{'method': 'PUT', 'path': TAG_PATH},
]
ACTION_DELETE_TAGS = [
{'method': 'DELETE', 'path': TAGS_PATH},
{'method': 'DELETE', 'path': TAG_PATH},
]
DEPRECATION_REASON = (
"The Floating IP API now supports system scope and default roles.")
@ -79,6 +94,14 @@ rules = [
deprecated_reason=DEPRECATION_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='get_floatingips_tags',
check_str=base.ADMIN_OR_PROJECT_READER,
description='Get the floating IP tags',
operations=ACTION_GET_TAGS,
scope_types=['project'],
),
policy.DocumentedRuleDefault(
name='update_floatingip',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
@ -96,6 +119,14 @@ rules = [
deprecated_reason=DEPRECATION_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='update_floatingips_tags',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
description='Update the floating IP tags',
operations=ACTION_PUT_TAGS,
scope_types=['project'],
),
policy.DocumentedRuleDefault(
name='delete_floatingip',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
@ -113,6 +144,13 @@ rules = [
deprecated_reason=DEPRECATION_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='delete_floatingips_tags',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
description='Delete the floating IP tags',
operations=ACTION_DELETE_TAGS,
scope_types=['project'],
),
]

40
neutron/conf/policies/network.py
View File

@ -22,6 +22,8 @@ The network API now supports system scope and default roles.
COLLECTION_PATH = '/networks'
RESOURCE_PATH = '/networks/{id}'
TAGS_PATH = RESOURCE_PATH + '/tags'
TAG_PATH = RESOURCE_PATH + '/tags/{tag_id}'
ACTION_POST = [
{'method': 'POST', 'path': COLLECTION_PATH},
@ -36,6 +38,18 @@ ACTION_GET = [
{'method': 'GET', 'path': COLLECTION_PATH},
{'method': 'GET', 'path': RESOURCE_PATH},
]
ACTION_GET_TAGS = [
{'method': 'GET', 'path': TAGS_PATH},
{'method': 'GET', 'path': TAG_PATH},
]
ACTION_PUT_TAGS = [
{'method': 'PUT', 'path': TAGS_PATH},
{'method': 'PUT', 'path': TAG_PATH},
]
ACTION_DELETE_TAGS = [
{'method': 'DELETE', 'path': TAGS_PATH},
{'method': 'DELETE', 'path': TAG_PATH},
]
rules = [
@ -233,6 +247,18 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='get_networks_tags',
check_str=neutron_policy.policy_or(
base.ADMIN_OR_PROJECT_READER,
'rule:shared',
'rule:external',
neutron_policy.RULE_ADVSVC
),
scope_types=['project'],
description='Get the network tags',
operations=ACTION_GET_TAGS,
),
policy.DocumentedRuleDefault(
name='update_network',
@ -348,6 +374,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='update_networks_tags',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
scope_types=['project'],
description='Update the network tags',
operations=ACTION_PUT_TAGS,
),
policy.DocumentedRuleDefault(
name='delete_network',
@ -361,6 +394,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='delete_networks_tags',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
scope_types=['project'],
description='Delete the network tags',
operations=ACTION_DELETE_TAGS,
),
]

39
neutron/conf/policies/network_segment_range.py
View File

@ -25,6 +25,21 @@ The network segment range API now supports project scope and default roles.
COLLECTION_PATH = '/network_segment_ranges'
RESOURCE_PATH = '/network_segment_ranges/{id}'
TAGS_PATH = RESOURCE_PATH + '/tags'
TAG_PATH = RESOURCE_PATH + '/tags/{tag_id}'
ACTION_GET_TAGS = [
{'method': 'GET', 'path': TAGS_PATH},
{'method': 'GET', 'path': TAG_PATH},
]
ACTION_PUT_TAGS = [
{'method': 'PUT', 'path': TAGS_PATH},
{'method': 'PUT', 'path': TAG_PATH},
]
ACTION_DELETE_TAGS = [
{'method': 'DELETE', 'path': TAGS_PATH},
{'method': 'DELETE', 'path': TAG_PATH},
]
rules = [
@ -45,6 +60,7 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='get_network_segment_range',
check_str=base.ADMIN,
@ -66,6 +82,14 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='get_network_segment_ranges_tags',
check_str=base.ADMIN,
scope_types=['project'],
description='Get the network segment range tags',
operations=ACTION_GET_TAGS,
),
policy.DocumentedRuleDefault(
name='update_network_segment_range',
check_str=base.ADMIN,
@ -83,6 +107,14 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='update_network_segment_ranges_tags',
check_str=base.ADMIN,
scope_types=['project'],
description='Update the network segment range tags',
operations=ACTION_PUT_TAGS,
),
policy.DocumentedRuleDefault(
name='delete_network_segment_range',
check_str=base.ADMIN,
@ -100,6 +132,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='delete_network_segment_ranges_tags',
check_str=base.ADMIN,
scope_types=['project'],
description='Delete the network segment range tags',
operations=ACTION_DELETE_TAGS,
),
]

46
neutron/conf/policies/port.py
View File

@ -22,6 +22,8 @@ DEPRECATED_REASON = (
COLLECTION_PATH = '/ports'
RESOURCE_PATH = '/ports/{id}'
TAGS_PATH = RESOURCE_PATH + '/tags'
TAG_PATH = RESOURCE_PATH + '/tags/{tag_id}'
ACTION_POST = [
{'method': 'POST', 'path': COLLECTION_PATH},
@ -36,6 +38,18 @@ ACTION_GET = [
{'method': 'GET', 'path': COLLECTION_PATH},
{'method': 'GET', 'path': RESOURCE_PATH},
]
ACTION_GET_TAGS = [
{'method': 'GET', 'path': TAGS_PATH},
{'method': 'GET', 'path': TAG_PATH},
]
ACTION_PUT_TAGS = [
{'method': 'PUT', 'path': TAGS_PATH},
{'method': 'PUT', 'path': TAG_PATH},
]
ACTION_DELETE_TAGS = [
{'method': 'DELETE', 'path': TAGS_PATH},
{'method': 'DELETE', 'path': TAG_PATH},
]
rules = [
@ -353,6 +367,17 @@ rules = [
description='Get ``hints`` attribute of a port',
operations=ACTION_GET,
),
policy.DocumentedRuleDefault(
name='get_ports_tags',
check_str=neutron_policy.policy_or(
neutron_policy.RULE_ADVSVC,
base.ADMIN_OR_NET_OWNER_READER,
base.PROJECT_READER
),
scope_types=['project'],
description='Get the port tags',
operations=ACTION_GET_TAGS,
),
# TODO(amotoki): Add get_port:binding:vnic_type
# TODO(amotoki): Add get_port:binding:data_plane_status
@ -591,6 +616,16 @@ rules = [
description='Update ``hints`` attribute of a port',
operations=ACTION_PUT,
),
policy.DocumentedRuleDefault(
name='update_ports_tags',
check_str=neutron_policy.policy_or(
base.ADMIN_OR_PROJECT_MEMBER,
neutron_policy.RULE_ADVSVC
),
scope_types=['project'],
description='Update the port tags',
operations=ACTION_PUT_TAGS,
),
policy.DocumentedRuleDefault(
name='delete_port',
@ -610,6 +645,17 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='delete_ports_tags',
check_str=neutron_policy.policy_or(
neutron_policy.RULE_ADVSVC,
base.PROJECT_MEMBER,
base.ADMIN_OR_NET_OWNER_MEMBER
),
scope_types=['project'],
description='Delete the port tags',
operations=ACTION_DELETE_TAGS,
)
]

35
neutron/conf/policies/router.py
View File

@ -21,6 +21,8 @@ DEPRECATED_REASON = (
COLLECTION_PATH = '/routers'
RESOURCE_PATH = '/routers/{id}'
TAGS_PATH = RESOURCE_PATH + '/tags'
TAG_PATH = RESOURCE_PATH + '/tags/{tag_id}'
ACTION_POST = [
{'method': 'POST', 'path': COLLECTION_PATH},
@ -35,6 +37,18 @@ ACTION_GET = [
{'method': 'GET', 'path': COLLECTION_PATH},
{'method': 'GET', 'path': RESOURCE_PATH},
]
ACTION_GET_TAGS = [
{'method': 'GET', 'path': TAGS_PATH},
{'method': 'GET', 'path': TAG_PATH},
]
ACTION_PUT_TAGS = [
{'method': 'PUT', 'path': TAGS_PATH},
{'method': 'PUT', 'path': TAG_PATH},
]
ACTION_DELETE_TAGS = [
{'method': 'DELETE', 'path': TAGS_PATH},
{'method': 'DELETE', 'path': TAG_PATH},
]
rules = [
@ -180,6 +194,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='get_routers_tags',
check_str=base.ADMIN_OR_PROJECT_READER,
scope_types=['project'],
description='Get the router tags',
operations=ACTION_GET_TAGS,
),
policy.DocumentedRuleDefault(
name='update_router',
@ -284,6 +305,13 @@ rules = [
'updating a router'),
operations=ACTION_POST,
),
policy.DocumentedRuleDefault(
name='update_routers_tags',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
scope_types=['project'],
description='Update the router tags',
operations=ACTION_PUT_TAGS,
),
policy.DocumentedRuleDefault(
name='delete_router',
@ -297,6 +325,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='delete_routers_tags',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
scope_types=['project'],
description='Delete the router tags',
operations=ACTION_DELETE_TAGS,
),
policy.DocumentedRuleDefault(
name='add_router_interface',

42
neutron/conf/policies/security_group.py
View File

@ -22,13 +22,31 @@ DEPRECATED_REASON = (
SG_COLLECTION_PATH = '/security-groups'
SG_RESOURCE_PATH = '/security-groups/{id}'
SG_TAGS_PATH = SG_RESOURCE_PATH + '/tags'
SG_TAG_PATH = SG_RESOURCE_PATH + '/tags/{tag_id}'
RULE_COLLECTION_PATH = '/security-group-rules'
RULE_RESOURCE_PATH = '/security-group-rules/{id}'
RULE_TAGS_PATH = RULE_RESOURCE_PATH + '/tags'
RULE_TAG_PATH = RULE_RESOURCE_PATH + '/tags/{tag_id}'
RULE_ADMIN_OR_SG_OWNER = 'rule:admin_or_sg_owner'
RULE_ADMIN_OWNER_OR_SG_OWNER = 'rule:admin_owner_or_sg_owner'
SG_ACTION_GET_TAGS = [
{'method': 'GET', 'path': SG_TAGS_PATH},
{'method': 'GET', 'path': SG_TAG_PATH},
]
SG_ACTION_PUT_TAGS = [
{'method': 'PUT', 'path': SG_TAGS_PATH},
{'method': 'PUT', 'path': SG_TAG_PATH},
]
SG_ACTION_DELETE_TAGS = [
{'method': 'DELETE', 'path': SG_TAGS_PATH},
{'method': 'DELETE', 'path': SG_TAG_PATH},
]
rules = [
policy.RuleDefault(
name='admin_or_sg_owner',
@ -91,6 +109,16 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='get_security_groups_tags',
check_str=neutron_policy.policy_or(
base.ADMIN_OR_PROJECT_READER,
'rule:shared_security_group'
),
scope_types=['project'],
description='Get the security group tags',
operations=SG_ACTION_GET_TAGS,
),
policy.DocumentedRuleDefault(
name='update_security_group',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
@ -108,6 +136,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='update_security_groups_tags',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
scope_types=['project'],
description='Update the security group tags',
operations=SG_ACTION_PUT_TAGS,
),
policy.DocumentedRuleDefault(
name='delete_security_group',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
@ -125,6 +160,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='delete_security_groups_tags',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
scope_types=['project'],
description='Delete the security group tags',
operations=SG_ACTION_DELETE_TAGS,
),
# TODO(amotoki): admin_or_owner is the right rule?
# Does an empty string make more sense for create_security_group_rule?

36
neutron/conf/policies/segment.py
View File

@ -21,6 +21,21 @@ DEPRECATED_REASON = (
COLLECTION_PATH = '/segments'
RESOURCE_PATH = '/segments/{id}'
TAGS_PATH = RESOURCE_PATH + '/tags'
TAG_PATH = RESOURCE_PATH + '/tags/{tag_id}'
ACTION_GET_TAGS = [
{'method': 'GET', 'path': TAGS_PATH},
{'method': 'GET', 'path': TAG_PATH},
]
ACTION_PUT_TAGS = [
{'method': 'PUT', 'path': TAGS_PATH},
{'method': 'PUT', 'path': TAG_PATH},
]
ACTION_DELETE_TAGS = [
{'method': 'DELETE', 'path': TAGS_PATH},
{'method': 'DELETE', 'path': TAG_PATH},
]
rules = [
@ -62,6 +77,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='get_segments_tags',
check_str=base.ADMIN,
scope_types=['project'],
description='Get the segment tags',
operations=ACTION_GET_TAGS,
),
policy.DocumentedRuleDefault(
name='update_segment',
check_str=base.ADMIN,
@ -79,6 +101,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='update_segments_tags',
check_str=base.ADMIN,
scope_types=['project'],
description='Update the segment tags',
operations=ACTION_PUT_TAGS,
),
policy.DocumentedRuleDefault(
name='delete_segment',
check_str=base.ADMIN,
@ -96,6 +125,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='delete_segments_tags',
check_str=base.ADMIN,
scope_types=['project'],
description='Delete the segment tags',
operations=ACTION_DELETE_TAGS,
),
]

42
neutron/conf/policies/subnet.py
View File

@ -21,6 +21,8 @@ DEPRECATED_REASON = (
COLLECTION_PATH = '/subnets'
RESOURCE_PATH = '/subnets/{id}'
TAGS_PATH = RESOURCE_PATH + '/tags'
TAG_PATH = RESOURCE_PATH + '/tags/{tag_id}'
ACTION_POST = [
{'method': 'POST', 'path': COLLECTION_PATH},
@ -35,6 +37,18 @@ ACTION_GET = [
{'method': 'GET', 'path': COLLECTION_PATH},
{'method': 'GET', 'path': RESOURCE_PATH},
]
ACTION_GET_TAGS = [
{'method': 'GET', 'path': TAGS_PATH},
{'method': 'GET', 'path': TAG_PATH},
]
ACTION_PUT_TAGS = [
{'method': 'PUT', 'path': TAGS_PATH},
{'method': 'PUT', 'path': TAG_PATH},
]
ACTION_DELETE_TAGS = [
{'method': 'DELETE', 'path': TAGS_PATH},
{'method': 'DELETE', 'path': TAG_PATH},
]
rules = [
@ -107,6 +121,16 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='get_subnets_tags',
check_str=neutron_policy.policy_or(
base.ADMIN_OR_NET_OWNER_MEMBER,
base.PROJECT_READER,
'rule:shared'),
scope_types=['project'],
description='Get the subnet tags',
operations=ACTION_GET_TAGS,
),
policy.DocumentedRuleDefault(
name='update_subnet',
check_str=neutron_policy.policy_or(
@ -145,6 +169,15 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='update_subnets_tags',
check_str=neutron_policy.policy_or(
base.ADMIN_OR_NET_OWNER_MEMBER,
base.PROJECT_MEMBER),
scope_types=['project'],
description='Update the subnet tags',
operations=ACTION_PUT_TAGS,
),
policy.DocumentedRuleDefault(
name='delete_subnet',
check_str=neutron_policy.policy_or(
@ -159,6 +192,15 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='delete_subnets_tags',
check_str=neutron_policy.policy_or(
base.ADMIN_OR_NET_OWNER_MEMBER,
base.PROJECT_MEMBER),
scope_types=['project'],
description='Delete the subnet tags',
operations=ACTION_DELETE_TAGS,
),
]

39
neutron/conf/policies/subnetpool.py
View File

@ -24,6 +24,21 @@ RESOURCE_PATH = '/subnetpools/{id}'
ONBOARD_PATH = '/subnetpools/{id}/onboard_network_subnets'
ADD_PREFIXES_PATH = '/subnetpools/{id}/add_prefixes'
REMOVE_PREFIXES_PATH = '/subnetpools/{id}/remove_prefixes'
TAGS_PATH = RESOURCE_PATH + '/tags'
TAG_PATH = RESOURCE_PATH + '/tags/{tag_id}'
ACTION_GET_TAGS = [
{'method': 'GET', 'path': TAGS_PATH},
{'method': 'GET', 'path': TAG_PATH},
]
ACTION_PUT_TAGS = [
{'method': 'PUT', 'path': TAGS_PATH},
{'method': 'PUT', 'path': TAG_PATH},
]
ACTION_DELETE_TAGS = [
{'method': 'DELETE', 'path': TAGS_PATH},
{'method': 'DELETE', 'path': TAG_PATH},
]
rules = [
@ -111,6 +126,16 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='get_subnetpools_tags',
check_str=neutron_policy.policy_or(
base.ADMIN_OR_PROJECT_READER,
'rule:shared_subnetpools'
),
scope_types=['project'],
description='Get the subnetpool tags',
operations=ACTION_GET_TAGS
),
policy.DocumentedRuleDefault(
name='update_subnetpool',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
@ -145,6 +170,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='update_subnetpools_tags',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
scope_types=['project'],
description='Update the subnetpool tags',
operations=ACTION_PUT_TAGS
),
policy.DocumentedRuleDefault(
name='delete_subnetpool',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
@ -162,6 +194,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='delete_subnetpools_tags',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
scope_types=['project'],
description='Delete the subnetpool tags',
operations=ACTION_DELETE_TAGS
),
policy.DocumentedRuleDefault(
name='onboard_network_subnets',
check_str=base.ADMIN_OR_PROJECT_MEMBER,

36
neutron/conf/policies/trunk.py
View File

@ -19,6 +19,21 @@ from neutron.conf.policies import base
COLLECTION_PATH = '/trunks'
RESOURCE_PATH = '/trunks/{id}'
TAGS_PATH = RESOURCE_PATH + '/tags'
TAG_PATH = RESOURCE_PATH + '/tags/{tag_id}'
ACTION_GET_TAGS = [
{'method': 'GET', 'path': TAGS_PATH},
{'method': 'GET', 'path': TAG_PATH},
]
ACTION_PUT_TAGS = [
{'method': 'PUT', 'path': TAGS_PATH},
{'method': 'PUT', 'path': TAG_PATH},
]
ACTION_DELETE_TAGS = [
{'method': 'DELETE', 'path': TAGS_PATH},
{'method': 'DELETE', 'path': TAG_PATH},
]
DEPRECATED_REASON = (
"The trunks API now supports system scope and default roles.")
@ -63,6 +78,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='get_trunks_tags',
check_str=base.ADMIN_OR_PROJECT_READER,
scope_types=['project'],
description='Get the trunk tags',
operations=ACTION_GET_TAGS
),
policy.DocumentedRuleDefault(
name='update_trunk',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
@ -80,6 +102,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='update_trunks_tags',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
scope_types=['project'],
description='Update the trunk tags',
operations=ACTION_PUT_TAGS
),
policy.DocumentedRuleDefault(
name='delete_trunk',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
@ -97,6 +126,13 @@ rules = [
deprecated_reason=DEPRECATED_REASON,
deprecated_since=versionutils.deprecated.WALLABY)
),
policy.DocumentedRuleDefault(
name='delete_trunks_tags',
check_str=base.ADMIN_OR_PROJECT_MEMBER,
scope_types=['project'],
description='Delete a trunk',
operations=ACTION_DELETE_TAGS
),
policy.DocumentedRuleDefault(
name='get_subports',
check_str=base.ADMIN_OR_PROJECT_READER,

119
neutron/extensions/tagging.py
View File

@ -28,6 +28,8 @@ import webob.exc
from neutron._i18n import _
from neutron.api import extensions
from neutron.api.v2 import resource as api_resource
from neutron.objects import subnet
from neutron import policy
TAG = 'tag'
@ -55,6 +57,7 @@ TAG_ATTRIBUTE_MAP_PORTS[TAGS] = {
'validate': {'type:list_of_unique_strings': MAX_TAG_LEN},
'default': [], 'is_visible': True, 'is_filter': True
}
RESOURCES_AND_PARENTS = {'subnets': ('network', subnet.Subnet.get_network_id)}
class TagResourceNotFound(exceptions.NotFound):
@ -95,75 +98,113 @@ class TaggingController(object):
self.plugin = directory.get_plugin(TAG_PLUGIN_TYPE)
self.supported_resources = TAG_SUPPORTED_RESOURCES
def _get_parent_resource_and_id(self, kwargs):
@staticmethod
def _get_target(ctx, res_id, p_res, p_res_id, tag_id=None):
target = {'id': res_id,
'tenant_id': ctx.project_id,
'project_id': ctx.project_id}
if p_res:
target[p_res + '_id'] = p_res_id
if tag_id:
target['tag_id'] = tag_id
return target
@staticmethod
def _get_pparent_resource_and_id(context, resource, resource_id):
"""Retrieve the parent of the resource and ID (e.g.: subnet->net)"""
parent, getter_id = RESOURCES_AND_PARENTS[resource]
parent_id = getter_id(context.elevated(), resource_id)
return parent, parent_id
def _get_parent_resource_and_id(self, context, kwargs):
parent, parent_id = None, None
for key in kwargs:
for resource in self.supported_resources:
if key == self.supported_resources[resource] + '_id':
return resource, kwargs[key]
return None, None
if resource in RESOURCES_AND_PARENTS.keys():
parent, parent_id = self._get_pparent_resource_and_id(
context, resource, kwargs[key])
return resource, kwargs[key], parent, parent_id
return None, None, None, None
def index(self, request, **kwargs):
# GET /v2.0/networks/{network_id}/tags
parent, parent_id = self._get_parent_resource_and_id(kwargs)
return self.plugin.get_tags(request.context, parent, parent_id)
# GET /v2.0/{parent_resource}/{parent_resource_id}/tags
ctx = request.context
res, res_id, p_res, p_res_id = self._get_parent_resource_and_id(
ctx, kwargs)
target = self._get_target(ctx, res_id, p_res, p_res_id)
policy.enforce(ctx, 'get_%s_%s' % (res, TAGS), target)
return self.plugin.get_tags(ctx, res, res_id)
def show(self, request, id, **kwargs):
# GET /v2.0/networks/{network_id}/tags/{tag}
# GET /v2.0/{parent_resource}/{parent_resource_id}/tags/{tag}
# id == tag
validate_tag(id)
parent, parent_id = self._get_parent_resource_and_id(kwargs)
return self.plugin.get_tag(request.context, parent, parent_id, id)
ctx = request.context
res, res_id, p_res, p_res_id = self._get_parent_resource_and_id(
ctx, kwargs)
target = self._get_target(ctx, res_id, p_res, p_res_id, tag_id=id)
policy.enforce(ctx, 'get_%s_%s' % (res, TAGS), target)
return self.plugin.get_tag(ctx, res, res_id, id)
def create(self, request, **kwargs):
# not supported
# POST /v2.0/networks/{network_id}/tags
# POST /v2.0/{parent_resource}/{parent_resource_id}/tags
raise webob.exc.HTTPNotFound("not supported")
def update(self, request, id, **kwargs):
# PUT /v2.0/networks/{network_id}/tags/{tag}
# PUT /v2.0/{parent_resource}/{parent_resource_id}/tags/{tag}
# id == tag
validate_tag(id)
parent, parent_id = self._get_parent_resource_and_id(kwargs)
notify_tag_action(request.context, 'create.start',
parent, parent_id, [id])
result = self.plugin.update_tag(request.context, parent, parent_id, id)
notify_tag_action(request.context, 'create.end',
parent, parent_id, [id])
ctx = request.context
res, res_id, p_res, p_res_id = self._get_parent_resource_and_id(
ctx, kwargs)
target = self._get_target(ctx, res_id, p_res, p_res_id, tag_id=id)
policy.enforce(ctx, 'update_%s_%s' % (res, TAGS), target)
notify_tag_action(ctx, 'create.start', res, res_id, [id])
result = self.plugin.update_tag(ctx, res, res_id, id)
notify_tag_action(ctx, 'create.end', res, res_id, [id])
return result
def update_all(self, request, body, **kwargs):
# PUT /v2.0/networks/{network_id}/tags
# PUT /v2.0/{parent_resource}/{parent_resource_id}/tags
# body: {"tags": ["aaa", "bbb"]}
validate_tags(body)
parent, parent_id = self._get_parent_resource_and_id(kwargs)
notify_tag_action(request.context, 'update.start',
parent, parent_id, body['tags'])
result = self.plugin.update_tags(request.context, parent,
parent_id, body)
notify_tag_action(request.context, 'update.end',
parent, parent_id, body['tags'])
ctx = request.context
res, res_id, p_res, p_res_id = self._get_parent_resource_and_id(
ctx, kwargs)
target = self._get_target(ctx, res_id, p_res, p_res_id)
policy.enforce(ctx, 'update_%s_%s' % (res, TAGS), target)
notify_tag_action(ctx, 'update.start', res, res_id, body['tags'])
result = self.plugin.update_tags(ctx, res, res_id, body)
notify_tag_action(ctx, 'update.end', res, res_id,
body['tags'])
return result
def delete(self, request, id, **kwargs):
# DELETE /v2.0/networks/{network_id}/tags/{tag}
# DELETE /v2.0/{parent_resource}/{parent_resource_id}/tags/{tag}
# id == tag
validate_tag(id)
parent, parent_id = self._get_parent_resource_and_id(kwargs)
notify_tag_action(request.context, 'delete.start',
parent, parent_id, [id])
result = self.plugin.delete_tag(request.context, parent, parent_id, id)
notify_tag_action(request.context, 'delete.end',
parent, parent_id, [id])
ctx = request.context
res, res_id, p_res, p_res_id = self._get_parent_resource_and_id(
ctx, kwargs)
target = self._get_target(ctx, res_id, p_res, p_res_id, tag_id=id)
policy.enforce(ctx, 'delete_%s_%s' % (res, TAGS), target)
notify_tag_action(ctx, 'delete.start', res, res_id, [id])
result = self.plugin.delete_tag(ctx, res, res_id, id)
notify_tag_action(ctx, 'delete.end', res, res_id, [id])
return result
def delete_all(self, request, **kwargs):
# DELETE /v2.0/networks/{network_id}/tags
parent, parent_id = self._get_parent_resource_and_id(kwargs)
notify_tag_action(request.context, 'delete_all.start',
parent, parent_id)
result = self.plugin.delete_tags(request.context, parent, parent_id)
notify_tag_action(request.context, 'delete_all.end',
parent, parent_id)
# DELETE /v2.0/{parent_resource}/{parent_resource_id}/tags
ctx = request.context
res, res_id, p_res, p_res_id = self._get_parent_resource_and_id(
ctx, kwargs)
target = self._get_target(ctx, res_id, p_res, p_res_id)
policy.enforce(ctx, 'delete_%s_%s' % (res, TAGS), target)
notify_tag_action(ctx, 'delete_all.start', res, res_id)
result = self.plugin.delete_tags(ctx, res, res_id)
notify_tag_action(ctx, 'delete_all.end', res, res_id)
return result

11
neutron/objects/subnet.py
View File

@ -13,6 +13,7 @@
import netaddr
from neutron_lib.api import validators
from neutron_lib import constants as const
from neutron_lib.db import api as db_api
from neutron_lib.db import model_query
from neutron_lib.objects import common_types
from neutron_lib.utils import net as net_utils
@ -22,6 +23,7 @@ from oslo_utils import versionutils
from oslo_versionedobjects import fields as obj_fields
from sqlalchemy import and_, or_
from sqlalchemy import orm
from sqlalchemy.orm import exc as orm_exc
from sqlalchemy.sql import exists
from neutron.db.models import dns as dns_models
@ -545,6 +547,15 @@ class Subnet(base.NeutronDbObject):
return [segment_id for (segment_id,) in query.all()]
@classmethod
@db_api.CONTEXT_READER
def get_network_id(cls, context, subnet_id):
try:
return context.session.query(cls.db_model.network_id).filter(
cls.db_model.id == subnet_id).one()[0]
except orm_exc.NoResultFound:
return None
@base.NeutronObjectRegistry.register
class NetworkSubnetLock(base.NeutronDbObject):

62
neutron/tests/unit/conf/policies/test_floatingip.py
View File

@ -69,6 +69,16 @@ class SystemAdminTests(FloatingIPAPITestCase):
policy.enforce,
self.context, "get_floatingip", self.alt_target)
def test_get_floatingips_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, "get_floatingips_tags", self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, "get_floatingips_tags", self.alt_target)
def test_update_floatingip(self):
self.assertRaises(
base_policy.InvalidScope,
@ -79,6 +89,16 @@ class SystemAdminTests(FloatingIPAPITestCase):
policy.enforce,
self.context, "update_floatingip", self.alt_target)
def test_update_floatingips_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, "update_floatingips_tags", self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, "update_floatingips_tags", self.alt_target)
def test_delete_floatingip(self):
self.assertRaises(
base_policy.InvalidScope,
@ -132,12 +152,27 @@ class AdminTests(FloatingIPAPITestCase):
self.assertTrue(
policy.enforce(self.context, "get_floatingip", self.alt_target))
def test_get_floatingips_tags(self):
self.assertTrue(
policy.enforce(self.context, "get_floatingips_tags", self.target))
self.assertTrue(
policy.enforce(self.context, "get_floatingips_tags",
self.alt_target))
def test_update_floatingip(self):
self.assertTrue(
policy.enforce(self.context, "update_floatingip", self.target))
self.assertTrue(
policy.enforce(self.context, "update_floatingip", self.alt_target))
def test_update_floatingips_tags(self):
self.assertTrue(
policy.enforce(self.context, "update_floatingips_tags",
self.target))
self.assertTrue(
policy.enforce(self.context, "update_floatingips_tags",
self.alt_target))
def test_delete_floatingip(self):
self.assertTrue(
policy.enforce(self.context, "delete_floatingip", self.target))
@ -178,6 +213,14 @@ class ProjectMemberTests(AdminTests):
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, "get_floatingip", self.alt_target)
def test_get_floatingips_tags(self):
self.assertTrue(
policy.enforce(self.context, "get_floatingips_tags", self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, "get_floatingips_tags",
self.alt_target)
def test_update_floatingip(self):
self.assertTrue(
policy.enforce(self.context, "update_floatingip", self.target))
@ -185,6 +228,15 @@ class ProjectMemberTests(AdminTests):
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, "update_floatingip", self.alt_target)
def test_update_floatingips_tags(self):
self.assertTrue(
policy.enforce(self.context, "update_floatingips_tags",
self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, "update_floatingips_tags",
self.alt_target)
def test_delete_floatingip(self):
self.assertTrue(
policy.enforce(self.context, "delete_floatingip", self.target))
@ -219,6 +271,16 @@ class ProjectReaderTests(ProjectMemberTests):
policy.enforce,
self.context, "update_floatingip", self.alt_target)
def test_update_floatingips_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, "update_floatingips_tags", self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, "update_floatingips_tags", self.alt_target)
def test_delete_floatingip(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,

92
neutron/tests/unit/conf/policies/test_network.py
View File

@ -184,6 +184,18 @@ class SystemAdminTests(NetworkAPITestCase):
self.context, 'get_network:provider:segmentation_id',
self.alt_target)
def test_get_networks_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_networks_tags',
self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_networks_tags',
self.alt_target)
def test_update_network(self):
self.assertRaises(
base_policy.InvalidScope,
@ -279,6 +291,15 @@ class SystemAdminTests(NetworkAPITestCase):
self.context, 'update_network:port_security_enabled',
self.alt_target)
def test_update_networks_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'update_networks_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'update_networks_tags',
self.alt_target)
def test_delete_network(self):
self.assertRaises(
base_policy.InvalidScope,
@ -287,6 +308,15 @@ class SystemAdminTests(NetworkAPITestCase):
base_policy.InvalidScope,
policy.enforce, self.context, 'delete_network', self.alt_target)
def test_delete_networks_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'delete_networks_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'delete_networks_tags',
self.alt_target)
class SystemMemberTests(SystemAdminTests):
@ -421,6 +451,12 @@ class AdminTests(NetworkAPITestCase):
'get_network:provider:segmentation_id',
self.alt_target))
def test_get_networks_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_networks_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'get_networks_tags', self.alt_target))
def test_update_network(self):
self.assertTrue(
policy.enforce(self.context, 'update_network', self.target))
@ -498,12 +534,26 @@ class AdminTests(NetworkAPITestCase):
'update_network:port_security_enabled',
self.alt_target))
def test_update_networks_tags(self):
self.assertTrue(
policy.enforce(self.context, 'update_networks_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'update_networks_tags',
self.alt_target))
def test_delete_network(self):
self.assertTrue(
policy.enforce(self.context, 'delete_network', self.target))
self.assertTrue(
policy.enforce(self.context, 'delete_network', self.alt_target))
def test_delete_networks_tags(self):
self.assertTrue(
policy.enforce(self.context, 'delete_networks_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'delete_networks_tags',
self.alt_target))
class ProjectMemberTests(AdminTests):
@ -657,6 +707,14 @@ class ProjectMemberTests(AdminTests):
self.context, 'get_network:provider:segmentation_id',
self.alt_target)
def test_get_networks_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_networks_tags', self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'get_networks_tags', self.alt_target)
def test_update_network(self):
self.assertTrue(
policy.enforce(self.context, 'update_network', self.target))
@ -751,6 +809,14 @@ class ProjectMemberTests(AdminTests):
self.context, 'update_network:port_security_enabled',
self.alt_target)
def test_update_networks_tags(self):
self.assertTrue(
policy.enforce(self.context, 'update_networks_tags', self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_networks_tags', self.alt_target)
def test_delete_network(self):
self.assertTrue(
policy.enforce(self.context, 'delete_network', self.target))
@ -759,6 +825,14 @@ class ProjectMemberTests(AdminTests):
policy.enforce,
self.context, 'delete_network', self.alt_target)
def test_delete_networks_tags(self):
self.assertTrue(
policy.enforce(self.context, 'delete_networks_tags', self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_networks_tags', self.alt_target)
class ProjectReaderTests(ProjectMemberTests):
@ -806,6 +880,15 @@ class ProjectReaderTests(ProjectMemberTests):
self.context, 'update_network:port_security_enabled',
self.alt_target)
def test_update_networks_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'update_networks_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'update_networks_tags',
self.alt_target)
def test_delete_network(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
@ -813,3 +896,12 @@ class ProjectReaderTests(ProjectMemberTests):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'delete_network', self.alt_target)
def test_delete_networks_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'delete_networks_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'delete_networks_tags',
self.alt_target)

51
neutron/tests/unit/conf/policies/test_network_segment_range.py
View File

@ -44,18 +44,36 @@ class SystemAdminTests(NetworkSegmentRangeAPITestCase):
policy.enforce,
self.context, 'get_network_segment_range', self.target)
def test_get_network_segment_ranges_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_network_segment_ranges_tags', self.target)
def test_update_network_segment_range(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_network_segment_range', self.target)
def test_update_network_segment_ranges_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_network_segment_ranges_tags', self.target)
def test_delete_network_segment_range(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_network_segment_range', self.target)
def test_delete_network_segment_ranges_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_network_segment_ranges_tags', self.target)
class SystemMemberTests(SystemAdminTests):
@ -87,16 +105,31 @@ class AdminTests(NetworkSegmentRangeAPITestCase):
policy.enforce(self.context,
'get_network_segment_range', self.target))
def test_get_network_segment_ranges_tags(self):
self.assertTrue(
policy.enforce(self.context,
'get_network_segment_ranges_tags', self.target))
def test_update_network_segment_range(self):
self.assertTrue(
policy.enforce(self.context,
'update_network_segment_range', self.target))
def test_update_network_segment_ranges_tags(self):
self.assertTrue(
policy.enforce(self.context,
'update_network_segment_ranges_tags', self.target))
def test_delete_network_segment_range(self):
self.assertTrue(
policy.enforce(self.context,
'delete_network_segment_range', self.target))
def test_delete_network_segment_ranges_tags(self):
self.assertTrue(
policy.enforce(self.context,
'delete_network_segment_ranges_tags', self.target))
class ProjectMemberTests(AdminTests):
@ -116,18 +149,36 @@ class ProjectMemberTests(AdminTests):
policy.enforce,
self.context, 'get_network_segment_range', self.target)
def test_get_network_segment_ranges_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'get_network_segment_ranges_tags', self.target)
def test_update_network_segment_range(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_network_segment_range', self.target)
def test_update_network_segment_ranges_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_network_segment_ranges_tags', self.target)
def test_delete_network_segment_range(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_network_segment_range', self.target)
def test_delete_network_segment_ranges_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_network_segment_ranges_tags', self.target)
class ProjectReaderTests(ProjectMemberTests):

36
neutron/tests/unit/conf/policies/test_port.py
View File

@ -246,6 +246,14 @@ class SystemAdminTests(PortAPITestCase):
policy.enforce, self.context, 'get_port:resource_request',
self.alt_target)
def test_get_ports_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'get_ports_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce, self.context, 'get_ports_tags', self.alt_target)
def test_update_port(self):
self.assertRaises(
base_policy.InvalidScope,
@ -596,6 +604,12 @@ class AdminTests(PortAPITestCase):
policy.enforce(
self.context, 'get_port:hints', self.alt_target))
def test_get_ports_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_ports_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'get_ports_tags', self.alt_target))
def test_update_port(self):
self.assertTrue(
policy.enforce(self.context, 'update_port', self.target))
@ -957,6 +971,13 @@ class ProjectMemberTests(AdminTests):
policy.enforce, self.context, 'get_port:hints',
self.alt_target)
def test_get_ports_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_ports_tags', self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'get_ports_tags', self.alt_target)
def test_update_port(self):
self.assertTrue(
policy.enforce(self.context, 'update_port', self.target))
@ -1113,6 +1134,13 @@ class ProjectMemberTests(AdminTests):
policy.enforce,
self.context, 'update_port:hints', self.alt_target)
def test_update_ports_tags(self):
self.assertTrue(
policy.enforce(self.context, 'update_ports_tags', self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'update_ports_tags', self.alt_target)
def test_delete_port(self):
self.assertTrue(
policy.enforce(self.context, 'delete_port', self.target))
@ -1163,6 +1191,14 @@ class ProjectReaderTests(ProjectMemberTests):
policy.enforce, self.context, 'update_port:binding:vnic_type',
self.alt_target)
def test_update_ports_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'update_ports_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'update_ports_tags', self.alt_target)
def test_delete_port(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,

94
neutron/tests/unit/conf/policies/test_router.py
View File

@ -168,6 +168,16 @@ class SystemAdminTests(RouterAPITestCase):
policy.enforce,
self.context, 'get_router:ha', self.alt_target)
def test_get_routers_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_routers_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_routers_tags', self.alt_target)
def test_update_router(self):
self.assertRaises(
base_policy.InvalidScope,
@ -272,6 +282,16 @@ class SystemAdminTests(RouterAPITestCase):
self.context, 'update_router:enable_default_route_ecmp',
self.alt_target)
def test_update_routers_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_routers_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_routers_tags', self.alt_target)
def test_delete_router(self):
self.assertRaises(
base_policy.InvalidScope,
@ -282,6 +302,16 @@ class SystemAdminTests(RouterAPITestCase):
policy.enforce,
self.context, 'delete_router', self.alt_target)
def test_delete_routers_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_routers_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_routers_tags', self.alt_target)
def test_add_router_interface(self):
self.assertRaises(
base_policy.InvalidScope,
@ -429,6 +459,12 @@ class AdminTests(RouterAPITestCase):
self.assertTrue(
policy.enforce(self.context, 'get_router:ha', self.alt_target))
def test_get_routers_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_routers_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'get_routers_tags', self.alt_target))
def test_update_router(self):
self.assertTrue(
policy.enforce(self.context, 'update_router', self.target))
@ -491,12 +527,26 @@ class AdminTests(RouterAPITestCase):
'update_router:external_gateway_info:external_fixed_ips',
self.alt_target))
def test_update_routers_tags(self):
self.assertTrue(
policy.enforce(self.context, 'update_routers_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'update_routers_tags',
self.alt_target))
def test_delete_router(self):
self.assertTrue(
policy.enforce(self.context, 'delete_router', self.target))
self.assertTrue(
policy.enforce(self.context, 'delete_router', self.alt_target))
def test_delete_routers_tags(self):
self.assertTrue(
policy.enforce(self.context, 'delete_routers_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'delete_routers_tags',
self.alt_target))
def test_add_router_interface(self):
self.assertTrue(
policy.enforce(self.context,
@ -652,6 +702,14 @@ class ProjectMemberTests(AdminTests):
policy.enforce,
self.context, 'get_router:ha', self.alt_target)
def test_get_routers_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_routers_tags', self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'get_routers_tags', self.alt_target)
def test_update_router(self):
self.assertTrue(
policy.enforce(self.context, 'update_router', self.target))
@ -728,6 +786,14 @@ class ProjectMemberTests(AdminTests):
'update_router:external_gateway_info:external_fixed_ips',
self.alt_target)
def test_update_routers_tags(self):
self.assertTrue(
policy.enforce(self.context, 'update_routers_tags', self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_routers_tags', self.alt_target)
def test_delete_router(self):
self.assertTrue(
policy.enforce(self.context, 'delete_router', self.target))
@ -736,6 +802,14 @@ class ProjectMemberTests(AdminTests):
policy.enforce,
self.context, 'delete_router', self.alt_target)
def test_delete_routers_tags(self):
self.assertTrue(
policy.enforce(self.context, 'delete_routers_tags', self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_routers_tags', self.alt_target)
def test_add_router_interface(self):
self.assertTrue(
policy.enforce(self.context,
@ -829,6 +903,16 @@ class ProjectReaderTests(ProjectMemberTests):
self.context, 'update_router:external_gateway_info:network_id',
self.alt_target)
def test_update_routers_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_routers_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_routers_tags', self.alt_target)
def test_delete_router(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
@ -839,6 +923,16 @@ class ProjectReaderTests(ProjectMemberTests):
policy.enforce,
self.context, 'delete_router', self.alt_target)
def test_delete_routers_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_routers_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_routers_tags', self.alt_target)
def test_add_router_interface(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,

110
neutron/tests/unit/conf/policies/test_security_group.py
View File

@ -56,6 +56,16 @@ class SystemAdminSecurityGroupTests(SecurityGroupAPITestCase):
policy.enforce,
self.context, 'get_security_group', self.alt_target)
def test_get_security_groups_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_security_groups_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_security_groups_tags', self.alt_target)
def test_update_security_group(self):
self.assertRaises(
base_policy.InvalidScope,
@ -66,6 +76,16 @@ class SystemAdminSecurityGroupTests(SecurityGroupAPITestCase):
policy.enforce,
self.context, 'update_security_group', self.alt_target)
def test_update_security_groups_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_security_groups_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_security_groups_tags', self.alt_target)
def test_delete_security_group(self):
self.assertRaises(
base_policy.InvalidScope,
@ -76,6 +96,16 @@ class SystemAdminSecurityGroupTests(SecurityGroupAPITestCase):
policy.enforce,
self.context, 'delete_security_group', self.alt_target)
def test_delete_security_groups_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_security_groups_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_security_groups_tags', self.alt_target)
class SystemMemberSecurityGroupTests(SystemAdminSecurityGroupTests):
@ -111,6 +141,14 @@ class AdminSecurityGroupTests(SecurityGroupAPITestCase):
policy.enforce(
self.context, 'get_security_group', self.alt_target))
def test_get_security_groups_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_security_groups_tags',
self.target))
self.assertTrue(
policy.enforce(self.context, 'get_security_groups_tags',
self.alt_target))
def test_update_security_group(self):
self.assertTrue(
policy.enforce(self.context, 'update_security_group', self.target))
@ -118,6 +156,14 @@ class AdminSecurityGroupTests(SecurityGroupAPITestCase):
policy.enforce(
self.context, 'update_security_group', self.alt_target))
def test_update_security_groups_tags(self):
self.assertTrue(
policy.enforce(self.context, 'update_security_groups_tags',
self.target))
self.assertTrue(
policy.enforce(self.context, 'update_security_groups_tags',
self.alt_target))
def test_delete_security_group(self):
self.assertTrue(
policy.enforce(self.context, 'delete_security_group', self.target))
@ -125,6 +171,14 @@ class AdminSecurityGroupTests(SecurityGroupAPITestCase):
policy.enforce(
self.context, 'delete_security_group', self.alt_target))
def test_delete_security_groups_tags(self):
self.assertTrue(
policy.enforce(self.context, 'delete_security_groups_tags',
self.target))
self.assertTrue(
policy.enforce(self.context, 'delete_security_groups_tags',
self.alt_target))
class ProjectMemberSecurityGroupTests(AdminSecurityGroupTests):
@ -148,6 +202,14 @@ class ProjectMemberSecurityGroupTests(AdminSecurityGroupTests):
policy.enforce,
self.context, 'get_security_group', self.alt_target)
def test_get_security_groups_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_security_groups_tags',
self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized, policy.enforce,
self.context, 'get_security_groups_tags', self.alt_target)
def test_update_security_group(self):
self.assertTrue(
policy.enforce(self.context, 'update_security_group', self.target))
@ -156,6 +218,15 @@ class ProjectMemberSecurityGroupTests(AdminSecurityGroupTests):
policy.enforce,
self.context, 'update_security_group', self.alt_target)
def test_update_security_groups_tags(self):
self.assertTrue(
policy.enforce(self.context, 'update_security_groups_tags',
self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce, self.context, 'update_security_groups_tags',
self.alt_target)
def test_delete_security_group(self):
self.assertTrue(
policy.enforce(self.context, 'delete_security_group', self.target))
@ -164,6 +235,14 @@ class ProjectMemberSecurityGroupTests(AdminSecurityGroupTests):
policy.enforce,
self.context, 'delete_security_group', self.alt_target)
def test_delete_security_groups_tags(self):
self.assertTrue(
policy.enforce(self.context, 'delete_security_groups_tags',
self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized, policy.enforce,
self.context, 'delete_security_groups_tags', self.alt_target)
class ProjectReaderSecurityGroupTests(ProjectMemberSecurityGroupTests):
@ -191,6 +270,16 @@ class ProjectReaderSecurityGroupTests(ProjectMemberSecurityGroupTests):
policy.enforce,
self.context, 'update_security_group', self.alt_target)
def test_update_security_groups_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_security_groups_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_security_groups_tags', self.alt_target)
def test_delete_security_group(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
@ -201,6 +290,16 @@ class ProjectReaderSecurityGroupTests(ProjectMemberSecurityGroupTests):
policy.enforce,
self.context, 'delete_security_group', self.alt_target)
def test_delete_security_groups_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_security_groups_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_security_groups_tags', self.alt_target)
class SecurityGroupRuleAPITestCase(base.PolicyBaseTestCase):
@ -332,17 +431,6 @@ class ProjectMemberSecurityGroupRuleTests(AdminSecurityGroupRuleTests):
policy.enforce,
self.context, 'get_security_group_rule', self.alt_target)
# Owner of the security group can get rule which belongs to that group,
# even if security group rule belongs to someone else
sg_owner_target = {
'project_id': 'some-other-project',
'security_group:tenant_id': self.project_id,
'security_group_id': self.sg['id'],
'ext_parent_security_group_id': self.sg['id']}
self.assertTrue(
policy.enforce(self.context,
'get_security_group_rule', sg_owner_target))
def test_delete_security_group_rule(self):
self.assertTrue(
policy.enforce(self.context,

48
neutron/tests/unit/conf/policies/test_segment.py
View File

@ -44,18 +44,36 @@ class SystemAdminTests(SegmentAPITestCase):
policy.enforce,
self.context, 'get_segment', self.target)
def test_get_segments_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_segments_tags', self.target)
def test_update_segment(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_segment', self.target)
def test_update_segments_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_segments_tags', self.target)
def test_delete_segment(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_segment', self.target)
def test_delete_segments_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_segments_tags', self.target)
class SystemMemberTests(SystemAdminTests):
@ -85,14 +103,26 @@ class AdminTests(SegmentAPITestCase):
self.assertTrue(
policy.enforce(self.context, 'get_segment', self.target))
def test_get_segments_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_segments_tags', self.target))
def test_update_segment(self):
self.assertTrue(
policy.enforce(self.context, 'update_segment', self.target))
def test_update_segments_tags(self):
self.assertTrue(
policy.enforce(self.context, 'update_segments_tags', self.target))
def test_delete_segment(self):
self.assertTrue(
policy.enforce(self.context, 'delete_segment', self.target))
def test_delete_segments_tags(self):
self.assertTrue(
policy.enforce(self.context, 'delete_segments_tags', self.target))
class ProjectMemberTests(AdminTests):
@ -112,18 +142,36 @@ class ProjectMemberTests(AdminTests):
policy.enforce,
self.context, 'get_segment', self.target)
def test_get_segments_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'get_segments_tags', self.target)
def test_update_segment(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_segment', self.target)
def test_update_segments_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_segments_tags', self.target)
def test_delete_segment(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_segment', self.target)
def test_delete_segments_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_segments_tags', self.target)
class ProjectReaderTests(ProjectMemberTests):

132
neutron/tests/unit/conf/policies/test_subnet.py
View File

@ -146,6 +146,20 @@ class SystemAdminTests(SubnetAPITestCase):
policy.enforce,
self.context, 'get_subnet:segment_id', self.alt_target)
def test_get_subnets_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_subnets_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_subnets_tags', self.target_net_alt_target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_subnets_tags', self.alt_target)
def test_update_subnet(self):
self.assertRaises(
base_policy.InvalidScope,
@ -190,6 +204,20 @@ class SystemAdminTests(SubnetAPITestCase):
policy.enforce,
self.context, 'update_subnet:service_types', self.alt_target)
def test_update_subnets_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_subnets_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_subnets_tags', self.target_net_alt_target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_subnets_tags', self.alt_target)
def test_delete_subnet(self):
self.assertRaises(
base_policy.InvalidScope,
@ -204,6 +232,20 @@ class SystemAdminTests(SubnetAPITestCase):
policy.enforce,
self.context, 'delete_subnet', self.alt_target)
def test_delete_subnets_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_subnets_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_subnets_tags', self.target_net_alt_target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_subnets_tags', self.alt_target)
class SystemMemberTests(SystemAdminTests):
@ -277,6 +319,15 @@ class AdminTests(SubnetAPITestCase):
policy.enforce(
self.context, 'get_subnet:segment_id', self.alt_target))
def test_get_subnets_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_subnets_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'get_subnets_tags',
self.target_net_alt_target))
self.assertTrue(
policy.enforce(self.context, 'get_subnets_tags', self.alt_target))
def test_update_subnet(self):
self.assertTrue(
policy.enforce(self.context, 'update_subnet', self.target))
@ -310,6 +361,16 @@ class AdminTests(SubnetAPITestCase):
policy.enforce(
self.context, 'update_subnet:service_types', self.alt_target))
def test_update_subnets_tags(self):
self.assertTrue(
policy.enforce(self.context, 'update_subnets_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'update_subnets_tags',
self.target_net_alt_target))
self.assertTrue(
policy.enforce(self.context, 'update_subnets_tags',
self.alt_target))
def test_delete_subnet(self):
self.assertTrue(
policy.enforce(self.context, 'delete_subnet', self.target))
@ -319,6 +380,16 @@ class AdminTests(SubnetAPITestCase):
self.assertTrue(
policy.enforce(self.context, 'delete_subnet', self.alt_target))
def test_delete_subnets_tags(self):
self.assertTrue(
policy.enforce(self.context, 'delete_subnets_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'delete_subnets_tags',
self.target_net_alt_target))
self.assertTrue(
policy.enforce(self.context, 'delete_subnets_tags',
self.alt_target))
class ProjectMemberTests(AdminTests):
@ -393,6 +464,17 @@ class ProjectMemberTests(AdminTests):
policy.enforce,
self.context, 'get_subnet:segment_id', self.alt_target)
def test_get_subnets_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_subnets_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'get_subnets_tags',
self.target_net_alt_target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'get_subnets_tags', self.alt_target)
def test_update_subnet(self):
self.assertTrue(
policy.enforce(self.context, 'update_subnet', self.target))
@ -434,6 +516,17 @@ class ProjectMemberTests(AdminTests):
policy.enforce,
self.context, 'update_subnet:service_types', self.alt_target)
def test_update_subnets_tags(self):
self.assertTrue(
policy.enforce(self.context, 'update_subnets_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'update_subnets_tags',
self.target_net_alt_target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_subnets_tags', self.alt_target)
def test_delete_subnet(self):
self.assertTrue(
policy.enforce(self.context, 'delete_subnet', self.target))
@ -445,6 +538,17 @@ class ProjectMemberTests(AdminTests):
policy.enforce,
self.context, 'delete_subnet', self.alt_target)
def test_delete_subnets_tags(self):
self.assertTrue(
policy.enforce(self.context, 'delete_subnets_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'delete_subnets_tags',
self.target_net_alt_target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_subnets_tags', self.alt_target)
class ProjectReaderTests(ProjectMemberTests):
@ -480,6 +584,20 @@ class ProjectReaderTests(ProjectMemberTests):
policy.enforce,
self.context, 'update_subnet', self.alt_target)
def test_update_subnets_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_subnets_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_subnets_tags', self.target_net_alt_target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_subnets_tags', self.alt_target)
def test_delete_subnet(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
@ -493,3 +611,17 @@ class ProjectReaderTests(ProjectMemberTests):
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_subnet', self.alt_target)
def test_delete_subnets_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_subnets_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_subnets_tags', self.target_net_alt_target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_subnets_tags', self.alt_target)

99
neutron/tests/unit/conf/policies/test_subnetpool.py
View File

@ -73,6 +73,16 @@ class SystemAdminTests(SubnetpoolAPITestCase):
policy.enforce,
self.context, 'get_subnetpool', self.alt_target)
def test_get_subnetpools_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_subnetpools_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_subnetpools_tags', self.alt_target)
def test_update_subnetpool(self):
self.assertRaises(
base_policy.InvalidScope,
@ -93,6 +103,16 @@ class SystemAdminTests(SubnetpoolAPITestCase):
policy.enforce,
self.context, 'update_subnetpool:is_default', self.alt_target)
def test_update_subnetpools_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_subnetpools_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_subnetpools_tags', self.alt_target)
def test_delete_subnetpool(self):
self.assertRaises(
base_policy.InvalidScope,
@ -103,6 +123,16 @@ class SystemAdminTests(SubnetpoolAPITestCase):
policy.enforce,
self.context, 'delete_subnetpool', self.alt_target)
def test_delete_subnetpools_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_subnetpools_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_subnetpools_tags', self.alt_target)
def test_onboard_network_subnets(self):
self.assertRaises(
base_policy.InvalidScope,
@ -182,6 +212,13 @@ class AdminTests(SubnetpoolAPITestCase):
self.assertTrue(
policy.enforce(self.context, 'get_subnetpool', self.alt_target))
def test_get_subnetpools_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_subnetpools_tags', self.target))
self.assertTrue(
policy.enforce(self.context, 'get_subnetpools_tags',
self.alt_target))
def test_update_subnetpool(self):
self.assertTrue(
policy.enforce(self.context, 'update_subnetpool', self.target))
@ -196,12 +233,28 @@ class AdminTests(SubnetpoolAPITestCase):
policy.enforce(
self.context, 'update_subnetpool:default', self.alt_target))
def test_update_subnetpools_tags(self):
self.assertTrue(
policy.enforce(self.context, 'update_subnetpools_tags',
self.target))
self.assertTrue(
policy.enforce(self.context, 'update_subnetpools_tags',
self.alt_target))
def test_delete_subnetpool(self):
self.assertTrue(
policy.enforce(self.context, 'delete_subnetpool', self.target))
self.assertTrue(
policy.enforce(self.context, 'delete_subnetpool', self.alt_target))
def test_delete_subnetpools_tags(self):
self.assertTrue(
policy.enforce(self.context, 'delete_subnetpools_tags',
self.target))
self.assertTrue(
policy.enforce(self.context, 'delete_subnetpools_tags',
self.alt_target))
def test_onboard_network_subnets(self):
self.assertTrue(
policy.enforce(self.context,
@ -265,6 +318,14 @@ class ProjectMemberTests(AdminTests):
policy.enforce,
self.context, 'get_subnetpool', self.alt_target)
def test_get_subnetpools_tags(self):
self.assertTrue(
policy.enforce(self.context, 'get_subnetpools_tags', self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'get_subnetpools_tags', self.alt_target)
def test_update_subnetpool(self):
self.assertTrue(
policy.enforce(self.context, 'update_subnetpool', self.target))
@ -283,6 +344,15 @@ class ProjectMemberTests(AdminTests):
policy.enforce,
self.context, 'update_subnetpool:is_default', self.alt_target)
def test_update_subnetpools_tags(self):
self.assertTrue(
policy.enforce(self.context, 'update_subnetpools_tags',
self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_subnetpools_tags', self.alt_target)
def test_delete_subnetpool(self):
self.assertTrue(
policy.enforce(self.context, 'delete_subnetpool', self.target))
@ -291,6 +361,15 @@ class ProjectMemberTests(AdminTests):
policy.enforce,
self.context, 'delete_subnetpool', self.alt_target)
def test_delete_subnetpools_tags(self):
self.assertTrue(
policy.enforce(self.context, 'delete_subnetpools_tags',
self.target))
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_subnetpools_tags', self.alt_target)
def test_onboard_network_subnets(self):
self.assertTrue(
policy.enforce(self.context,
@ -343,6 +422,16 @@ class ProjectReaderTests(ProjectMemberTests):
policy.enforce,
self.context, 'update_subnetpool', self.alt_target)
def test_update_subnetpools_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_subnetpools_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'update_subnetpools_tags', self.alt_target)
def test_delete_subnetpool(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
@ -353,6 +442,16 @@ class ProjectReaderTests(ProjectMemberTests):
policy.enforce,
self.context, 'delete_subnetpool', self.alt_target)
def test_delete_subnetpools_tags(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_subnetpools_tags', self.target)
self.assertRaises(
base_policy.PolicyNotAuthorized,
policy.enforce,
self.context, 'delete_subnetpools_tags', self.alt_target)
def test_onboard_network_subnets(self):
self.assertRaises(
base_policy.PolicyNotAuthorized,

30
neutron/tests/unit/conf/policies/test_trunk.py
View File

@ -53,6 +53,16 @@ class SystemAdminTests(TrunkAPITestCase):
policy.enforce,
self.context, 'get_trunk', self.alt_target)
def test_get_trunks_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_trunks_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'get_trunks_tags', self.alt_target)
def test_update_trunk(self):
self.assertRaises(
base_policy.InvalidScope,
@ -63,6 +73,16 @@ class SystemAdminTests(TrunkAPITestCase):
policy.enforce,
self.context, 'update_trunk', self.alt_target)
def test_update_trunks_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_trunks_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'update_trunks_tags', self.alt_target)
def test_delete_trunk(self):
self.assertRaises(
base_policy.InvalidScope,
@ -73,6 +93,16 @@ class SystemAdminTests(TrunkAPITestCase):
policy.enforce,
self.context, 'delete_trunk', self.alt_target)
def test_delete_trunks_tags(self):
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_trunks_tags', self.target)
self.assertRaises(
base_policy.InvalidScope,
policy.enforce,
self.context, 'delete_trunks_tags', self.alt_target)
def test_get_subports(self):
self.assertRaises(
base_policy.InvalidScope,

7
releasenotes/notes/resource-tags-policies-a2ffd52e57d7b4b8.yaml Normal file
View File

@ -0,0 +1,7 @@
---
features:
- |
Added the tags policies for the following resources: network, subnet, port,
router, floating IP, network segment, network segment range, security
group and security group rule. The policies control the creation, the
update and the deletion of the resource tags.