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

Add support for External Access model

Browse Source 
Client and CLI for:
External Policies
External Segments
Nat Pools

Also, updates to L3 Policies.

Partially Implements: blueprint external-connectivity

Change-Id: Ia05d21a7032baeaf6739f047d7f5eaa11e70990a
This commit is contained in:
Sumit Naiksatam 2014-12-14 03:29:05 -08:00
parent f264be8652
commit 7d6eeb857e
7 changed files with 1015 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -28,6 +28,14 @@ def _format_network_service_params(net_svc_policy):
return ''
def _format_host_routes(subnet):
try:
return '\n'.join([jsonutils.dumps(route) for route in
subnet['host_routes']])
except (TypeError, KeyError):
return ''
class ListPolicyTarget(neutronV20.ListCommand):
"""List policy_targets that belong to a given tenant."""
@ -389,6 +397,11 @@ class CreateL3Policy(neutronV20.CreateCommand):
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:...>'
'(this option can be repeated)'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of L3 policy to create'))
@ -396,6 +409,17 @@ class CreateL3Policy(neutronV20.CreateCommand):
def args2body(self, parsed_args):
body = {self.resource: {}, }
if parsed_args.external_segments:
external_segments_dict = {}
for external_segment in parsed_args.external_segments:
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(':')
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',
@ -417,6 +441,51 @@ class UpdateL3Policy(neutronV20.UpdateCommand):
resource = 'l3_policy'
log = logging.getLogger(__name__ + '.UpdateL3Policy')
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'))
parser.add_argument(
'--subnet-prefix-length',
type=int,
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:...>'
'(this option can be repeated)'))
parser.add_argument(
'--name', metavar='NAME',
help=_('Name of L3 policy to create'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
if parsed_args.external_segments:
external_segments_dict = {}
for external_segment in parsed_args.external_segments:
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(':')
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',
'subnet_prefix_length'])
return body
class ListNetworkServicePolicy(neutronV20.ListCommand):
"""List Network Service Policies that belong to a given tenant."""
@ -497,7 +566,7 @@ class UpdateNetworkServicePolicy(neutronV20.UpdateCommand):
help=_('Description of the network_service_policy'))
parser.add_argument(
'--name',
help=_('Name of network_service_policy to create'))
help=_('New name of the network_service_policy'))
parser.add_argument(
'--network-service-params',
metavar='type=PARAM_TYPE,name=PARAM_NAME,value=PARAM_VALUE',
@ -860,3 +929,380 @@ class UpdatePolicyRuleSet(neutronV20.UpdateCommand):
['name', 'description', 'policy_rules',
'child_policy_rule_sets'])
return body
class ListExternalPolicy(neutronV20.ListCommand):
"""List External Policies that belong to a given tenant."""
resource = 'external_policy'
log = logging.getLogger(__name__ + '.ListExternalPolicy')
list_columns = ['id', 'name', 'description', 'shared']
pagination_support = True
sorting_support = True
class ShowExternalPolicy(neutronV20.ShowCommand):
"""Show information of a given External Policy."""
resource = 'external_policy'
log = logging.getLogger(__name__ + '.ShowExternalPolicy')
class CreateExternalPolicy(neutronV20.CreateCommand):
"""Create a External Policy for a given tenant."""
resource = 'external_policy'
log = logging.getLogger(__name__ + '.CreateExternalPolicy')
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('Description of the External Policy'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of External Policy to create'))
parser.add_argument(
'--external-segments', type=string.split,
help=_('List of External Segment uuids'))
parser.add_argument(
'--provided-policy-rule-sets', type=utils.str2dict,
help=_('Dictionary of provided policy rule set uuids'))
parser.add_argument(
'--consumed-policy-rule-sets', type=utils.str2dict,
help=_('Dictionary of consumed policy rule set uuids'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
if parsed_args.provided_policy_rule_sets:
for key in parsed_args.provided_policy_rule_sets.keys():
id_key = neutronV20.find_resourceid_by_name_or_id(
self.get_client(), 'policy_rule_set',
key)
parsed_args.provided_policy_rule_sets[id_key] = (
parsed_args.provided_policy_rule_sets.pop(key))
if parsed_args.consumed_policy_rule_sets:
for key in parsed_args.consumed_policy_rule_sets.keys():
id_key = neutronV20.find_resourceid_by_name_or_id(
self.get_client(), 'policy_rule_set',
key)
parsed_args.consumed_policy_rule_sets[id_key] = (
parsed_args.consumed_policy_rule_sets.pop(key))
if parsed_args.external_segments:
for external_segment in parsed_args.external_segments:
external_segment_id = neutronV20.find_resourceid_by_name_or_id(
self.get_client(), 'external_segment', external_segment)
parsed_args.external_segments.remove(external_segment)
parsed_args.external_segments.append(external_segment_id)
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'tenant_id', 'description',
'provided_policy_rule_sets',
'external_segments',
'consumed_policy_rule_sets', 'shared'])
return body
class DeleteExternalPolicy(neutronV20.DeleteCommand):
"""Delete a given External Policy."""
resource = 'external_policy'
log = logging.getLogger(__name__ + '.DeleteExternalPolicy')
class UpdateExternalPolicy(neutronV20.UpdateCommand):
"""Update External Policy's information."""
resource = 'external_policy'
log = logging.getLogger(__name__ + '.UpdateExternalPolicy')
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('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'))
parser.add_argument(
'--provided-policy-rule-sets', type=utils.str2dict,
help=_('Dictionary of provided policy rule set uuids'))
parser.add_argument(
'--consumed-policy-rule-sets', type=utils.str2dict,
help=_('Dictionary of consumed policy rule set uuids'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
if parsed_args.provided_policy_rule_sets:
for key in parsed_args.provided_policy_rule_sets.keys():
id_key = neutronV20.find_resourceid_by_name_or_id(
self.get_client(), 'policy_rule_set',
key)
parsed_args.provided_policy_rule_sets[id_key] = (
parsed_args.provided_policy_rule_sets.pop(key))
if parsed_args.consumed_policy_rule_sets:
for key in parsed_args.consumed_policy_rule_sets.keys():
id_key = neutronV20.find_resourceid_by_name_or_id(
self.get_client(), 'policy_rule_set',
key)
parsed_args.consumed_policy_rule_sets[id_key] = (
parsed_args.consumed_policy_rule_sets.pop(key))
if parsed_args.external_segments:
for external_segment in parsed_args.external_segments:
external_segment_id = neutronV20.find_resourceid_by_name_or_id(
self.get_client(), 'external_segment', external_segment)
parsed_args.external_segments.remove(external_segment)
parsed_args.external_segments.append(external_segment_id)
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'tenant_id', 'description',
'provided_policy_rule_sets',
'external_segments',
'consumed_policy_rule_sets', 'shared'])
return body
class ListExternalSegment(neutronV20.ListCommand):
"""List External Segments that belong to a given tenant."""
resource = 'external_segment'
log = logging.getLogger(__name__ + '.ListExternalSegment')
_formatters = {'external_routes': _format_host_routes, }
list_columns = ['id', 'name', 'description', 'cidr',
'external_routes', 'port_address_translation', 'shared']
pagination_support = True
sorting_support = True
class ShowExternalSegment(neutronV20.ShowCommand):
"""Show information of a given External Segment."""
resource = 'external_segment'
log = logging.getLogger(__name__ + '.ShowExternalSegment')
class CreateExternalSegment(neutronV20.CreateCommand):
"""Create a External Segment for a given tenant."""
resource = 'external_segment'
log = logging.getLogger(__name__ + '.CreateExternalSegment')
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('Description of the External Segment'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of External Segment to create'))
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).'))
parser.add_argument(
'--port-address-translation', type=bool,
help=_('Perform port-based address translation, default is False'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
if parsed_args.external_routes:
body['external_segment']['external_routes'] = (
parsed_args.external_routes)
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'tenant_id', 'description',
'ip_version', 'cidr',
'external_routes', 'port_address_translation',
'shared'])
return body
class DeleteExternalSegment(neutronV20.DeleteCommand):
"""Delete a given External Segment."""
resource = 'external_segment'
log = logging.getLogger(__name__ + '.DeleteExternalSegment')
class UpdateExternalSegment(neutronV20.UpdateCommand):
"""Update External Segment's information."""
resource = 'external_segment'
log = logging.getLogger(__name__ + '.UpdateExternalSegment')
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('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).'))
parser.add_argument(
'--port-address-translation', type=bool,
help=_('Perform port-based address translation, default is False'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
if parsed_args.external_routes:
body['external_segment']['external_routes'] = (
parsed_args.external_routes)
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'tenant_id', 'description',
'ip_version', 'cidr',
'external_routes', 'port_address_translation',
'shared'])
return body
class ListNatPool(neutronV20.ListCommand):
"""List NAT Pools that belong to a given tenant."""
resource = 'nat_pool'
log = logging.getLogger(__name__ + '.ListNatPool')
list_columns = ['id', 'name', 'description', 'ip_pool',
'external_segment_id', 'shared']
pagination_support = True
sorting_support = True
class ShowNatPool(neutronV20.ShowCommand):
"""Show information of a given NAT Pool."""
resource = 'nat_pool'
log = logging.getLogger(__name__ + '.ShowNatPool')
class CreateNatPool(neutronV20.CreateCommand):
"""Create a NAT Pool for a given tenant."""
resource = 'nat_pool'
log = logging.getLogger(__name__ + '.CreateNatPool')
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('Description of the NAT Pool'))
parser.add_argument(
'name', metavar='NAME',
help=_('Name of NAT Pool to create'))
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'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'tenant_id', 'description',
'ip_version', 'ip_pool', 'shared'])
if parsed_args.external_segment:
body[self.resource]['external_segment_id'] = (
neutronV20.find_resourceid_by_name_or_id(
self.get_client(), 'external_segment',
parsed_args.external_segment))
return body
class DeleteNatPool(neutronV20.DeleteCommand):
"""Delete a given NAT Pool."""
resource = 'nat_pool'
log = logging.getLogger(__name__ + '.DeleteNatPool')
class UpdateNatPool(neutronV20.UpdateCommand):
"""Update NAT Pool's information."""
resource = 'nat_pool'
log = logging.getLogger(__name__ + '.UpdateNatPool')
def add_known_arguments(self, parser):
parser.add_argument(
'--description',
help=_('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'))
parser.add_argument(
'--shared', type=bool,
help=_('Shared flag'))
def args2body(self, parsed_args):
body = {self.resource: {}, }
neutronV20.update_dict(parsed_args, body[self.resource],
['name', 'tenant_id', 'description',
'ip_version', 'ip_pool', 'shared'])
if parsed_args.external_segment:
body[self.resource]['external_segment_id'] = (
neutronV20.find_resourceid_by_name_or_id(
self.get_client(), 'external_segment',
parsed_args.external_segment))
return body

15
gbpclient/gbpshell.py
View File

@ -117,6 +117,21 @@ COMMAND_V2 = {
'network-service-policy-update': gbp.UpdateNetworkServicePolicy,
'network-service-policy-list': gbp.ListNetworkServicePolicy,
'network-service-policy-show': gbp.ShowNetworkServicePolicy,
'external-policy-create': gbp.CreateExternalPolicy,
'external-policy-delete': gbp.DeleteExternalPolicy,
'external-policy-update': gbp.UpdateExternalPolicy,
'external-policy-list': gbp.ListExternalPolicy,
'external-policy-show': gbp.ShowExternalPolicy,
'external-segment-create': gbp.CreateExternalSegment,
'external-segment-delete': gbp.DeleteExternalSegment,
'external-segment-update': gbp.UpdateExternalSegment,
'external-segment-list': gbp.ListExternalSegment,
'external-segment-show': gbp.ShowExternalSegment,
'nat-pool-create': gbp.CreateNatPool,
'nat-pool-delete': gbp.DeleteNatPool,
'nat-pool-update': gbp.UpdateNatPool,
'nat-pool-list': gbp.ListNatPool,
'nat-pool-show': gbp.ShowNatPool,
'policy-classifier-create': gbp.CreatePolicyClassifier,
'policy-classifier-delete': gbp.DeletePolicyClassifier,
'policy-classifier-update': gbp.UpdatePolicyClassifier,

124
gbpclient/tests/unit/test_cli20_externalpolicy.py Normal file
View File

@ -0,0 +1,124 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import logging
import sys
from gbpclient.gbp.v2_0 import groupbasedpolicy as gbp
from gbpclient.tests.unit import test_cli20
class CLITestV20ExternalPolicyJSON(test_cli20.CLITestV20Base):
LOG = logging.getLogger(__name__)
def setUp(self):
super(CLITestV20ExternalPolicyJSON, self).setUp()
def test_create_external_policy_with_mandatory_params(self):
"""external-policy-create with all mandatory params."""
resource = 'external_policy'
cmd = gbp.CreateExternalPolicy(test_cli20.MyApp(sys.stdout), None)
name = 'my-name'
tenant_id = 'my-tenant'
my_id = 'my-id'
args = ['--tenant-id', tenant_id,
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)
def test_create_external_policy_with_all_params(self):
"""external-policy-create with all params."""
resource = 'external_policy'
cmd = gbp.CreateExternalPolicy(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
tenant_id = 'mytenant'
description = 'My External Policy'
my_id = 'someid'
provided_policy_rule_sets = "prs1=true,prs2=true"
consumed_policy_rule_sets = "prs3=true,prs4=true"
shared = 'True'
args = ['--tenant-id', tenant_id,
'--description', description,
'--provided-policy-rule-sets', provided_policy_rule_sets,
'--consumed-policy-rule-sets', consumed_policy_rule_sets,
'--shared', shared,
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,
description=description,
provided_policy_rule_sets=
{'prs1': 'true', 'prs2': 'true'},
consumed_policy_rule_sets=
{'prs3': 'true', 'prs4': 'true'},
shared=True)
def test_list_external_policies(self):
"""external-policy-list."""
resource = 'external_policies'
cmd = gbp.ListExternalPolicy(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resource, cmd, True)
def test_show_external_policy_name(self):
"""external-policy-show."""
resource = 'external_policy'
cmd = gbp.ShowExternalPolicy(test_cli20.MyApp(sys.stdout), None)
args = ['--fields', 'id', self.test_id]
self._test_show_resource(resource, cmd, self.test_id, args, ['id'])
def test_update_external_policy(self):
"external-policy-update myid --name myname --tags a b."
resource = 'external_policy'
cmd = gbp.UpdateExternalPolicy(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(resource, cmd, 'myid',
['myid', '--name', 'myname',
'--tags', 'a', 'b'],
{'name': 'myname', 'tags': ['a', 'b'], })
def test_update_external_policy_with_all_params(self):
resource = 'external_policy'
cmd = gbp.UpdateExternalPolicy(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
description = 'My External Policy'
my_id = 'someid'
provided_policy_rule_sets = "prs1=true,prs2=true"
consumed_policy_rule_sets = "prs3=true,prs4=true"
shared = 'True'
args = ['--name', name,
'--description', description,
'--provided-policy-rule-sets', provided_policy_rule_sets,
'--consumed-policy-rule-sets', consumed_policy_rule_sets,
'--shared', shared,
my_id]
params = {
'name': name,
'description': description,
'provided_policy_rule_sets': {'prs1': 'true', 'prs2': 'true'},
'consumed_policy_rule_sets': {'prs3': 'true', 'prs4': 'true'},
'shared': True
}
self._test_update_resource(resource, cmd, my_id, args, params)
def test_delete_external_policy_name(self):
"""external-policy-delete."""
resource = 'external_policy'
cmd = gbp.DeleteExternalPolicy(test_cli20.MyApp(sys.stdout), None)
my_id = 'my-id'
args = [my_id]
self._test_delete_resource(resource, cmd, my_id, args)

138
gbpclient/tests/unit/test_cli20_externalsegment.py Normal file
View File

@ -0,0 +1,138 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import logging
import sys
from gbpclient.gbp.v2_0 import groupbasedpolicy as gbp
from gbpclient.tests.unit import test_cli20
class CLITestV20ExternalSegmentJSON(test_cli20.CLITestV20Base):
LOG = logging.getLogger(__name__)
def setUp(self):
super(CLITestV20ExternalSegmentJSON, self).setUp()
def test_create_external_segment_with_mandatory_params(self):
"""external-segment-create with all mandatory params."""
resource = 'external_segment'
cmd = gbp.CreateExternalSegment(test_cli20.MyApp(sys.stdout), None)
name = 'my-name'
tenant_id = 'my-tenant'
my_id = 'my-id'
args = ['--tenant-id', tenant_id,
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)
def test_create_external_segment_with_all_params(self):
"""external-segment-create with all params."""
resource = 'external_segment'
cmd = gbp.CreateExternalSegment(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
tenant_id = 'mytenant'
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'}]
port_address_translation = 'True'
shared = 'True'
args = ['--tenant-id', tenant_id,
'--description', description,
'--ip-version', ip_version,
'--cidr', cidr,
'--external-route', external_route,
'--port-address-translation', port_address_translation,
'--shared', shared,
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,
description=description,
ip_version=4,
cidr=cidr,
external_routes=expected_external_routes,
port_address_translation=True,
shared=True)
def test_list_external_segments(self):
"""external-segment-list."""
resource = 'external_segments'
cmd = gbp.ListExternalSegment(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resource, cmd, True)
def test_show_external_segment_name(self):
"""external-segment-show."""
resource = 'external_segment'
cmd = gbp.ShowExternalSegment(test_cli20.MyApp(sys.stdout), None)
args = ['--fields', 'id', self.test_id]
self._test_show_resource(resource, cmd, self.test_id, args, ['id'])
def test_update_external_segment(self):
"external-segment-update myid --name myname --tags a b."
resource = 'external_segment'
cmd = gbp.UpdateExternalSegment(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(resource, cmd, 'myid',
['myid', '--name', 'myname',
'--tags', 'a', 'b'],
{'name': 'myname', 'tags': ['a', 'b'], })
def test_update_external_segment_with_all_params(self):
resource = 'external_segment'
cmd = gbp.UpdateExternalSegment(test_cli20.MyApp(sys.stdout), None)
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'}]
port_address_translation = 'True'
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,
my_id]
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_delete_external_segment_name(self):
"""external-segment-delete."""
resource = 'external_segment'
cmd = gbp.DeleteExternalSegment(test_cli20.MyApp(sys.stdout), None)
my_id = 'my-id'
args = [my_id]
self._test_delete_resource(resource, cmd, my_id, args)

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

@ -39,6 +39,38 @@ class CLITestV20L3PolicyJSON(test_cli20.CLITestV20Base):
position_names, position_values,
tenant_id=tenant_id)
def test_create_l3_policy_with_all_params(self):
"""l3-policy-create with all params."""
resource = 'l3_policy'
cmd = gbp.CreateL3Policy(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
tenant_id = 'mytenant'
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']}
args = ['--tenant-id', tenant_id,
'--description', description,
'--ip-version', ip_version,
'--ip-pool', ip_pool,
'--subnet-prefix-length', subnet_prefix_length,
'--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,
description=description,
ip_version=4,
ip_pool=ip_pool,
subnet_prefix_length=24,
external_segments=
expected_external_segments)
def test_list_l3_policies(self):
resource = 'l3_policies'
cmd = gbp.ListL3Policy(test_cli20.MyApp(sys.stdout), None)
@ -58,6 +90,34 @@ class CLITestV20L3PolicyJSON(test_cli20.CLITestV20Base):
'--tags', 'a', 'b'],
{'name': 'myname', 'tags': ['a', 'b'], })
def test_update_l3_policy_with_all_params(self):
resource = 'l3_policy'
cmd = gbp.UpdateL3Policy(test_cli20.MyApp(sys.stdout), None)
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']}
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_delete_l3_policy_name(self):
resource = 'l3_policy'
cmd = gbp.DeleteL3Policy(test_cli20.MyApp(sys.stdout), None)

128
gbpclient/tests/unit/test_cli20_natpool.py Normal file
View File

@ -0,0 +1,128 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import logging
import sys
from gbpclient.gbp.v2_0 import groupbasedpolicy as gbp
from gbpclient.tests.unit import test_cli20
class CLITestV20NatPoolJSON(test_cli20.CLITestV20Base):
LOG = logging.getLogger(__name__)
def setUp(self):
super(CLITestV20NatPoolJSON, self).setUp()
def test_create_nat_pool_with_mandatory_params(self):
"""nat-pool-create with all mandatory params."""
resource = 'nat_pool'
cmd = gbp.CreateNatPool(test_cli20.MyApp(sys.stdout), None)
name = 'my-name'
tenant_id = 'my-tenant'
my_id = 'my-id'
args = ['--tenant-id', tenant_id,
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)
def test_create_nat_pool_with_all_params(self):
"""nat-pool-create with all params."""
resource = 'nat_pool'
cmd = gbp.CreateNatPool(test_cli20.MyApp(sys.stdout), None)
name = 'myname'
tenant_id = 'mytenant'
description = 'My Nat Pool'
my_id = 'someid'
ip_version = '4'
ip_pool = '192.168.0.0/24'
external_segment_id = "segmentid"
shared = 'True'
args = ['--tenant-id', tenant_id,
'--description', description,
'--ip-version', ip_version,
'--ip-pool', ip_pool,
'--external-segment', external_segment_id,
'--shared', shared,
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,
description=description,
ip_version=4,
ip_pool=ip_pool,
external_segment_id=external_segment_id,
shared=True)
def test_list_nat_pools(self):
"""nat-pool-list."""
resource = 'nat_pools'
cmd = gbp.ListNatPool(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resource, cmd, True)
def test_show_nat_pool_name(self):
"""nat-pool-show."""
resource = 'nat_pool'
cmd = gbp.ShowNatPool(test_cli20.MyApp(sys.stdout), None)
args = ['--fields', 'id', self.test_id]
self._test_show_resource(resource, cmd, self.test_id, args, ['id'])
def test_update_nat_pool(self):
"nat-pool-update myid --name myname --tags a b."
resource = 'nat_pool'
cmd = gbp.UpdateNatPool(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(resource, cmd, 'myid',
['myid', '--name', 'myname',
'--tags', 'a', 'b'],
{'name': 'myname', 'tags': ['a', 'b'], })
def test_update_nat_pool_with_all_params(self):
resource = 'nat_pool'
cmd = gbp.UpdateNatPool(test_cli20.MyApp(sys.stdout), None)
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
}
self._test_update_resource(resource, cmd, my_id, args, params)
def test_delete_nat_pool_name(self):
"""nat-pool-delete."""
resource = 'nat_pool'
cmd = gbp.DeleteNatPool(test_cli20.MyApp(sys.stdout), None)
my_id = 'my-id'
args = [my_id]
self._test_delete_resource(resource, cmd, my_id, args)

103
gbpclient/v2_0/client.py
View File

@ -157,6 +157,12 @@ class Client(object):
l3_policy_path = "/grouppolicy/l3_policies/%s"
network_service_policies_path = "/grouppolicy/network_service_policies"
network_service_policy_path = "/grouppolicy/network_service_policies/%s"
external_policies_path = "/grouppolicy/external_policies"
external_policy_path = "/grouppolicy/external_policies/%s"
external_segments_path = "/grouppolicy/external_segments"
external_segment_path = "/grouppolicy/external_segments/%s"
nat_pools_path = "/grouppolicy/nat_pools"
nat_pool_path = "/grouppolicy/nat_pools/%s"
policy_classifiers_path = "/grouppolicy/policy_classifiers"
policy_classifier_path = "/grouppolicy/policy_classifiers/%s"
policy_actions_path = "/grouppolicy/policy_actions"
@ -178,6 +184,9 @@ class Client(object):
'l2_policies': 'l2_policy',
'l3_policies': 'l3_policy',
'network_service_policies': 'network_service_policy',
'external_policies': 'external_policy',
'external_segments': 'external_segment',
'nat_pools': 'nat_pool',
'policy_classifiers': 'policy_classifier',
'policy_actions': 'policy_action',
'policy_rules': 'policy_rule',
@ -329,6 +338,100 @@ class Client(object):
return self.delete(
self.network_service_policy_path % (network_service_policy))
@APIParamsCall
def list_external_policies(self, retrieve_all=True, **_params):
"""Fetches a list of all external_policies for a tenant."""
# Pass filters in "params" argument to do_request
return self.list('external_policies',
self.external_policies_path,
retrieve_all, **_params)
@APIParamsCall
def show_external_policy(self, external_policy, **_params):
"""Fetches information of a certain external_policy."""
return self.get(
self.external_policy_path % (external_policy),
params=_params)
@APIParamsCall
def create_external_policy(self, body=None):
"""Creates a new external_policy."""
return self.post(self.external_policies_path, body=body)
@APIParamsCall
def update_external_policy(self, external_policy, body=None):
"""Updates a external_policy."""
return self.put(
self.external_policy_path % (external_policy),
body=body)
@APIParamsCall
def delete_external_policy(self, external_policy):
"""Deletes the specified external_policy."""
return self.delete(
self.external_policy_path % (external_policy))
@APIParamsCall
def list_external_segments(self, retrieve_all=True, **_params):
"""Fetches a list of all external_segments for a tenant."""
# Pass filters in "params" argument to do_request
return self.list('external_segments',
self.external_segments_path,
retrieve_all, **_params)
@APIParamsCall
def show_external_segment(self, external_segment, **_params):
"""Fetches information of a certain external_segment."""
return self.get(
self.external_segment_path % (external_segment),
params=_params)
@APIParamsCall
def create_external_segment(self, body=None):
"""Creates a new external_segment."""
return self.post(self.external_segments_path, body=body)
@APIParamsCall
def update_external_segment(self, external_segment, body=None):
"""Updates a external_segment."""
return self.put(
self.external_segment_path % (external_segment),
body=body)
@APIParamsCall
def delete_external_segment(self, external_segment):
"""Deletes the specified external_segment."""
return self.delete(
self.external_segment_path % (external_segment))
@APIParamsCall
def list_nat_pools(self, retrieve_all=True, **_params):
"""Fetches a list of all nat_pools for a tenant."""
# Pass filters in "params" argument to do_request
return self.list('nat_pools',
self.nat_pools_path,
retrieve_all, **_params)
@APIParamsCall
def show_nat_pool(self, nat_pool, **_params):
"""Fetches information of a certain nat_pool."""
return self.get(self.nat_pool_path % (nat_pool), params=_params)
@APIParamsCall
def create_nat_pool(self, body=None):
"""Creates a new nat_pool."""
return self.post(self.nat_pools_path, body=body)
@APIParamsCall
def update_nat_pool(self, nat_pool, body=None):
"""Updates a nat_pool."""
return self.put(self.nat_pool_path % (nat_pool), body=body)
@APIParamsCall
def delete_nat_pool(self, nat_pool):
"""Deletes the specified nat_pool."""
return self.delete(self.nat_pool_path % (nat_pool))
@APIParamsCall
def list_l3_policies(self, retrieve_all=True, **_params):
"""Fetches a list of all l3_policies for a tenant."""