Re-implement TaaS CLI as a NeutronClient extension

The following patch introduces the NeutronClient extension
of tap-as-a-service, thus integrating TAAS with the python-neutronclient.

Change-Id: I8ce10e3001c46e24980dc701540cca8e837c3700
Closes-Bug: #1517357
Depends-On: I41fb7b538c81eba848b78c071edaf806663063a1
This commit is contained in:
Reedip Banerjee 2015-11-25 10:52:19 +05:30 committed by reedip
parent c6a42502bb
commit 22af25fa1f
7 changed files with 462 additions and 0 deletions

View File

View File

@ -0,0 +1,121 @@
# Copyright 2015 NEC Corporation
# 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 extension
from neutronclient.common import utils
from neutronclient.i18n import _
from neutronclient.neutron import v2_0 as neutronv20
def _add_updatable_args(parser):
parser.add_argument(
'--name',
help=_('Name of this Tap flow.'))
parser.add_argument(
'--description',
help=_('Description for this Tap flow.'))
def _updatable_args2body(parsed_args, body):
neutronv20.update_dict(parsed_args, body, ['name', 'description'])
class TapFlow(extension.NeutronClientExtension):
# Define required variables for resource operations.
resource = 'tap-flow'
resource_plural = '%ss' % resource
object_path = '/taas/%s' % resource_plural
resource_path = '/taas/%s/%%s' % resource_plural
versions = ['2.0']
class ListTapFlow(extension.ClientExtensionList, TapFlow):
# List tap flows.
shell_command = 'tap-flow-list'
list_columns = ['id', 'name', 'source_port', 'tap_service_id']
pagination_support = True
sorting_support = True
class CreateTapFlow(extension.ClientExtensionCreate, TapFlow):
# Create a tap flow.
shell_command = 'tap-flow-create'
list_columns = ['id', 'name', 'direction', 'source_port']
def add_known_arguments(self, parser):
_add_updatable_args(parser)
parser.add_argument(
'--port',
required=True,
metavar="SOURCE_PORT",
help=_('Source port to which the Tap Flow is connected.'))
parser.add_argument(
'--service',
required=True,
dest='service_id',
metavar="SERVICE",
help=_('Tap Service to which the flow belongs.'))
parser.add_argument(
'--direction',
required=True,
metavar="DIRECTION",
choices=['IN', 'OUT', 'BOTH'],
type=utils.convert_to_uppercase,
help=_('Direction of the Tap flow .'))
def args2body(self, parsed_args):
client = self.get_client()
source_port = neutronv20.find_resourceid_by_name_or_id(
client, 'port',
parsed_args.port)
service_id = neutronv20.find_resourceid_by_name_or_id(
client, 'tap-service',
parsed_args.service_id)
body = {'source_port': source_port,
'tap_service_id': service_id}
neutronv20.update_dict(parsed_args, body, ['tenant_id', 'direction'])
_updatable_args2body(parsed_args, body)
return {self.resource: body}
class DeleteTapFlow(extension.ClientExtensionDelete, TapFlow):
# Delete a tap flow.
shell_command = 'tap-flow-delete'
class ShowTapFlow(extension.ClientExtensionShow, TapFlow):
# Show a tap flow.
shell_command = 'tap-flow-show'
class UpdateTapFlow(extension.ClientExtensionUpdate, TapFlow):
# Update a tap flow.
shell_command = 'tap-flow-update'
list_columns = ['id', 'name']
def add_known_arguments(self, parser):
_add_updatable_args(parser)
def args2body(self, parsed_args):
body = {}
_updatable_args2body(parsed_args, body)
return {self.resource: body}

View File

@ -0,0 +1,114 @@
# Copyright 2015 NEC Corporation
# 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 extension
from neutronclient.i18n import _
from neutronclient.neutron import v2_0 as neutronv20
def _add_updatable_args(parser):
parser.add_argument(
'--name',
help=_('Name of this Tap service.'))
parser.add_argument(
'--description',
help=_('Description for this Tap service.'))
def _updatable_args2body(parsed_args, body):
neutronv20.update_dict(parsed_args, body, ['name', 'description'])
class TapService(extension.NeutronClientExtension):
# Define required variables for resource operations.
resource = 'tap-service'
resource_plural = '%ss' % resource
object_path = '/taas/%s' % resource_plural
resource_path = '/taas/%s/%%s' % resource_plural
versions = ['2.0']
class ListTapService(extension.ClientExtensionList, TapService):
# List tap services.
shell_command = 'tap-service-list'
list_columns = ['id', 'name', 'port']
pagination_support = True
sorting_support = True
class CreateTapService(extension.ClientExtensionCreate, TapService):
# Create a tap service.
shell_command = 'tap-service-create'
list_columns = ['id', 'name', 'port', 'network']
def add_known_arguments(self, parser):
_add_updatable_args(parser)
parser.add_argument(
'--port',
dest='port_id',
required=True,
metavar="PORT",
help=_('Port to which the Tap service is connected.'))
parser.add_argument(
'--network',
dest='network_id',
required=True,
metavar="NETWORK",
help=_('Network to which the Tap service is connected.'))
def args2body(self, parsed_args):
client = self.get_client()
port_id = neutronv20.find_resourceid_by_name_or_id(
client, 'port',
parsed_args.port_id)
network_id = neutronv20.find_resourceid_by_name_or_id(
client, 'network',
parsed_args.network_id)
body = {'port_id': port_id,
'network_id': network_id,
'tenant_id': parsed_args.tenant_id}
_updatable_args2body(parsed_args, body)
return {self.resource: body}
class DeleteTapService(extension.ClientExtensionDelete, TapService):
# Delete a tap service.
shell_command = 'tap-service-delete'
class ShowTapService(extension.ClientExtensionShow, TapService):
# Show a tap service.
shell_command = 'tap-service-show'
class UpdateTapService(extension.ClientExtensionUpdate, TapService):
# Update a tap service.
shell_command = 'tap-service-update'
list_columns = ['id', 'name']
def add_known_arguments(self, parser):
_add_updatable_args(parser)
def args2body(self, parsed_args):
body = {}
_updatable_args2body(parsed_args, body)
return {self.resource: body}

View File

@ -0,0 +1,112 @@
# Copyright 2015 NEC Corporation
# 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 mock
import sys
from neutron_taas.taas_client import tapflow
from neutronclient import shell
from neutronclient.tests.unit import test_cli20
class CLITestV20TapFlowJSON(test_cli20.CLITestV20Base,
tapflow.TapFlow):
def setUp(self):
self._mock_extension_loading()
super(CLITestV20TapFlowJSON, self).setUp()
self.resources = self.resource_plural
self.register_non_admin_status_resource(self.resource)
def _create_patch(self, name, func=None):
patcher = mock.patch(name)
thing = patcher.start()
self.addCleanup(patcher.stop)
return thing
def _mock_extension_loading(self):
ext_pkg = 'neutronclient.common.extension'
contrib = self._create_patch(ext_pkg + '._discover_via_entry_points')
contrib.return_value = [("_tap_flow", tapflow)]
return contrib
def test_ext_cmd_loaded(self):
shell.NeutronShell('2.0')
extension_cmd = {'tap-flow-create': tapflow.CreateTapFlow,
'tap-flow-delete': tapflow.DeleteTapFlow,
'tap-flow-show': tapflow.ShowTapFlow,
'tap-flow-list': tapflow.ListTapFlow}
self.assertDictContainsSubset(extension_cmd, shell.COMMANDS['2.0'])
def _test_create_tap_flow(self, port_id="random_port",
service_id="random_service",
direction="BOTH", arg_attr=None,
name_attr=None, val_attr=None,
name=''):
# Common definition for creating Tap flow
arg_attr = arg_attr or []
name_attr = name_attr or []
val_attr = val_attr or []
cmd = tapflow.CreateTapFlow(test_cli20.MyApp(sys.stdout), None)
tenant_id = 'my-tenant'
my_id = 'my-id'
args = ['--tenant-id', tenant_id,
'--port', port_id,
'--service', service_id,
'--direction', direction] + arg_attr
pos_names = ['source_port', 'tap_service_id', 'direction'] + name_attr
pos_values = [port_id, service_id, direction] + val_attr
self._test_create_resource(self.resource, cmd, name, my_id, args,
pos_names, pos_values,
tenant_id=tenant_id)
def test_create_tap_flow_mandatory_params(self):
self._test_create_tap_flow()
def test_create_tap_flow_all_params(self):
name = 'dummyTapFlow'
description = 'Create a dummy tap flow'
self._test_create_tap_flow(name=name,
arg_attr=[
'--name', name,
'--description', description],
name_attr=['name', 'description'],
val_attr=[name, description])
def test_delete_tap_flow(self):
# Delete tap_flow: myid.
cmd = tapflow.DeleteTapFlow(test_cli20.MyApp(sys.stdout), None)
myid = 'myid'
args = [myid]
self._test_delete_resource(self.resource, cmd, myid, args)
def test_update_tap_flow(self):
# Update tap_flow: myid --name myname.
cmd = tapflow.UpdateTapFlow(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(self.resource, cmd, 'myid',
['myid', '--name', 'myname'],
{'name': 'myname'})
def test_list_tap_flows(self):
# List tap_flows.
cmd = tapflow.ListTapFlow(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(self.resources, cmd, True)
def test_show_tap_flow(self):
# Show tap_flow: --fields id --fields name myid.
cmd = tapflow.ShowTapFlow(test_cli20.MyApp(sys.stdout), None)
args = ['--fields', 'id', '--fields', 'name', self.test_id]
self._test_show_resource(self.resource, cmd, self.test_id,
args, ['id', 'name'])

View File

@ -0,0 +1,112 @@
# Copyright 2015 NEC Corporation
# 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 mock
import sys
from neutron_taas.taas_client import tapservice
from neutronclient import shell
from neutronclient.tests.unit import test_cli20
class CLITestV20TapServiceJSON(test_cli20.CLITestV20Base,
tapservice.TapService):
def setUp(self):
self._mock_extension_loading()
super(CLITestV20TapServiceJSON, self).setUp()
self.resources = self.resource_plural
self.register_non_admin_status_resource(self.resource)
def _create_patch(self, name, func=None):
patcher = mock.patch(name)
thing = patcher.start()
self.addCleanup(patcher.stop)
return thing
def _mock_extension_loading(self):
ext_pkg = 'neutronclient.common.extension'
contrib = self._create_patch(ext_pkg + '._discover_via_entry_points')
contrib.return_value = [("_tap_service", tapservice)]
return contrib
def test_ext_cmd_loaded(self):
shell.NeutronShell('2.0')
extension_cmd = {'tap-service-create': tapservice.CreateTapService,
'tap-service-delete': tapservice.DeleteTapService,
'tap-service-show': tapservice.ShowTapService,
'tap-service-list': tapservice.ListTapService}
self.assertDictContainsSubset(extension_cmd, shell.COMMANDS['2.0'])
def _test_create_tap_service(self, port_id="random_port",
network_id="random_net", name='',
args_attr=None, position_names_attr=None,
position_values_attr=None):
cmd = tapservice.CreateTapService(test_cli20.MyApp(sys.stdout), None)
args_attr = args_attr or []
position_names_attr = position_names_attr or []
position_values_attr = position_values_attr or []
name = name
tenant_id = 'my-tenant'
my_id = 'my-id'
args = ['--tenant-id', tenant_id,
'--port', port_id,
'--network', network_id] + args_attr
position_names = ['port_id', 'network_id'] + position_names_attr
position_values = [port_id, network_id] + position_values_attr
self._test_create_resource(self.resource, cmd, name, my_id, args,
position_names, position_values,
tenant_id=tenant_id)
def test_create_tap_service_mandatory_params(self):
# Create tap_service: --port random_port --network random_network
self._test_create_tap_service()
def test_create_tap_service_all_params(self):
# Create tap_service with mandatory params, --name and --description
name = 'new-tap-service'
description = 'This defines a new tap-service'
args_attr = ['--name', name, '--description', description]
position_names_attr = ['name', 'description']
position_val_attr = [name, description]
self._test_create_tap_service(name=name, args_attr=args_attr,
position_names_attr=position_names_attr,
position_values_attr=position_val_attr)
def test_delete_tap_service(self):
# Delete tap_service: myid.
cmd = tapservice.DeleteTapService(test_cli20.MyApp(sys.stdout), None)
myid = 'myid'
args = [myid]
self._test_delete_resource(self.resource, cmd, myid, args)
def test_update_tap_service(self):
# Update tap_service: myid --name myname.
cmd = tapservice.UpdateTapService(test_cli20.MyApp(sys.stdout), None)
self._test_update_resource(self.resource, cmd, 'myid',
['myid', '--name', 'myname'],
{'name': 'myname'})
def test_list_tap_services(self):
# List tap_services.
cmd = tapservice.ListTapService(test_cli20.MyApp(sys.stdout), None)
self._test_list_resources(self.resources, cmd, True)
def test_show_tap_service(self):
# Show tap_service: --fields id --fields name myid.
cmd = tapservice.ShowTapService(test_cli20.MyApp(sys.stdout), None)
args = ['--fields', 'id', '--fields', 'name', self.test_id]
self._test_show_resource(self.resource, cmd, self.test_id,
args, ['id', 'name'])

View File

@ -54,3 +54,6 @@ neutron.db.alembic_migrations =
tap-as-a-service = neutron_taas.db.migration:alembic_migration
tempest.test_plugins =
tap-as-a-service = neutron_taas.tests.tempest_plugin.plugin:NeutronTaaSPlugin
neutronclient.extension =
tap_service = neutron_taas.taas_client.tapservice
tap_flow = neutron_taas.taas_client.tapflow