L2 gateway client implementation

This is a code for l2-gateway-client. Create/Update/Delete/List/Show CLIs for
l2gateway and l2-gateway-connection are implemented here.

Minor tweak to the requirements was needed to get the package to install
correctly and versioned to the tag we expect.

Change-Id: I1c722643b63a523670862a6c804b19b0b6bac585
stable/ocata
preeti-mirji 8 years ago committed by armando-migliaccio
parent fce1ea47d6
commit 3171db8634

@ -0,0 +1,3 @@
from neutronclient.neutron import v2_0 as neutronV2_0
_get_resource_plural = neutronV2_0._get_resource_plural

@ -0,0 +1,81 @@
# Copyright 2015 OpenStack Foundation.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
from neutronclient.common import utils
from neutronclient.i18n import _
from neutronclient.neutron import v2_0 as l2gatewayV20
from oslo.serialization import jsonutils
L2_GW = 'l2_gateway'
def _format_devices(l2_gateway):
try:
return '\n'.join([jsonutils.dumps(gateway) for gateway in
l2_gateway['devices']])
except (TypeError, KeyError):
return ''
class Listl2gateway(l2gatewayV20.ListCommand):
"""List l2gateways that belongs to a given tenant."""
resource = L2_GW
_formatters = {'devices': _format_devices, }
list_columns = ['id', 'name', 'devices']
pagination_support = True
sorting_support = True
class Showl2gateway(l2gatewayV20.ShowCommand):
"""Show information of a given l2gateway."""
resource = L2_GW
class Deletel2gateway(l2gatewayV20.DeleteCommand):
"""Delete a given l2gateway."""
resource = L2_GW
class Createl2gateway(l2gatewayV20.CreateCommand):
"""Create l2gateway information."""
resource = L2_GW
def add_known_arguments(self, parser):
parser.add_argument(
'name', metavar='<GATEWAY-NAME>',
help=_('Descriptive name for logical gateway.'))
parser.add_argument(
'--device', metavar='name=name,interface_name=interface_name',
action='append', dest='devices', type=utils.str2dict,
help=_('names or identifiers of the L2 gateways.'
'(This option can be repeated)'))
def args2body(self, parsed_args):
body = {'l2_gateway': {'name': parsed_args.name,
'devices': parsed_args.devices}, }
return body
class Updatel2gateway(l2gatewayV20.UpdateCommand):
"""Update a given l2gateway."""
resource = L2_GW

@ -0,0 +1,76 @@
# Copyright 2015 OpenStack Foundation.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
from neutronclient.i18n import _
from neutronclient.neutron import v2_0 as l2gatewayV20
L2_GW_CONNECTION = 'l2_gateway_connection'
class Listl2gatewayConnection(l2gatewayV20.ListCommand):
"""List l2gateway-connections."""
resource = L2_GW_CONNECTION
list_columns = ['id', 'l2_gateway_id', 'network_id', 'segmentation_id']
pagination_support = True
sorting_support = True
class Showl2gatewayConnection(l2gatewayV20.ShowCommand):
"""Show information of a given l2gateway-connection."""
resource = L2_GW_CONNECTION
class Deletel2gatewayConnection(l2gatewayV20.DeleteCommand):
"""Delete a given l2gateway-connection."""
resource = L2_GW_CONNECTION
class Createl2gatewayConnection(l2gatewayV20.CreateCommand):
"""Create l2gateway-connection information."""
resource = L2_GW_CONNECTION
def add_known_arguments(self, parser):
parser.add_argument(
'gateway_name', metavar='<GATEWAY-NAME/UUID>',
help=_('Descriptive name for logical gateway.'))
parser.add_argument(
'network', metavar='<NETWORK-NAME/UUID>',
help=_('Network name or uuid.'))
parser.add_argument(
'--default-segmentation-id',
dest='seg_id',
help=_('default segmentation-id that will '
'be applied to the interfaces for which '
'segmentation id was not specified '
'in l2-gateway-create command.'))
def args2body(self, parsed_args):
body = {'l2_gateway_connection':
{'l2_gateway_id': parsed_args.gateway_name,
'network_id': parsed_args.network,
'segmentation_id': parsed_args.seg_id}}
return body
class Updatel2gatewayConnection(l2gatewayV20.UpdateCommand):
"""Update a given l2gateway-connection."""
resource = L2_GW_CONNECTION

@ -0,0 +1,89 @@
# Copyright 2015 OpenStack Foundation.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
"""
Command-line interface to the L2gateway APIs
"""
from __future__ import print_function
import sys
from cliff import commandmanager
from neutronclient.common import clientmanager
from neutronclient.common import exceptions as exc
from neutronclient import shell as neutronshell
from oslo.utils import encodeutils
from networking_l2gw.l2gatewayclient.l2gateway.v2_0 import l2gateway
from networking_l2gw.l2gatewayclient.l2gateway.v2_0 import l2gateway_connection
VERSION = '2.0'
NEUTRON_API_VERSION = '2.0'
clientmanager.neutron_client.API_VERSIONS = {
'2.0': 'l2gatewayclient.v2_0.client.Client',
}
COMMAND_V2 = {
'l2-gateway-create': l2gateway.Createl2gateway,
'l2-gateway-list': l2gateway.Listl2gateway,
'l2-gateway-show': l2gateway.Showl2gateway,
'l2-gateway-delete': l2gateway.Deletel2gateway,
'l2-gateway-update': l2gateway.Updatel2gateway,
'l2-gateway-connection-create': (l2gateway_connection.
Createl2gatewayConnection),
'l2-gateway-connection-list': (l2gateway_connection.
Listl2gatewayConnection),
'l2-gateway-connection-show': (l2gateway_connection.
Showl2gatewayConnection),
'l2-gateway-connection-delete': (l2gateway_connection.
Deletel2gatewayConnection),
'l2-gateway-connection-update': (l2gateway_connection.
Updatel2gatewayConnection),
}
COMMANDS = {'2.0': COMMAND_V2}
class L2gatewayShell(neutronshell.NeutronShell):
def __init__(self, apiversion):
super(neutronshell.NeutronShell, self).__init__(
description=__doc__.strip(),
version=VERSION,
command_manager=commandmanager.CommandManager('l2gateway.cli'), )
self.commands = COMMANDS
for k, v in self.commands[apiversion].items():
self.command_manager.add_command(k, v)
# This is instantiated in initialize_app() only when using
# password flow auth
self.auth_client = None
self.api_version = apiversion
def main(argv=sys.argv[1:]):
try:
return L2gatewayShell(NEUTRON_API_VERSION).run(list(map(
encodeutils.safe_decode, argv)))
except exc.NeutronClientException:
return 1
except Exception as e:
print(unicode(e))
return 1
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))

@ -0,0 +1,148 @@
# Copyright 2015 OpenStack Foundation.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import logging
from neutronclient import client
from neutronclient.v2_0 import client as V2_Client
_logger = logging.getLogger(__name__)
class APIParamsCall(V2_Client.APIParamsCall):
pass
class Client(V2_Client.Client):
"""Client for the L2gateway v2.0 API.
:param string username: Username for authentication. (optional)
:param string user_id: User ID for authentication. (optional)
:param string password: Password for authentication. (optional)
:param string token: Token for authentication. (optional)
:param string tenant_name: Tenant name. (optional)
:param string tenant_id: Tenant id. (optional)
:param string auth_url: Keystone service endpoint for authorization.
:param string service_type: Network service type to pull from the
keystone catalog (e.g. 'network') (optional)
:param string endpoint_type: Network service endpoint type to pull from the
keystone catalog (e.g. 'publicURL',
'internalURL', or 'adminURL') (optional)
:param string region_name: Name of a region to select when choosing an
endpoint from the service catalog.
:param string endpoint_url: A user-supplied endpoint URL for the neutron
service. Lazy-authentication is possible for API
service calls if endpoint is set at
instantiation.(optional)
:param integer timeout: Allows customization of the timeout for client
http requests. (optional)
:param bool insecure: SSL certificate validation. (optional)
:param string ca_cert: SSL CA bundle file to use. (optional)
:param integer retries: How many times idempotent (GET, PUT, DELETE)
requests to Neutron server should be retried if
they fail (default: 0).
:param bool raise_errors: If True then exceptions caused by connection
failure are propagated to the caller.
(default: True)
:param session: Keystone client auth session to use. (optional)
:param auth: Keystone auth plugin to use. (optional)
"""
l2_gateways_path = "/l2-gateways"
l2_gateway_path = "/l2-gateways/%s"
l2_gateway_connections_path = "/l2-gateway-connections"
l2_gateway_connection_path = "/l2-gateway-connections/%s"
# API has no way to report plurals, so we have to hard code them
EXTED_PLURALS = {'l2gateways': 'l2-gateways',
'l2gateway-connections': 'l2-gateway-connections',
}
@APIParamsCall
def create_l2_gateway(self, body=None):
"""Creates a new l2gateway."""
return self.post(self.l2_gateways_path, body=body)
@APIParamsCall
def list_l2_gateways(self, retrieve_all=True, **_params):
"""Fetches a list of l2gateways."""
return self.list('l2_gateways', self.l2_gateways_path, retrieve_all,
**_params)
@APIParamsCall
def show_l2_gateway(self, l2gateway, **_params):
"""Fetches information of a certain l2gateway."""
return self.get(self.l2_gateway_path % (l2gateway), params=_params)
@APIParamsCall
def delete_l2_gateway(self, l2gateway):
"""Deletes the specified l2gateway."""
return self.delete(self.l2_gateway_path % (l2gateway))
@APIParamsCall
def update_l2_gateway(self, l2gateway, body):
"""Updates the specified l2gateway."""
return self.put(self.l2_gateway_path % (l2gateway), body=body)
@APIParamsCall
def create_l2_gateway_connection(self, body=None):
"""Creates a new l2gateway-connection."""
return self.post(self.l2_gateway_connections_path, body=body)
@APIParamsCall
def list_l2_gateway_connections(self, retrieve_all=True, **_params):
"""Fetches a list of l2gateway-connections."""
return self.list('l2_gateway_connections',
self.l2_gateway_connections_path, retrieve_all,
**_params)
@APIParamsCall
def show_l2_gateway_connection(self, con_id, **_params):
"""Fetches information of a certain l2gateway-connection."""
return self.get(self.l2_gateway_connection_path % (con_id),
params=_params)
@APIParamsCall
def delete_l2_gateway_connection(self, con_id):
"""Deletes the specified l2gateway-connection."""
return self.delete(self.l2_gateway_connection_path % (con_id))
@APIParamsCall
def update_l2_gateway_connection(self, l2gateway_conn, body):
"""Updates the specified l2gateway-connection."""
return self.put(self.l2_gateway_connection_path % (l2gateway_conn),
body=body)
def __init__(self, **kwargs):
"""Initialize a new client for the L2gateway v2.0 API."""
super(Client, self).__init__()
self.retries = kwargs.pop('retries', 0)
self.raise_errors = kwargs.pop('raise_errors', True)
self.httpclient = client.construct_http_client(**kwargs)
self.version = '2.0'
self.format = 'json'
self.action_prefix = "/v%s" % (self.version)
self.retry_interval = 1

@ -0,0 +1,21 @@
# Copyright (c) 2015 Hewlett-Packard Development Company, L.P.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import pbr.version
__version__ = (pbr.version.VersionInfo('python-l2gatewayclient').
version_string())

@ -0,0 +1,91 @@
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
from mox3 import mox
from networking_l2gw.l2gatewayclient.l2gateway import v2_0 as l2gatewayV2_0
from networking_l2gw.l2gatewayclient.v2_0 import client as l2gatewayclient
from neutronclient import shell as neutronshell
from neutronclient.tests.unit import test_cli20 as neutron_test_cli20
TOKEN = neutron_test_cli20.TOKEN
end_url = neutron_test_cli20.end_url
class MyResp(neutron_test_cli20.MyResp):
pass
class MyApp(neutron_test_cli20.MyApp):
pass
class MyComparator(neutron_test_cli20.MyComparator):
pass
class CLITestV20Base(neutron_test_cli20.CLITestV20Base):
def setUp(self, plurals=None):
super(CLITestV20Base, self).setUp()
self.client = l2gatewayclient.Client(token=TOKEN,
endpoint_url=self.endurl)
def _test_create_resource(self, resource, cmd, name, myid, args,
position_names, position_values,
tenant_id=None, tags=None, admin_state_up=True,
extra_body=None, cmd_resource=None,
parent_id=None, **kwargs):
self.mox.StubOutWithMock(cmd, "get_client")
self.mox.StubOutWithMock(self.client.httpclient, "request")
cmd.get_client().MultipleTimes().AndReturn(self.client)
if not cmd_resource:
cmd_resource = resource
body = {resource: {}, }
body[resource].update(kwargs)
for i in range(len(position_names)):
body[resource].update({position_names[i]: position_values[i]})
ress = {resource:
{self.id_field: myid}, }
if name:
ress[resource].update({'name': name})
self.client.format = self.format
resstr = self.client.serialize(ress)
# url method body
resource_plural = l2gatewayV2_0._get_resource_plural(cmd_resource,
self.client)
path = getattr(self.client, resource_plural + "_path")
if self.format == 'json':
mox_body = MyComparator(body, self.client)
else:
mox_body = self.client.serialize(body)
self.client.httpclient.request(
end_url(path, format=self.format), 'POST',
body=mox_body,
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((MyResp(200), resstr))
args.extend(['--request-format', self.format])
self.mox.ReplayAll()
cmd_parser = cmd.get_parser('create_' + resource)
neutronshell.run_command(cmd, cmd_parser, args)
self.mox.VerifyAll()
self.mox.UnsetStubs()
_str = self.fake_stdout.make_string()
self.assertIn(myid, _str)
if name:
self.assertIn(name, _str)

@ -0,0 +1,73 @@
# Copyright 2015 OpenStack Foundation.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import sys
from networking_l2gw.l2gatewayclient.l2gateway.v2_0 import l2gateway
from networking_l2gw.tests.unit.l2gatewayclient import test_cli20
class CLITestV20L2gatewayJSON(test_cli20.CLITestV20Base):
def setUp(self):
super(CLITestV20L2gatewayJSON, self).setUp(plurals={'tags': 'tag'})
def test_create_l2gateway(self):
"""Test Create l2gateway."""
resource = 'l2_gateway'
cmd = l2gateway.Createl2gateway(test_cli20.MyApp(sys.stdout), None)
name = 'l2gw'
args = [name, '--device', 'name=d1,interface_name=i1']
device = [{'name': 'd1', 'interface_name': 'i1'}]
position_names = ['name', 'devices']
position_values = [name, device]
self._test_create_resource(resource, cmd, name, 'myid', args,
position_names, position_values)
def test_list_l2gateway(self):
"""Test List l2gateway."""
resources = "l2_gateways"
cmd = l2gateway.Listl2gateway(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(resources, cmd, True)
def test_delete_l2gateway(self):
"""Test Delete l2gateway."""
resource = 'l2_gateway'
cmd = l2gateway.Deletel2gateway(test_cli20.MyApp(sys.stdout), None)
my_id = 'my-id'
args = [my_id]
self._test_delete_resource(resource, cmd, my_id, args)
def test_show_l2gateway(self):
"""Test Show l2gateway: --fields id --fields name myid."""
resource = 'l2_gateway'
cmd = l2gateway.Showl2gateway(test_cli20.MyApp(sys.stdout), None)
args = ['--fields', 'id', '--fields', 'name', self.test_id]
self._test_show_resource(resource, cmd, self.test_id, args,
['id', 'name'])
def test_update_l2gateway(self):
"""Test Update l2gateway."""
resource = 'l2_gateway'
cmd = l2gateway.Updatel2gateway(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(resource, cmd, 'myid',
['myid', '--name', 'myname', ],
{'name': 'myname', }
)

@ -0,0 +1,79 @@
# Copyright 2015 OpenStack Foundation.
# All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import sys
from networking_l2gw.l2gatewayclient.l2gateway.v2_0 import l2gateway_connection
from networking_l2gw.tests.unit.l2gatewayclient import test_cli20
class CLITestV20L2gatewayConnectionJSON(test_cli20.CLITestV20Base):
def setUp(self):
super(CLITestV20L2gatewayConnectionJSON, self).setUp(
plurals={'tags': 'tag'})
def test_create_l2gateway_connection(self):
"""Test Create l2gateway-connection."""
resource = 'l2_gateway_connection'
cmd = l2gateway_connection.Createl2gatewayConnection(test_cli20.MyApp(
sys.stdout), None)
l2gw_id = 'l2gw-id'
net_id = 'net-id'
args = [l2gw_id, net_id, '--default-segmentation-id', 'seg-id']
position_names = ['l2_gateway_id', 'network_id', 'segmentation_id']
position_values = [l2gw_id, net_id, 'seg-id']
self._test_create_resource(resource, cmd, l2gw_id, 'myid', args,
position_names, position_values)
def test_list_l2gateway_connection(self):
"""Test List l2gateway-connection."""
resources = "l2_gateway_connections"
cmd = l2gateway_connection.Listl2gatewayConnection(test_cli20.MyApp(
sys.stdout), None)
self._test_list_resources(resources, cmd, True)
def test_delete_l2gateway_connection(self):
"""Test Delete l2gateway-connection."""
resource = 'l2_gateway_connection'
cmd = l2gateway_connection.Deletel2gatewayConnection(test_cli20.MyApp(
sys.stdout), None)
my_id = 'my-id'
args = [my_id]
self._test_delete_resource(resource, cmd, my_id, args)
def test_show_l2gateway_connection(self):
"""Test Show l2gateway-connection: --fields id --fields name myid."""
resource = 'l2_gateway_connection'
cmd = l2gateway_connection.Showl2gatewayConnection(test_cli20.MyApp(
sys.stdout), None)
args = ['--fields', 'id', '--fields', 'name', self.test_id]
self._test_show_resource(resource, cmd, self.test_id, args,
['id', 'name'])
def test_update_l2gateway_connection(self):
"""Test Update l2gateway-connection."""
resource = 'l2_gateway_connection'
cmd = l2gateway_connection.Updatel2gatewayConnection(test_cli20.MyApp(
sys.stdout), None)
self._test_update_resource(resource, cmd, 'myid',
['myid', '--segmentation_id', 'seg-id', ],
{'segmentation_id': 'seg-id', }
)

@ -4,5 +4,5 @@
pbr>=0.6,!=0.7,<1.0
Babel>=1.3
python-neutronclient>=2.3.10
-e git://git.openstack.org/openstack/neutron.git#egg=neutron
-e git://git.openstack.org/openstack/python-neutronclient.git#egg=neutronclient

@ -24,6 +24,10 @@ classifier =
packages =
networking_l2gw
[entry_points]
console_scripts =
neutron-l2gw = networking_l2gw.l2gatewayclient.shell:main
[build_sphinx]
source-dir = doc/source
build-dir = doc/build

@ -25,9 +25,10 @@ commands = python setup.py build_sphinx
[flake8]
# H803 skipped on purpose per list discussion.
# H302 import only modules
# E123, E125 skipped as they are invalid PEP-8.
show-source = True
ignore = E123,E125,H803
ignore = E123,E125,H803,H302
builtins = _
exclude=.venv,.git,.tox,dist,doc,*openstack/common*,*lib/python*,*egg,build

Loading…
Cancel
Save