Remove the old CLI code from taas

With [1] all the CLI code was removed from python-neutronclient,
This causes test discovery failure for tap-as-a-servic, and
the old neutron tap-service-create type commads can be removed from
the tap-as-a-service repository.

[1]: https://review.opendev.org/c/openstack/python-neutronclient/+/871711

Change-Id: I44d2009ca661ff88d7b4a2ed89528aefc3a1f2eb
Related-Bug: #2003861
This commit is contained in:
elajkat 2023-05-16 11:22:53 +02:00
parent 30cbe66dba
commit bdc5f4b951
5 changed files with 0 additions and 463 deletions

View File

@ -1,128 +0,0 @@
# Copyright (C) 2018 AT&T
# 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 neutron_taas._i18n import _
from neutronclient.common import extension
from neutronclient.common import utils
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', 'status',
'vlan_filter']
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(
'--tap-service',
required=True,
metavar="TAP_SERVICE",
help=_('Tap Service to which the Tap 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.'))
parser.add_argument(
'--vlan-filter',
required=False,
metavar="VLAN_FILTER",
help=_('VLAN Ids to be mirrored in the form of range string.'))
def args2body(self, parsed_args):
client = self.get_client()
source_port = neutronv20.find_resourceid_by_name_or_id(
client, 'port',
parsed_args.port)
tap_service_id = neutronv20.find_resourceid_by_name_or_id(
client, 'tap_service',
parsed_args.tap_service)
body = {'source_port': source_port,
'tap_service_id': tap_service_id}
neutronv20.update_dict(parsed_args, body, ['tenant_id', 'direction',
'vlan_filter'])
_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

@ -1,105 +0,0 @@
# 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 neutron_taas._i18n import _
from neutronclient.common import extension
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', 'status']
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']
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.'))
def args2body(self, parsed_args):
client = self.get_client()
port_id = neutronv20.find_resourceid_by_name_or_id(
client, 'port',
parsed_args.port_id)
body = {'port_id': port_id}
if parsed_args.tenant_id:
body['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

@ -1,114 +0,0 @@
# 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 sys
from unittest import mock
from neutron_taas.taas_client import tapflow
from neutronclient import shell
from neutronclient.tests.unit import test_cli20
class CLITestV20TapFlowJSON(test_cli20.CLITestV20Base):
resource = 'tap_flow'
resource_plural = '%ss' % resource
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()
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):
neutron_shell = 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}
for cmd_name, cmd_class in extension_cmd.items():
found = neutron_shell.command_manager.find_command([cmd_name])
self.assertEqual(cmd_class, found[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,
'--tap-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

@ -1,113 +0,0 @@
# 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 sys
from unittest import mock
from neutron_taas.taas_client import tapservice
from neutronclient import shell
from neutronclient.tests.unit import test_cli20
class CLITestV20TapServiceJSON(test_cli20.CLITestV20Base):
resource = 'tap_service'
resource_plural = '%ss' % resource
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()
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):
neutron_shell = 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}
for cmd_name, cmd_class in extension_cmd.items():
found = neutron_shell.command_manager.find_command([cmd_name])
self.assertEqual(cmd_class, found[0])
def _test_create_tap_service(self, port_id="random_port",
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] + args_attr
position_names = ['port_id'] + position_names_attr
position_values = [port_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
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

@ -65,9 +65,6 @@ neutron.service_plugins =
taas = neutron_taas.services.taas.taas_plugin:TaasPlugin taas = neutron_taas.services.taas.taas_plugin:TaasPlugin
neutron.db.alembic_migrations = neutron.db.alembic_migrations =
tap-as-a-service = neutron_taas.db.migration:alembic_migration tap-as-a-service = neutron_taas.db.migration:alembic_migration
neutronclient.extension =
tap_service = neutron_taas.taas_client.tapservice
tap_flow = neutron_taas.taas_client.tapflow
oslo.policy.policies = oslo.policy.policies =
tap-as-a-service = neutron_taas.policies:list_rules tap-as-a-service = neutron_taas.policies:list_rules
neutron.policies = neutron.policies =