x
/
python-group-based-policy-client
Code Issues Proposed changes

Make list and dict input consistent, other fixes 

With some recent changes, a dictionary argument could be provided
as:
key1,key2,key3

But list arguments had to be space separated:
"x1 x2 x3"

This patch makes this consistent, to always be comma separated. At the
same time backward compatibility is being preserved.

Some other bugs are also getting fixed on account of this clean up:
Closes-bug: 1505452
Closes-bug: 1452155
Closes-bug: 1482310
Closes-bug: 1505343
Closes-bug: 1501659
Closes-bug: 1452156
Closes-bug: 1486725
Closes-bug: 1403192

Change-Id: Ie9128e35fb75af71249dfdcfeef2c1f4f812f118
This commit is contained in:
Sumit Naiksatam 2015-10-13 04:28:06 -07:00
parent a12a46bb8e
commit 3cbbf82073
12 changed files with 526 additions and 213 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
gbpclient/common/utils.py
View File

@ -12,6 +12,9 @@
#
import re
def str2dict(strdict):
"""Convert key1=value1,key2=value2,... string into dictionary.
@ -29,3 +32,16 @@ def str2dict(strdict):
return {}
return dict([kv.split('=', 1) if '=' in kv else [kv, ""]
for kv in strdict.split(',')])
def str2list(strlist):
"""Convert key1,key2,... string into list.
:param strlist: key1,key2
strlist can be comma or space separated.
"""
if strlist is not None:
strlist = strlist.strip(', ')
if not strlist:
return []
return re.split("[, ]+", strlist)

374
gbpclient/gbp/v2_0/groupbasedpolicy.py
View File

@ -12,7 +12,6 @@
#
import logging
import string
from neutronclient.i18n import _
from neutronclient.neutron import v2_0 as neutronV20
@ -43,7 +42,8 @@ class ListPolicyTarget(neutronV20.ListCommand):
resource = 'policy_target'
log = logging.getLogger(__name__ + '.ListPolicyTarget')
_formatters = {}
list_columns = ['id', 'name', 'description', 'policy_target_group_id']
list_columns = ['id', 'name', 'description', 'policy_target_group_id',
'port_id']
pagination_support = True
sorting_support = True
@ -66,16 +66,14 @@ class CreatePolicyTarget(neutronV20.CreateCommand):
'--description',
help=_('Description of the Policy Target'))
parser.add_argument(
'--policy-target-group', metavar='PTG',
default='',
help=_('Policy Target Group uuid'))
'--policy-target-group',
help=_('Policy Target Group (required argument)'))
parser.add_argument(
'--port-id',
default='',
'--port-id', default='',
help=_('Neutron Port UUID'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of policy target to create'))
help=_('Name of Policy Target to create (required argument)'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -111,11 +109,7 @@ class UpdatePolicyTarget(neutronV20.UpdateCommand):
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('Description of the Policy Target'))
parser.add_argument(
'--policy-target-group', metavar='PTG',
default='',
help=_('Policy Target Group uuid'))
help=_('New description of the Policy Target'))
parser.add_argument(
'--name',
help=_('New name of the Policy Target'))
@ -125,11 +119,6 @@ class UpdatePolicyTarget(neutronV20.UpdateCommand):
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'tenant_id', 'description'])
if parsed_args.policy_target_group:
body[self.resource]['policy_target_group_id'] = \
neutronV20.find_resourceid_by_name_or_id(
self.get_client(), 'policy_target_group',
parsed_args.policy_target_group)
return body
@ -139,7 +128,7 @@ class ListPolicyTargetGroup(neutronV20.ListCommand):
resource = 'policy_target_group'
log = logging.getLogger(__name__ + '.ListPolicyTargetGroup')
list_columns = ['id', 'name', 'description']
list_columns = ['id', 'name', 'description', 'l2_policy_id', 'subnets']
pagination_support = True
sorting_support = True
@ -163,29 +152,28 @@ class CreatePolicyTargetGroup(neutronV20.CreateCommand):
help=_('Description of the Policy Target Group'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of Policy Target Group to create'))
help=_('Name of Policy Target Group to create '
'(required argument)'))
parser.add_argument(
'--l2-policy', metavar='L2_POLICY',
default='',
help=_('L2 policy uuid'))
help=_('L2 Policy UUID (if not specified, default is used)'))
parser.add_argument(
'--provided-policy-rule-sets', type=utils.str2dict,
# default={},
help=_('Dictionary of provided policy rule set uuids'))
help=_('Comma separated list of Policy Rule Sets'))
parser.add_argument(
'--consumed-policy-rule-sets', type=utils.str2dict,
# default={},
help=_('Dictionary of consumed policy rule set uuids'))
help=_('Comma separated list of Policy Rule Sets'))
parser.add_argument(
'--network-service-policy', metavar='NETWORK_SERVICE_POLICY',
default='',
help=_('Network service policy uuid'))
help=_('Network Service Policy'))
parser.add_argument(
'--subnets', type=string.split,
help=_('List of neutron subnet uuids'))
'--subnets', type=utils.str2list,
help=_('Comma separated list of Neutron Subnet UUIDs'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing, default is False'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -242,25 +230,28 @@ class UpdatePolicyTargetGroup(neutronV20.UpdateCommand):
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('Description of the Policy Target Group'))
help=_('New description of the Policy Target Group'))
parser.add_argument(
'--l2-policy', metavar='L2_POLICY',
help=_('L2 policy uuid'))
help=_('New L2 policy'))
parser.add_argument(
'--network-service-policy', metavar='NETWORK_SERVICE_POLICY',
help=_('Network Service Policy uuid'))
help=_('New Network Service Policy'))
parser.add_argument(
'--provided-policy-rule-sets', type=utils.str2dict,
help=_('Dictionary of provided policy rule set uuids'))
help=_('New comma separated list of Policy Rule Sets '
'(to unset use "")'))
parser.add_argument(
'--consumed-policy-rule-sets', type=utils.str2dict,
help=_('Dictionary of consumed policy rule set uuids'))
help=_('New comma separated list of Policy Rule Sets '
'(to unset use "")'))
parser.add_argument(
'--subnets', type=string.split,
help=_('List of neutron subnet uuids'))
'--subnets', type=utils.str2list,
help=_('New comma separated list of Neutron Subnet UUIDs '
'(to unset use "")'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing, default is False'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -271,7 +262,9 @@ class UpdatePolicyTargetGroup(neutronV20.UpdateCommand):
self.get_client(), 'l2_policy',
parsed_args.l2_policy)
if parsed_args.network_service_policy:
if parsed_args.network_service_policy == '':
body[self.resource]['network_service_policy_id'] = None
elif parsed_args.network_service_policy:
body[self.resource]['network_service_policy_id'] = \
neutronV20.find_resourceid_by_name_or_id(
self.get_client(), 'network_service_policy',
@ -307,7 +300,7 @@ class ListL2Policy(neutronV20.ListCommand):
resource = 'l2_policy'
log = logging.getLogger(__name__ + '.ListL2Policy')
_formatters = {}
list_columns = ['id', 'name', 'description', 'l3_policy_id']
list_columns = ['id', 'name', 'description', 'l3_policy_id', 'network_id']
pagination_support = True
sorting_support = True
@ -331,17 +324,19 @@ class CreateL2Policy(neutronV20.CreateCommand):
help=_('Description of the L2 Policy'))
parser.add_argument(
'--network',
help=_('Neutron network uuid to map the L2 Policy to'))
help=_('Neutron Network UUID to map the L2 Policy to '
'(if not specified, new Neutron Network is created '
'implicitly)'))
parser.add_argument(
'--l3-policy',
default='',
help=_('L3 Policy uuid'))
help=_('L3 Policy UUID (if not specified default is used)'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of L2 Policy to create'))
help=_('Name of L2 Policy to create (required argument)'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing, default is False'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -376,17 +371,17 @@ class UpdateL2Policy(neutronV20.UpdateCommand):
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('Description of the L2 Policy'))
help=_('New description of the L2 Policy'))
parser.add_argument(
'--l3-policy',
default='',
help=_('L3 Policy uuid'))
help=_('New L3 Policy'))
parser.add_argument(
'--name', metavar='NAME',
'--name',
help=_('New name of the L2 Policy'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -442,19 +437,25 @@ class CreateL3Policy(neutronV20.CreateCommand):
parser.add_argument(
'--subnet-prefix-length',
type=int,
# default=24,
help=_('Subnet prefix length, default is 24'))
parser.add_argument(
'--external-segment',
action='append', dest='external_segments', type=utils.str2dict,
help=_('Use format <ext-segment-id-1>=<ip-addr1:ipaddr2:...>'
# Note: The following format is also supported but we do not
# show it to avoid confusion
# help=_('Use format <ext-segment-id-1>=<ip-addr1:ipaddr2:...>'
# '(this option can be repeated)'))
help=_('Comma separated list of External Segments'
'(this option can be repeated)'))
parser.add_argument(
'--routers', type=utils.str2list,
help=_('Comma separated list of Neutron Router UUIDs'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of L3 policy to create'))
help=_('Name of L3 policy to create (required argument)'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing, default is False'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -465,14 +466,18 @@ class CreateL3Policy(neutronV20.CreateCommand):
external_segment_id = neutronV20.find_resourceid_by_name_or_id(
self.get_client(), 'external_segment',
external_segment.keys()[0])
ipaddrs = external_segment.itervalues().next().split(':')
ipaddrs = external_segment.itervalues().next()
if ipaddrs is "":
ipaddrs = []
else:
ipaddrs = external_segment.itervalues().next().split(':')
external_segments_dict[external_segment_id] = ipaddrs
body[self.resource]['external_segments'] = external_segments_dict
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'tenant_id', 'description',
'ip_version', 'ip_pool',
'ip_version', 'ip_pool', 'routers',
'subnet_prefix_length', 'shared'])
return body
@ -494,29 +499,29 @@ class UpdateL3Policy(neutronV20.UpdateCommand):
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('Description of the L3 Policy'))
parser.add_argument(
'--ip-version',
type=int,
help=_('IP version, default is 4'))
parser.add_argument(
'--ip-pool',
help=_('CIDR of IP pool to create, default is 10.0.0.0/8'))
help=_('New description of the L3 Policy'))
parser.add_argument(
'--subnet-prefix-length',
type=int,
help=_('Subnet prefix length, default is 24'))
help=_('New subnet prefix length'))
parser.add_argument(
'--external-segment',
action='append', dest='external_segments', type=utils.str2dict,
help=_('Use format <ext-segment-id-1>=<ip-addr1:ipaddr2:...>'
# Note: The following format is also supported but we do not
# show it to avoid confusion
# help=_('Use format <ext-segment-id-1>=<ip-addr1:ipaddr2:...>'
# '(this option can be repeated)'))
help=_('New comma separated list of External Segments'
'(this option can be repeated)'))
parser.add_argument(
'--name', metavar='NAME',
'--routers', type=utils.str2list,
help=_('New comma separated list of Neutron Router UUIDs'))
parser.add_argument(
'--name',
help=_('New name of the L3 Policy'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -529,14 +534,18 @@ class UpdateL3Policy(neutronV20.UpdateCommand):
external_segment_id = neutronV20.find_resourceid_by_name_or_id(
self.get_client(), 'external_segment',
external_segment.keys()[0])
ipaddrs = external_segment.itervalues().next().split(':')
ipaddrs = external_segment.itervalues().next()
if ipaddrs is "":
ipaddrs = []
else:
ipaddrs = external_segment.itervalues().next().split(':')
external_segments_dict[external_segment_id] = ipaddrs
body[self.resource]['external_segments'] = external_segments_dict
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'tenant_id', 'description',
'ip_version', 'ip_pool',
'ip_version', 'ip_pool', 'routers',
'subnet_prefix_length', 'shared'])
return body
@ -548,7 +557,8 @@ class ListNetworkServicePolicy(neutronV20.ListCommand):
resource = 'network_service_policy'
log = logging.getLogger(__name__ + '.ListNetworkServicePolicy')
_formatters = {'network_servie_params': _format_network_service_params}
list_columns = ['id', 'name', 'description', 'network_service_params']
list_columns = ['id', 'name', 'description', 'network_service_params',
'policy_target_groups']
pagination_support = True
sorting_support = True
@ -572,17 +582,18 @@ class CreateNetworkServicePolicy(neutronV20.CreateCommand):
help=_('Description of the network_service_policy'))
parser.add_argument(
'name',
help=_('Name of network_service_policy to create'))
help=_('Name of network_service_policy to create (required '
'argument)'))
parser.add_argument(
'--network-service-params',
metavar='type=PARAM_TYPE,name=PARAM_NAME,value=PARAM_VALUE',
action='append', dest='network_service_params',
type=utils.str2dict,
help=_('Network service params for this network service policy'
'(This option can be repeated).'))
help=_('Params for this Network Service '
'Policy (this option can be repeated)'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing, default is False'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -609,7 +620,7 @@ class UpdateNetworkServicePolicy(neutronV20.UpdateCommand):
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('Description of the network_service_policy'))
help=_('New description of the network_service_policy'))
parser.add_argument(
'--name',
help=_('New name of the network_service_policy'))
@ -618,11 +629,11 @@ class UpdateNetworkServicePolicy(neutronV20.UpdateCommand):
metavar='type=PARAM_TYPE,name=PARAM_NAME,value=PARAM_VALUE',
action='append', dest='network_service_params',
type=utils.str2dict,
help=_('Network service params for this network service policy'
'(This option can be repeated).'))
help=_('New params for this Network Service '
'Policy (this option can be repeated)'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -674,10 +685,10 @@ class CreatePolicyClassifier(neutronV20.CreateCommand):
help=_('Direction'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of classifier to create'))
help=_('Name of classifier to create (required argument)'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing, default is False'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -706,24 +717,24 @@ class UpdatePolicyClassifier(neutronV20.UpdateCommand):
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('Description of the policy classifier'))
help=_('New description of the policy classifier'))
parser.add_argument(
'--protocol',
choices=['tcp', 'udp', 'icmp'],
help=_('Protocol'))
help=_('New Protocol'))
parser.add_argument(
'--port-range',
help=_('Port range'))
help=_('New Port range'))
parser.add_argument(
'--direction',
choices=['in', 'out', 'bi', ''],
help=_('Direction'))
help=_('New Direction'))
parser.add_argument(
'--name',
help=_('New name of the classifier'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -772,10 +783,10 @@ class CreatePolicyAction(neutronV20.CreateCommand):
help=_('Name/UUID of servicechain spec for redirect action'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of action to create'))
help=_('Name of action to create (required argument)'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing, default is False'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -809,16 +820,16 @@ class UpdatePolicyAction(neutronV20.UpdateCommand):
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('Description of the policy action'))
help=_('New description of the policy action'))
parser.add_argument(
'--action-value',
help=_('Name/UUID of servicechain spec for redirect action'))
help=_('New name/UUID of servicechain spec for redirect action'))
parser.add_argument(
'--name',
help=_('New name of the action'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -867,19 +878,20 @@ class CreatePolicyRule(neutronV20.CreateCommand):
help=_('Description of the policy_rule'))
parser.add_argument(
'--enabled', type=bool,
help=_('Enable flag'))
help=_('Enable flag, default is True '
'(if False, this Policy Rule is ignored)'))
parser.add_argument(
'--classifier',
help=_('uuid of policy classifier'))
help=_('Policy Classifier (required argument)'))
parser.add_argument(
'--actions', type=string.split,
help=_('List of policy actions'))
'--actions', type=utils.str2list,
help=_('Comma separated list of Policy Action(s)'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of policy_rule to create'))
help=_('Name of Policy Rule to create (required argument)'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing, default is False'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -919,23 +931,32 @@ class UpdatePolicyRule(neutronV20.UpdateCommand):
log = logging.getLogger(__name__ + '.UpdatePolicyRule')
def add_known_arguments(self, parser):
parser.add_argument(
'--name',
help=_('New name of the Policy Rule'))
parser.add_argument(
'--description',
help=_('New description of the Policy Rule'))
parser.add_argument(
'--enabled', type=bool,
help=_('Enable flag'))
help=_('Enable flag (if False, this Policy Rule is ignored)'))
parser.add_argument(
'--classifier',
help=_('uuid of policy classifier'))
help=_('New Policy Classifier'))
parser.add_argument(
'--actions', type=string.split,
help=_('List of policy actions'))
'--actions', type=utils.str2list,
help=_('New comma separated list of Policy Actions '
'(to unset use "")'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
if parsed_args.actions:
if parsed_args.actions == []:
body[self.resource]['policy_actions'] = []
elif parsed_args.actions:
body[self.resource]['policy_actions'] = [
neutronV20.find_resourceid_by_name_or_id(
self.get_client(),
@ -961,7 +982,7 @@ class ListPolicyRuleSet(neutronV20.ListCommand):
resource = 'policy_rule_set'
log = logging.getLogger(__name__ + '.ListPolicyRuleSet')
_formatters = {}
list_columns = ['id', 'name', 'ploicy_rules']
list_columns = ['id', 'name', 'policy_rules']
pagination_support = True
sorting_support = True
@ -984,17 +1005,17 @@ class CreatePolicyRuleSet(neutronV20.CreateCommand):
'--description',
help=_('Description of the policy rule set'))
parser.add_argument(
'--policy-rules', type=string.split,
help=_('List of policy rules'))
'--policy-rules', type=utils.str2list,
help=_('Comma separated list of Policy Rules'))
parser.add_argument(
'--child-policy-rule-sets', type=string.split,
help=_('List of child policy rule sets'))
'--child-policy-rule-sets', type=utils.str2list,
help=_('Comma separated list of child Policy Rule Sets'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of policy rule set to create'))
help=_('Name of Policy Rule Set to create (required argument)'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing, default is False'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -1033,36 +1054,44 @@ class UpdatePolicyRuleSet(neutronV20.UpdateCommand):
def add_known_arguments(self, parser):
parser.add_argument(
'--policy-rules', type=string.split,
help=_('List of policy rules'))
'--name',
help=_('New name of the Policy Rule Set'))
parser.add_argument(
'--child-policy-rule-sets', type=string.split,
help=_('List of child policy rule sets'))
'--description',
help=_('New description of the Policy Rule Set'))
parser.add_argument(
'--policy-rules', type=utils.str2list,
help=_('New comma separated list of Policy Rules '
'(to unset use "")'))
parser.add_argument(
'--child-policy-rule-sets', type=utils.str2list,
help=_('New comma separated list of child Policy Rule Sets '
'(to unset use "")'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
if parsed_args.policy_rules:
if parsed_args.policy_rules == []:
body[self.resource]['policy_rules'] = []
elif parsed_args.policy_rules:
body[self.resource]['policy_rules'] = [
neutronV20.find_resourceid_by_name_or_id(
self.get_client(),
'policy_rule',
elem) for elem in parsed_args.policy_rules]
parsed_args.policy_rules = body[self.resource]['policy_rules']
if parsed_args.child_policy_rule_sets:
if parsed_args.child_policy_rule_sets == []:
body[self.resource]['child_policy_rule_sets'] = []
elif parsed_args.child_policy_rule_sets:
body[self.resource]['child_policy_rule_sets'] = [
neutronV20.find_resourceid_by_name_or_id(
self.get_client(),
'policy_rule_set',
elem) for elem in parsed_args.child_policy_rule_sets]
parsed_args.child_policy_rule_sets = (
parsed_args.child_policy_rule_sets)
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'description', 'policy_rules',
'child_policy_rule_sets', 'shared'])
['name', 'description', 'shared'])
return body
@ -1071,7 +1100,7 @@ class ListExternalPolicy(neutronV20.ListCommand):
resource = 'external_policy'
log = logging.getLogger(__name__ + '.ListExternalPolicy')
list_columns = ['id', 'name', 'description', 'shared']
list_columns = ['id', 'name', 'description', 'shared', 'external_segments']
pagination_support = True
sorting_support = True
@ -1095,19 +1124,19 @@ class CreateExternalPolicy(neutronV20.CreateCommand):
help=_('Description of the External Policy'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of External Policy to create'))
help=_('Name of External Policy to create (required argument)'))
parser.add_argument(
'--external-segments', type=string.split,
help=_('List of External Segment uuids'))
'--external-segments', type=utils.str2list,
help=_('Comma separated list of External Segments'))
parser.add_argument(
'--provided-policy-rule-sets', type=utils.str2dict,
help=_('Dictionary of provided policy rule set uuids'))
help=_('Comma separated list of Policy Rule Sets'))
parser.add_argument(
'--consumed-policy-rule-sets', type=utils.str2dict,
help=_('Dictionary of consumed policy rule set uuids'))
help=_('Comma separated list of Policy Rule Sets'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing, default is False'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -1159,22 +1188,25 @@ class UpdateExternalPolicy(neutronV20.UpdateCommand):
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('Description of the External Policy'))
help=_('New description of the External Policy'))
parser.add_argument(
'--name',
help=_('New name of the External Policy'))
parser.add_argument(
'--external-segments', type=string.split,
help=_('List of External Segment uuids'))
'--external-segments', type=utils.str2list,
help=_('New comma separated list of External Segments '
'(to unset use "")'))
parser.add_argument(
'--provided-policy-rule-sets', type=utils.str2dict,
help=_('Dictionary of provided policy rule set uuids'))
help=_('New comma separated list of Policy Rule Sets '
'(to unset use "")'))
parser.add_argument(
'--consumed-policy-rule-sets', type=utils.str2dict,
help=_('Dictionary of consumed policy rule set uuids'))
help=_('New comma separated list of Policy Rule Sets '
'(to unset use "")'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -1195,7 +1227,9 @@ class UpdateExternalPolicy(neutronV20.UpdateCommand):
parsed_args.consumed_policy_rule_sets[id_key] = (
parsed_args.consumed_policy_rule_sets.pop(key))
if parsed_args.external_segments:
if parsed_args.external_segments == []:
body[self.resource]['external_segments'] = []
elif parsed_args.external_segments:
body[self.resource]['external_segments'] = [
neutronV20.find_resourceid_by_name_or_id(
self.get_client(),
@ -1240,8 +1274,8 @@ class CreateExternalSegment(neutronV20.CreateCommand):
'--description',
help=_('Description of the External Segment'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of External Segment to create'))
'name',
help=_('Name of External Segment to create (required argument)'))
parser.add_argument(
'--ip-version',
type=int, choices=[4, 6],
@ -1252,26 +1286,31 @@ class CreateExternalSegment(neutronV20.CreateCommand):
parser.add_argument(
'--external-route', metavar='destination=CIDR,nexthop=IP_ADDR',
action='append', dest='external_routes', type=utils.str2dict,
help=_('External route (This option can be repeated).'))
help=_('If no nexthop, use format: destination=CIDR,nexthop '
'(this option can be repeated)'))
parser.add_argument(
'--port-address-translation', type=bool,
help=_('Perform port-based address translation, default is False'))
help=_('Enable port-based address translation, default is False'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing, default is False'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
if parsed_args.external_routes:
body['external_segment']['external_routes'] = (
parsed_args.external_routes)
eroutes = []
for er in parsed_args.external_routes:
if 'nexthop' in er and er['nexthop'] == '':
er['nexthop'] = None
if er:
eroutes.append(er)
body['external_segment']['external_routes'] = eroutes
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'tenant_id', 'description',
'ip_version', 'cidr',
'external_routes', 'port_address_translation',
'shared'])
'port_address_translation', 'shared'])
return body
@ -1292,40 +1331,40 @@ class UpdateExternalSegment(neutronV20.UpdateCommand):
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('Description of the External Segment'))
help=_('New description of the External Segment'))
parser.add_argument(
'--name',
help=_('New name of External Segment'))
parser.add_argument(
'--ip-version',
type=int, choices=[4, 6],
help=_('IP version, default is 4'))
parser.add_argument(
'--cidr',
help=_('CIDR of External Segment, default is 172.16.0.0/12'))
parser.add_argument(
'--external-route', metavar='destination=CIDR,nexthop=IP_ADDR',
action='append', dest='external_routes', type=utils.str2dict,
help=_('External route (This option can be repeated).'))
help=_('If no nexthop, use format: destination=CIDR,nexthop '
'(this option can be repeated)'))
parser.add_argument(
'--port-address-translation', type=bool,
help=_('Perform port-based address translation, default is False'))
help=_('Enable or disable port-based address translation'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
if parsed_args.external_routes:
body['external_segment']['external_routes'] = (
parsed_args.external_routes)
if parsed_args.external_routes == [{}]:
body[self.resource]['external_routes'] = []
elif parsed_args.external_routes:
eroutes = []
for er in parsed_args.external_routes:
if 'nexthop' in er and er['nexthop'] == '':
er['nexthop'] = None
if er:
eroutes.append(er)
body[self.resource]['external_routes'] = eroutes
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'tenant_id', 'description',
'ip_version', 'cidr',
'external_routes', 'port_address_translation',
'shared'])
'port_address_translation', 'shared'])
return body
@ -1360,7 +1399,7 @@ class CreateNatPool(neutronV20.CreateCommand):
help=_('Description of the NAT Pool'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of NAT Pool to create'))
help=_('Name of NAT Pool to create (required argument)'))
parser.add_argument(
'--ip-version',
type=int, choices=[4, 6],
@ -1370,10 +1409,10 @@ class CreateNatPool(neutronV20.CreateCommand):
help=_('CIDR for NAT Pool'))
parser.add_argument(
'--external-segment',
help=_('External Segment name or UUID'))
help=_('External Segment'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing, default is False'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
@ -1407,23 +1446,16 @@ class UpdateNatPool(neutronV20.UpdateCommand):
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('Description of the NAT Pool'))
help=_('New description of the NAT Pool'))
parser.add_argument(
'--name',
help=_('New name of NAT Pool'))
parser.add_argument(
'--ip-version',
type=int, choices=[4, 6],
help=_('IP version, default is 4'))
parser.add_argument(
'--ip-pool',
help=_('CIDR for NAT Pool'))
parser.add_argument(
'--external-segment',
help=_('External Segment name or UUID'))
help=_('New External Segment'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
help=_('Enable or disable resource sharing'))
def args2body(self, parsed_args):
body = {self.resource: {}, }

25
gbpclient/gbp/v2_0/servicechain.py
View File

@ -17,7 +17,6 @@
import json
import logging
import os
import string
from heatclient.common import template_utils
@ -25,6 +24,8 @@ from neutronclient.common import exceptions as exc
from neutronclient.i18n import _
from neutronclient.neutron import v2_0 as neutronV20
from gbpclient.common import utils
class ListServiceChainInstance(neutronV20.ListCommand):
"""List service chain instances that belong to a given tenant."""
@ -90,8 +91,7 @@ class CreateServiceChainInstance(neutronV20.CreateCommand):
parsed_args.consumer_ptg)
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'tenant_id', 'description',
'servicechain_spec', 'provider_ptg',
'consumer_ptg', 'param_values'])
'param_values'])
return body
@ -134,9 +134,7 @@ class UpdateServiceChainInstance(neutronV20.UpdateCommand):
self.get_client(), 'policy_target_group',
parsed_args.consumer_ptg)
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'description',
'servicechain_spec', 'provider_ptg',
'consumer_ptg', 'param_values'])
['name', 'description', 'param_values'])
return body
@ -394,8 +392,8 @@ class CreateServiceChainSpec(neutronV20.CreateCommand):
'--description',
help=_('Description of the Service Chain Specification.'))
parser.add_argument(
'--nodes', metavar='NODES', type=string.split,
help=_('Service Chain Node ID or name of the Service Chain Node'))
'--nodes', metavar='NODES', type=utils.str2list,
help=_('Comma separated list of Service Chain Nodes'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
@ -423,16 +421,19 @@ class UpdateServiceChainSpec(neutronV20.UpdateCommand):
def add_known_arguments(self, parser):
parser.add_argument(
'--nodes', type=string.split,
help=_('List of Service Chain Node IDs or names of the Service '
'Chain Nodes'))
'--nodes', type=utils.str2list,
help=_('New comma separated list of Service Chain Nodes '
'(to unset use "")'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
if parsed_args.nodes:
if parsed_args.nodes == []:
body[self.resource]['nodes'] = []
elif parsed_args.nodes:
body[self.resource]['nodes'] = [
neutronV20.find_resourceid_by_name_or_id(
self.get_client(),

90
gbpclient/gbpshell.py
View File

@ -182,6 +182,96 @@ COMMAND_V2 = {
'servicechain-instance-update': (
servicechain.UpdateServiceChainInstance
),
'pt-create': gbp.CreatePolicyTarget,
'pt-delete': gbp.DeletePolicyTarget,
'pt-update': gbp.UpdatePolicyTarget,
'pt-list': gbp.ListPolicyTarget,
'pt-show': gbp.ShowPolicyTarget,
'ptg-create': gbp.CreatePolicyTargetGroup,
'ptg-delete': gbp.DeletePolicyTargetGroup,
'ptg-update': gbp.UpdatePolicyTargetGroup,
'ptg-list': gbp.ListPolicyTargetGroup,
'ptg-show': gbp.ShowPolicyTargetGroup,
'l2p-create': gbp.CreateL2Policy,
'l2p-delete': gbp.DeleteL2Policy,
'l2p-update': gbp.UpdateL2Policy,
'l2p-list': gbp.ListL2Policy,
'l2p-show': gbp.ShowL2Policy,
'l3p-create': gbp.CreateL3Policy,
'l3p-delete': gbp.DeleteL3Policy,
'l3p-update': gbp.UpdateL3Policy,
'l3p-list': gbp.ListL3Policy,
'l3p-show': gbp.ShowL3Policy,
'nsp-create': gbp.CreateNetworkServicePolicy,
'nsp-delete': gbp.DeleteNetworkServicePolicy,
'nsp-update': gbp.UpdateNetworkServicePolicy,
'nsp-list': gbp.ListNetworkServicePolicy,
'nsp-show': gbp.ShowNetworkServicePolicy,
'ep-create': gbp.CreateExternalPolicy,
'ep-delete': gbp.DeleteExternalPolicy,
'ep-update': gbp.UpdateExternalPolicy,
'ep-list': gbp.ListExternalPolicy,
'ep-show': gbp.ShowExternalPolicy,
'es-create': gbp.CreateExternalSegment,
'es-delete': gbp.DeleteExternalSegment,
'es-update': gbp.UpdateExternalSegment,
'es-list': gbp.ListExternalSegment,
'es-show': gbp.ShowExternalSegment,
'np-create': gbp.CreateNatPool,
'np-delete': gbp.DeleteNatPool,
'np-update': gbp.UpdateNatPool,
'np-list': gbp.ListNatPool,
'np-show': gbp.ShowNatPool,
'pc-create': gbp.CreatePolicyClassifier,
'pc-delete': gbp.DeletePolicyClassifier,
'pc-update': gbp.UpdatePolicyClassifier,
'pc-list': gbp.ListPolicyClassifier,
'pc-show': gbp.ShowPolicyClassifier,
'pa-create': gbp.CreatePolicyAction,
'pa-delete': gbp.DeletePolicyAction,
'pa-update': gbp.UpdatePolicyAction,
'pa-list': gbp.ListPolicyAction,
'pa-show': gbp.ShowPolicyAction,
'pr-create': gbp.CreatePolicyRule,
'pr-delete': gbp.DeletePolicyRule,
'pr-update': gbp.UpdatePolicyRule,
'pr-list': gbp.ListPolicyRule,
'pr-show': gbp.ShowPolicyRule,
'prs-create': gbp.CreatePolicyRuleSet,
'prs-delete': gbp.DeletePolicyRuleSet,
'prs-update': gbp.UpdatePolicyRuleSet,
'prs-list': gbp.ListPolicyRuleSet,
'prs-show': gbp.ShowPolicyRuleSet,
'sp-list': servicechain.ListServiceProfile,
'sp-show': servicechain.ShowServiceProfile,
'sp-create': servicechain.CreateServiceProfile,
'sp-delete': servicechain.DeleteServiceProfile,
'sp-update': servicechain.UpdateServiceProfile,
'scn-list': servicechain.ListServiceChainNode,
'scn-show': servicechain.ShowServiceChainNode,
'scn-create': servicechain.CreateServiceChainNode,
'scn-delete': servicechain.DeleteServiceChainNode,
'scn-update': servicechain.UpdateServiceChainNode,
'scs-list': servicechain.ListServiceChainSpec,
'scs-show': servicechain.ShowServiceChainSpec,
'scs-create': servicechain.CreateServiceChainSpec,
'scs-delete': servicechain.DeleteServiceChainSpec,
'scs-update': servicechain.UpdateServiceChainSpec,
'sci-list': (
servicechain.ListServiceChainInstance
),
'sci-show': (
servicechain.ShowServiceChainInstance
),
'sci-create': (
servicechain.CreateServiceChainInstance
),
'sci-delete': (
servicechain.DeleteServiceChainInstance
),
'sci-update': (
servicechain.UpdateServiceChainInstance
),
}
COMMANDS = {'2.0': COMMAND_V2}

28
gbpclient/tests/unit/test_cli20_externalpolicy.py
View File

@ -50,7 +50,7 @@ class CLITestV20ExternalPolicyJSON(test_cli20.CLITestV20Base):
my_id = 'someid'
provided_policy_rule_sets = "prs1=true,prs2=true"
consumed_policy_rule_sets = "prs3=true,prs4=true"
external_segments = "ES1 ES2"
external_segments = "ES1,ES2"
shared = 'True'
args = ['--tenant-id', tenant_id,
'--description', description,
@ -103,7 +103,7 @@ class CLITestV20ExternalPolicyJSON(test_cli20.CLITestV20Base):
my_id = 'someid'
provided_policy_rule_sets = "prs1=true,prs2=true"
consumed_policy_rule_sets = "prs3=true,prs4=true"
external_segments = "ES1 ES2"
external_segments = "ES1,ES2"
shared = 'True'
args = ['--name', name,
'--description', description,
@ -122,6 +122,30 @@ class CLITestV20ExternalPolicyJSON(test_cli20.CLITestV20Base):
}
self._test_update_resource(resource, cmd, my_id, args, params)
def test_update_external_policy_unset_external_segment(self):
resource = 'external_policy'
cmd = gbp.UpdateExternalPolicy(test_cli20.MyApp(sys.stdout), None)
my_id = 'someid'
external_segments = ""
args = ['--external-segments', external_segments, my_id]
params = {'external_segments': []}
self._test_update_resource(resource, cmd, my_id, args, params)
def test_update_external_policy_unset_prs(self):
resource = 'external_policy'
cmd = gbp.UpdateExternalPolicy(test_cli20.MyApp(sys.stdout), None)
my_id = 'someid'
provided_policy_rule_sets = ""
consumed_policy_rule_sets = ""
args = ['--provided-policy-rule-sets', provided_policy_rule_sets,
'--consumed-policy-rule-sets', consumed_policy_rule_sets,
my_id]
params = {
'provided_policy_rule_sets': {},
'consumed_policy_rule_sets': {},
}
self._test_update_resource(resource, cmd, my_id, args, params)
def test_delete_external_policy_name(self):
"""external-policy-delete."""
resource = 'external_policy'

53
gbpclient/tests/unit/test_cli20_externalsegment.py
View File

@ -75,6 +75,26 @@ class CLITestV20ExternalSegmentJSON(test_cli20.CLITestV20Base):
port_address_translation=True,
shared=True)
def test_create_external_segment_with_external_route_no_nexthop(self):
"""external-segment-create with all params."""
resource = 'external_segment'
cmd = gbp.CreateExternalSegment(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
tenant_id = 'mytenant'
my_id = 'someid'
external_route = 'destination=172.16.1.0/24,nexthop'
expected_external_routes = [{'destination': '172.16.1.0/24', 'nexthop':
None}]
args = ['--tenant-id', tenant_id,
'--external-route', external_route,
name]
position_names = ['name', ]
position_values = [name, ]
self._test_create_resource(resource, cmd, name, my_id, args,
position_names, position_values,
tenant_id=tenant_id,
external_routes=expected_external_routes)
def test_list_external_segments(self):
"""external-segment-list."""
resource = 'external_segments'
@ -103,8 +123,6 @@ class CLITestV20ExternalSegmentJSON(test_cli20.CLITestV20Base):
name = 'myname'
description = 'My External Segment'
my_id = 'someid'
ip_version = '4'
cidr = '192.168.0.0/24'
external_route = 'destination=172.16.1.0/24,nexthop=192.168.0.10'
expected_external_routes = [{'destination': '172.16.1.0/24', 'nexthop':
'192.168.0.10'}]
@ -112,8 +130,6 @@ class CLITestV20ExternalSegmentJSON(test_cli20.CLITestV20Base):
shared = 'True'
args = ['--name', name,
'--description', description,
'--ip-version', ip_version,
'--cidr', cidr,
'--external-route', external_route,
'--port-address-translation', port_address_translation,
'--shared', shared,
@ -121,14 +137,39 @@ class CLITestV20ExternalSegmentJSON(test_cli20.CLITestV20Base):
params = {
'name': name,
'description': description,
'ip_version': 4,
'cidr': cidr,
'external_routes': expected_external_routes,
'port_address_translation': True,
'shared': True
}
self._test_update_resource(resource, cmd, my_id, args, params)
def test_update_external_segment_with_external_route_no_nexthop(self):
resource = 'external_segment'
cmd = gbp.UpdateExternalSegment(test_cli20.MyApp(sys.stdout), None)
my_id = 'someid'
external_route = 'destination=172.16.1.0/24,nexthop'
expected_external_routes = [{'destination': '172.16.1.0/24', 'nexthop':
None}]
args = ['--external-route', external_route,
my_id]
params = {
'external_routes': expected_external_routes,
}
self._test_update_resource(resource, cmd, my_id, args, params)
def test_update_external_segment_with_unset_external_route(self):
resource = 'external_segment'
cmd = gbp.UpdateExternalSegment(test_cli20.MyApp(sys.stdout), None)
my_id = 'someid'
external_route = ''
expected_external_routes = []
args = ['--external-route', external_route,
my_id]
params = {
'external_routes': expected_external_routes,
}
self._test_update_resource(resource, cmd, my_id, args, params)
def test_delete_external_segment_name(self):
"""external-segment-delete."""
resource = 'external_segment'

51
gbpclient/tests/unit/test_cli20_l3policy.py
View File

@ -52,6 +52,7 @@ class CLITestV20L3PolicyJSON(test_cli20.CLITestV20Base):
subnet_prefix_length = '24'
external_segment = 'seg_uuid1=1.1.1.0:2.2.2.0'
expected_external_segments = {'seg_uuid1': ['1.1.1.0', '2.2.2.0']}
routers = 'uuid1,uuid2'
shared = 'True'
args = ['--tenant-id', tenant_id,
'--description', description,
@ -59,6 +60,7 @@ class CLITestV20L3PolicyJSON(test_cli20.CLITestV20Base):
'--ip-pool', ip_pool,
'--subnet-prefix-length', subnet_prefix_length,
'--external-segment', external_segment,
'--routers', routers,
'--shared', shared,
name]
position_names = ['name', ]
@ -70,9 +72,30 @@ class CLITestV20L3PolicyJSON(test_cli20.CLITestV20Base):
ip_version=4,
ip_pool=ip_pool,
subnet_prefix_length=24,
routers=['uuid1', 'uuid2'],
external_segments=
expected_external_segments, shared=True)
def test_create_l3_policy_with_external_segment(self):
"""l3-policy-create with all params."""
resource = 'l3_policy'
cmd = gbp.CreateL3Policy(test_cli20.MyApp(sys.stdout), None)
name = 'name'
tenant_id = 'mytenant'
my_id = 'someid'
external_segment = 'seg_uuid1'
expected_external_segments = {'seg_uuid1': []}
args = ['--tenant-id', tenant_id,
'--external-segment', external_segment,
name]
position_names = ['name', ]
position_values = [name, ]
self._test_create_resource(resource, cmd, name, my_id, args,
position_names, position_values,
tenant_id=tenant_id,
external_segments=
expected_external_segments)
def test_list_l3_policies(self):
resource = 'l3_policies'
cmd = gbp.ListL3Policy(test_cli20.MyApp(sys.stdout), None)
@ -98,27 +121,25 @@ class CLITestV20L3PolicyJSON(test_cli20.CLITestV20Base):
name = 'myname'
description = 'My L3 Policy'
my_id = 'someid'
ip_version = '4'
ip_pool = '172.16.0.0/12'
subnet_prefix_length = '24'
external_segment = 'seg_uuid1=1.1.1.0:2.2.2.0'
expected_external_segments = {'seg_uuid1': ['1.1.1.0', '2.2.2.0']}
shared = 'True'
routers = 'uuid1,uuid2'
args = ['--name', name,
'--description', description,
'--ip-version', ip_version,
'--ip-pool', ip_pool,
'--subnet-prefix-length', subnet_prefix_length,
'--external-segment', external_segment,
'--routers', routers,
'--shared', shared,
my_id]
params = {
'name': name,
'description': description,
'ip_version': 4,
'ip_pool': ip_pool,
'subnet_prefix_length': 24,
'external_segments': expected_external_segments,
'routers': routers,
'routers': ['uuid1', 'uuid2'],
'shared': True
}
self._test_update_resource(resource, cmd, my_id, args, params)
@ -129,28 +150,34 @@ class CLITestV20L3PolicyJSON(test_cli20.CLITestV20Base):
name = 'myname'
description = 'My L3 Policy'
my_id = 'someid'
ip_version = '4'
ip_pool = '172.16.0.0/12'
subnet_prefix_length = '24'
external_segment = ''
expected_external_segments = {}
args = ['--name', name,
'--description', description,
'--ip-version', ip_version,
'--ip-pool', ip_pool,
'--subnet-prefix-length', subnet_prefix_length,
'--external-segment', external_segment,
my_id]
params = {
'name': name,
'description': description,
'ip_version': 4,
'ip_pool': ip_pool,
'subnet_prefix_length': 24,
'external_segments': expected_external_segments,
}
self._test_update_resource(resource, cmd, my_id, args, params)
def test_update_l3_policy_unset_routers(self):
resource = 'l3_policy'
cmd = gbp.UpdateL3Policy(test_cli20.MyApp(sys.stdout), None)
my_id = 'someid'
routers = ''
args = ['--routers', routers,
my_id]
params = {
'routers': [],
}
self._test_update_resource(resource, cmd, my_id, args, params)
def test_delete_l3_policy_name(self):
resource = 'l3_policy'
cmd = gbp.DeleteL3Policy(test_cli20.MyApp(sys.stdout), None)

6
gbpclient/tests/unit/test_cli20_natpool.py
View File

@ -98,22 +98,16 @@ class CLITestV20NatPoolJSON(test_cli20.CLITestV20Base):
name = 'myname'
description = 'My Nat Pool'
my_id = 'someid'
ip_version = '4'
ip_pool = '192.168.0.0/24'
external_segment_id = "segmentid"
shared = 'True'
args = ['--name', name,
'--description', description,
'--ip-version', ip_version,
'--ip-pool', ip_pool,
'--external-segment', external_segment_id,
'--shared', shared,
my_id]
params = {
'name': name,
'description': description,
'ip_version': 4,
'ip_pool': ip_pool,
'external_segment_id': external_segment_id,
'shared': True
}

14
gbpclient/tests/unit/test_cli20_policyrule.py
View File

@ -47,7 +47,7 @@ class CLITestV20PolicyRuleJSON(test_cli20.CLITestV20Base):
enabled = True
policy_classifier_id = 'pc-id'
policy_actions_res = ["pa1", "pa2"]
policy_actions_arg = "pa1 pa2"
policy_actions_arg = "pa1,pa2"
shared = 'True'
args = ['--tenant-id', tenant_id,
'--description', description,
@ -124,7 +124,7 @@ class CLITestV20PolicyRuleJSON(test_cli20.CLITestV20Base):
enabled = True
policy_classifier_id = 'pc-id'
policy_actions_res = ["pa1", "pa2"]
policy_actions_arg = "pa1 pa2"
policy_actions_arg = "pa1,pa2"
my_id = 'someid'
shared = 'True'
cmd = gbp.UpdatePolicyRule(test_cli20.MyApp(sys.stdout), None)
@ -141,6 +141,16 @@ class CLITestV20PolicyRuleJSON(test_cli20.CLITestV20Base):
'--shared', shared, ]
self._test_update_resource(resource, cmd, my_id, args, body)
def test_update_policy_rule_unset_actions(self):
resource = 'policy_rule'
policy_actions_res = []
policy_actions_arg = ""
my_id = 'someid'
cmd = gbp.UpdatePolicyRule(test_cli20.MyApp(sys.stdout), None)
body = {'policy_actions': policy_actions_res}
args = [my_id, '--actions', policy_actions_arg]
self._test_update_resource(resource, cmd, my_id, args, body)
def test_delete_policy_classifier(self):
"""grouppolicy-policy-rule-delete my-id."""
resource = 'policy_rule'

28
gbpclient/tests/unit/test_cli20_policytargetgroup.py
View File

@ -138,6 +138,34 @@ class CLITestV20PolicyTargetGroupJSON(test_cli20.CLITestV20Base):
}
self._test_update_resource(resource, cmd, my_id, args, params)
def test_update_policy_target_group_unset_prs(self):
"""policy-target-group-update."""
resource = 'policy_target_group'
cmd = gbp.UpdatePolicyTargetGroup(test_cli20.MyApp(sys.stdout), None)
my_id = 'my-id'
provided_prs = ""
consumed_prs = ""
args = [my_id,
'--provided-policy-rule-sets', provided_prs,
'--consumed-policy-rule-sets', consumed_prs
]
params = {
'provided_policy_rule_sets': {},
'consumed_policy_rule_sets': {}
}
self._test_update_resource(resource, cmd, my_id, args, params)
def test_update_policy_target_group_unset_nsp(self):
"""policy-target-group-update."""
resource = 'policy_target_group'
cmd = gbp.UpdatePolicyTargetGroup(test_cli20.MyApp(sys.stdout), None)
my_id = 'my-id'
network_service_policy_id = ''
args = [my_id,
'--network-service-policy', network_service_policy_id]
params = {'network_service_policy_id': None}
self._test_update_resource(resource, cmd, my_id, args, params)
def test_delete_policy_target_group_name(self):
"""policy-target-group-delete."""
resource = 'policy_target_group'

14
gbpclient/tests/unit/test_cli20_servicechain_spec.py
View File

@ -45,7 +45,7 @@ class CLITestV20ServiceChainSpecJSON(test_cli20.CLITestV20Base):
cmd = servicechain.CreateServiceChainSpec(test_cli20.MyApp(sys.stdout),
None)
name = 'my-name'
nodes_arg = 'node1 node2'
nodes_arg = 'node1,node2'
nodes_res = ['node1', 'node2']
tenant_id = 'my-tenant'
description = 'My Service Chain Spec'
@ -126,7 +126,7 @@ class CLITestV20ServiceChainSpecJSON(test_cli20.CLITestV20Base):
resource = 'servicechain_spec'
cmd = servicechain.UpdateServiceChainSpec(test_cli20.MyApp(sys.stdout),
None)
nodes_arg = 'node1 node2'
nodes_arg = 'node1,node2'
nodes_res = ['node1', 'node2']
body = {
'name': 'new_name',
@ -140,6 +140,16 @@ class CLITestV20ServiceChainSpecJSON(test_cli20.CLITestV20Base):
'--shared', 'True']
self._test_update_resource(resource, cmd, 'myid', args, body)
def test_update_servicechain_node_unset_nodes(self):
resource = 'servicechain_spec'
cmd = servicechain.UpdateServiceChainSpec(test_cli20.MyApp(sys.stdout),
None)
nodes_arg = ''
nodes_res = []
body = {'nodes': nodes_res}
args = ['myid', '--nodes', nodes_arg]
self._test_update_resource(resource, cmd, 'myid', args, body)
def test_delete_servicechain_spec(self):
"""service-chain-spec-delete my-id."""
resource = 'servicechain_spec'

40
gbpclient/tests/unit/test_utils.py
View File

@ -37,3 +37,43 @@ class TestUtils(testtools.TestCase):
input_str = None
expected = {}
self.assertEqual(expected, utils.str2dict(input_str))
def test_string_to_list(self):
input_str = 'key1'
expected = ['key1']
self.assertEqual(expected, utils.str2list(input_str))
input_str = 'key1, key2'
expected = ['key1', 'key2']
self.assertEqual(expected, utils.str2list(input_str))
input_str = 'key1,key2'
expected = ['key1', 'key2']
self.assertEqual(expected, utils.str2list(input_str))
input_str = 'key1,key2,'
expected = ['key1', 'key2']
self.assertEqual(expected, utils.str2list(input_str))
input_str = 'key1,key2,'
expected = ['key1', 'key2']
self.assertEqual(expected, utils.str2list(input_str))
input_str = ',key1,key2 '
expected = ['key1', 'key2']
self.assertEqual(expected, utils.str2list(input_str))
input_str = 'key1 key2'
expected = ['key1', 'key2']
self.assertEqual(expected, utils.str2list(input_str))
input_str = ' key1 key2 '
expected = ['key1', 'key2']
self.assertEqual(expected, utils.str2list(input_str))
input_str = 'key1 key2, key3 '
expected = ['key1', 'key2', 'key3']
self.assertEqual(expected, utils.str2list(input_str))
input_str = ' , key1 key2, , key3 '
expected = ['key1', 'key2', 'key3']
self.assertEqual(expected, utils.str2list(input_str))
def test_none_string_to_list(self):
input_str = ''
expected = []
self.assertEqual(expected, utils.str2list(input_str))
input_str = None
expected = []
self.assertEqual(expected, utils.str2list(input_str))