summaryrefslogtreecommitdiff
path: root/openstackclient
diff options
context:
space:
mode:
authorYang JianFeng <yjf1970231893@gmail.com>2020-08-31 11:39:48 +0000
committeryangjianfeng <yjf1970231893@gmail.com>2022-08-26 08:15:43 +0800
commitb36cd0f4c0416182ce69d71a4e99d89507d53e7f (patch)
tree52faf7f0694c9896e35b170e9383148bc986a11f /openstackclient
parent27b2496e03ceb95b56b3b1b68a57058fe90251b5 (diff)
downloadpython-openstackclient-b36cd0f4c0416182ce69d71a4e99d89507d53e7f.tar.gz
Add router ndp proxy commands
Depends-on: https://review.opendev.org/749036 Change-Id: I77e12cc2dfe4000bd5ae6511878c6591f52d9791 Related-Bug: #1877301
Diffstat (limited to 'openstackclient')
-rw-r--r--openstackclient/network/v2/ndp_proxy.py269
-rw-r--r--openstackclient/network/v2/router.py41
-rw-r--r--openstackclient/tests/functional/network/v2/test_network_ndp_proxy.py176
-rw-r--r--openstackclient/tests/unit/network/v2/fakes.py73
-rw-r--r--openstackclient/tests/unit/network/v2/test_ndp_proxy.py454
5 files changed, 1013 insertions, 0 deletions
diff --git a/openstackclient/network/v2/ndp_proxy.py b/openstackclient/network/v2/ndp_proxy.py
new file mode 100644
index 00000000..25b287f3
--- /dev/null
+++ b/openstackclient/network/v2/ndp_proxy.py
@@ -0,0 +1,269 @@
+# Copyright (c) 2020 Troila.
+# 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.
+
+"""Router NDP proxy action implementations"""
+import logging
+
+from osc_lib.command import command
+from osc_lib import exceptions
+from osc_lib import utils
+
+from openstackclient.i18n import _
+from openstackclient.identity import common as identity_common
+
+
+LOG = logging.getLogger(__name__)
+
+
+def _get_columns(item):
+ column_map = {}
+ hidden_columns = ['location']
+ return utils.get_osc_show_columns_for_sdk_resource(
+ item, column_map, hidden_columns)
+
+
+class CreateNDPProxy(command.ShowOne):
+ _description = _("Create NDP proxy")
+
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ parser.add_argument(
+ 'router',
+ metavar='<router>',
+ help=_("The name or ID of a router"))
+ parser.add_argument(
+ '--name',
+ metavar='<name>',
+ help=_("New NDP proxy name")
+ )
+ parser.add_argument(
+ '--port',
+ metavar='<port>',
+ required=True,
+ help=_("The name or ID of the network port associated "
+ "to the NDP proxy"))
+ parser.add_argument(
+ '--ip-address',
+ metavar='<ip-address>',
+ help=_("The IPv6 address that is to be proxied. In case the port "
+ "has multiple addresses assigned, use this option to "
+ "select which address is to be used."))
+ parser.add_argument(
+ '--description',
+ metavar='<description>',
+ help=_("A text to describe/contextualize the use of the "
+ "NDP proxy configuration")
+ )
+
+ return parser
+
+ def take_action(self, parsed_args):
+ attrs = {'name': parsed_args.name}
+ client = self.app.client_manager.network
+ router = client.find_router(
+ parsed_args.router,
+ ignore_missing=False,
+ )
+ attrs['router_id'] = router.id
+
+ if parsed_args.ip_address:
+ attrs['ip_address'] = parsed_args.ip_address
+
+ port = client.find_port(parsed_args.port,
+ ignore_missing=False)
+ attrs['port_id'] = port.id
+
+ if parsed_args.description is not None:
+ attrs['description'] = parsed_args.description
+
+ obj = client.create_ndp_proxy(**attrs)
+ display_columns, columns = _get_columns(obj)
+ data = utils.get_item_properties(obj, columns)
+ return (display_columns, data)
+
+
+class DeleteNDPProxy(command.Command):
+ _description = _("Delete NDP proxy")
+
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ parser.add_argument(
+ 'ndp_proxy',
+ nargs="+",
+ metavar="<ndp-proxy>",
+ help=_("NDP proxy(s) to delete (name or ID)")
+ )
+ return parser
+
+ def take_action(self, parsed_args):
+ client = self.app.client_manager.network
+ result = 0
+
+ for ndp_proxy in parsed_args.ndp_proxy:
+ try:
+ obj = client.find_ndp_proxy(ndp_proxy, ignore_missing=False)
+ client.delete_ndp_proxy(obj)
+ except Exception as e:
+ result += 1
+ LOG.error(_("Failed to delete NDP proxy "
+ "'%(ndp_proxy)s': %(e)s"),
+ {'ndp_proxy': ndp_proxy, 'e': e})
+ if result > 0:
+ total = len(parsed_args.ndp_proxy)
+ msg = (_("%(result)s of %(total)s NDP Proxy failed "
+ "to delete.") % {'result': result, 'total': total})
+ raise exceptions.CommandError(msg)
+
+
+class ListNDPProxy(command.Lister):
+ _description = _("List NDP proxies")
+
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ parser.add_argument(
+ '--router',
+ metavar='<router>',
+ help=_("List only NDP proxies belong to this router (name or ID)")
+ )
+ parser.add_argument(
+ '--port',
+ metavar='<port>',
+ help=_("List only NDP proxies assocate to this port (name or ID)")
+ )
+ parser.add_argument(
+ '--ip-address',
+ metavar='ip-address',
+ help=_("List only NDP proxies according to their IPv6 address")
+ )
+ parser.add_argument(
+ '--project',
+ metavar='<project>',
+ help=_("List NDP proxies according to their project (name or ID)")
+ )
+ parser.add_argument(
+ '--name',
+ metavar='<name>',
+ help=_("List NDP proxies according to their name")
+ )
+
+ identity_common.add_project_domain_option_to_parser(parser)
+
+ return parser
+
+ def take_action(self, parsed_args):
+ client = self.app.client_manager.network
+ identity_client = self.app.client_manager.identity
+
+ columns = (
+ 'id',
+ 'name',
+ 'router_id',
+ 'ip_address',
+ 'project_id',
+ )
+ headers = (
+ 'ID',
+ 'Name',
+ 'Router ID',
+ 'IP Address',
+ 'Project',
+ )
+
+ query = {}
+
+ if parsed_args.router:
+ router = client.find_router(parsed_args.router,
+ ignore_missing=False)
+ query['router_id'] = router.id
+ if parsed_args.port:
+ port = client.find_port(parsed_args.port,
+ ignore_missing=False)
+ query['port_id'] = port.id
+ if parsed_args.ip_address is not None:
+ query['ip_address'] = parsed_args.ip_address
+ if parsed_args.project:
+ project_id = identity_common.find_project(
+ identity_client,
+ parsed_args.project,
+ parsed_args.project_domain,
+ ).id
+ query['project_id'] = project_id
+ if parsed_args.name:
+ query['name'] = parsed_args.name
+
+ data = client.ndp_proxies(**query)
+
+ return (headers,
+ (utils.get_item_properties(
+ s, columns,
+ formatters={},
+ ) for s in data))
+
+
+class SetNDPProxy(command.Command):
+ _description = _("Set NDP proxy properties")
+
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ parser.add_argument(
+ 'ndp_proxy',
+ metavar='<ndp-proxy>',
+ help=_("The ID or name of the NDP proxy to update")
+ )
+ parser.add_argument(
+ '--name',
+ metavar='<name>',
+ help=_("Set NDP proxy name")
+ )
+ parser.add_argument(
+ '--description',
+ metavar='<description>',
+ help=_("A text to describe/contextualize the use of "
+ "the NDP proxy configuration")
+ )
+ return parser
+
+ def take_action(self, parsed_args):
+ client = self.app.client_manager.network
+ attrs = {}
+ if parsed_args.description is not None:
+ attrs['description'] = parsed_args.description
+ if parsed_args.name is not None:
+ attrs['name'] = parsed_args.name
+
+ obj = client.find_ndp_proxy(
+ parsed_args.ndp_proxy, ignore_missing=False)
+ client.update_ndp_proxy(obj, **attrs)
+
+
+class ShowNDPProxy(command.ShowOne):
+ _description = _("Display NDP proxy details")
+
+ def get_parser(self, prog_name):
+ parser = super().get_parser(prog_name)
+ parser.add_argument(
+ 'ndp_proxy',
+ metavar="<ndp-proxy>",
+ help=_("The ID or name of the NDP proxy")
+ )
+ return parser
+
+ def take_action(self, parsed_args):
+ client = self.app.client_manager.network
+ obj = client.find_ndp_proxy(parsed_args.ndp_proxy,
+ ignore_missing=False)
+ display_columns, columns = _get_columns(obj)
+ data = utils.get_item_properties(obj, columns)
+ return (display_columns, data)
diff --git a/openstackclient/network/v2/router.py b/openstackclient/network/v2/router.py
index f1fce298..8302ee01 100644
--- a/openstackclient/network/v2/router.py
+++ b/openstackclient/network/v2/router.py
@@ -369,6 +369,21 @@ class CreateRouter(command.ShowOne, common.NeutronCommandWithExtraArgs):
action='store_true',
help=_("Disable Source NAT on external gateway")
)
+ ndp_proxy_group = parser.add_mutually_exclusive_group()
+ ndp_proxy_group.add_argument(
+ '--enable-ndp-proxy',
+ dest='enable_ndp_proxy',
+ default=None,
+ action='store_true',
+ help=_("Enable IPv6 NDP proxy on external gateway")
+ )
+ ndp_proxy_group.add_argument(
+ '--disable-ndp-proxy',
+ dest='enable_ndp_proxy',
+ default=None,
+ action='store_false',
+ help=_("Disable IPv6 NDP proxy on external gateway")
+ )
return parser
@@ -383,6 +398,14 @@ class CreateRouter(command.ShowOne, common.NeutronCommandWithExtraArgs):
attrs.update(
self._parse_extra_properties(parsed_args.extra_properties))
+ if parsed_args.enable_ndp_proxy and not parsed_args.external_gateway:
+ msg = (_("You must specify '--external-gateway' in order "
+ "to enable router's NDP proxy"))
+ raise exceptions.CommandError(msg)
+
+ if parsed_args.enable_ndp_proxy is not None:
+ attrs['enable_ndp_proxy'] = parsed_args.enable_ndp_proxy
+
obj = client.create_router(**attrs)
# tags cannot be set when created, so tags need to be set later.
_tag.update_tags_for_set(client, obj, parsed_args)
@@ -737,6 +760,21 @@ class SetRouter(common.NeutronCommandWithExtraArgs):
action='store_true',
help=_("Disable Source NAT on external gateway")
)
+ ndp_proxy_group = parser.add_mutually_exclusive_group()
+ ndp_proxy_group.add_argument(
+ '--enable-ndp-proxy',
+ dest='enable_ndp_proxy',
+ default=None,
+ action='store_true',
+ help=_("Enable IPv6 NDP proxy on external gateway")
+ )
+ ndp_proxy_group.add_argument(
+ '--disable-ndp-proxy',
+ dest='enable_ndp_proxy',
+ default=None,
+ action='store_false',
+ help=_("Disable IPv6 NDP proxy on external gateway")
+ )
qos_policy_group = parser.add_mutually_exclusive_group()
qos_policy_group.add_argument(
'--qos-policy',
@@ -804,6 +842,9 @@ class SetRouter(common.NeutronCommandWithExtraArgs):
attrs.update(
self._parse_extra_properties(parsed_args.extra_properties))
+ if parsed_args.enable_ndp_proxy is not None:
+ attrs['enable_ndp_proxy'] = parsed_args.enable_ndp_proxy
+
if attrs:
client.update_router(obj, **attrs)
# tags is a subresource and it needs to be updated separately.
diff --git a/openstackclient/tests/functional/network/v2/test_network_ndp_proxy.py b/openstackclient/tests/functional/network/v2/test_network_ndp_proxy.py
new file mode 100644
index 00000000..e89197ad
--- /dev/null
+++ b/openstackclient/tests/functional/network/v2/test_network_ndp_proxy.py
@@ -0,0 +1,176 @@
+# 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 json
+
+from openstackclient.tests.functional.network.v2 import common
+
+
+class L3NDPProxyTests(common.NetworkTests):
+
+ def setUp(self):
+ super().setUp()
+ # Nothing in this class works with Nova Network
+ if not self.haz_network:
+ self.skipTest("No Network service present")
+ if not self.is_extension_enabled('l3-ndp-proxy'):
+ self.skipTest("No l3-ndp-proxy extension present")
+
+ self.ROT_NAME = self.getUniqueString()
+ self.EXT_NET_NAME = self.getUniqueString()
+ self.EXT_SUB_NAME = self.getUniqueString()
+ self.INT_NET_NAME = self.getUniqueString()
+ self.INT_SUB_NAME = self.getUniqueString()
+ self.INT_PORT_NAME = self.getUniqueString()
+ self.created_ndp_proxies = []
+
+ json_output = json.loads(
+ self.openstack('network create -f json '
+ '--external ' + self.EXT_NET_NAME))
+ self.assertIsNotNone(json_output['id'])
+ self.EXT_NET_ID = json_output['id']
+ json_output = json.loads(
+ self.openstack(
+ 'subnet create -f json --ip-version 6 --subnet-range '
+ '2002::1:0/112 --network %(net_id)s %(sub_name)s' % {
+ 'net_id': self.EXT_NET_ID,
+ 'sub_name': self.EXT_SUB_NAME}))
+ self.assertIsNotNone(json_output['id'])
+ self.EXT_SUB_ID = json_output['id']
+ json_output = json.loads(
+ self.openstack('router create -f json ' + self.ROT_NAME))
+ self.assertIsNotNone(json_output['id'])
+ self.ROT_ID = json_output['id']
+ output = self.openstack(
+ 'router set %(router_id)s --external-gateway %(net_id)s' % {
+ 'router_id': self.ROT_ID,
+ 'net_id': self.EXT_NET_ID})
+ self.assertEqual('', output)
+ output = self.openstack('router set --enable-ndp-proxy ' + self.ROT_ID)
+ self.assertEqual('', output)
+ json_output = json.loads(
+ self.openstack(
+ 'router show -f json -c enable_ndp_proxy ' + self.ROT_ID))
+ self.assertTrue(json_output['enable_ndp_proxy'])
+ json_output = json.loads(
+ self.openstack('network create -f json ' + self.INT_NET_NAME))
+ self.assertIsNotNone(json_output['id'])
+ self.INT_NET_ID = json_output['id']
+ json_output = json.loads(
+ self.openstack(
+ 'subnet create -f json --ip-version 6 --subnet-range '
+ '2002::2:0/112 --network %(net_id)s %(sub_name)s' % {
+ 'net_id': self.INT_NET_ID,
+ 'sub_name': self.INT_SUB_NAME}))
+ self.assertIsNotNone(json_output['id'])
+ self.INT_SUB_ID = json_output['id']
+ json_output = json.loads(
+ self.openstack(
+ 'port create -f json --network %(net_id)s '
+ '%(port_name)s' % {
+ 'net_id': self.INT_NET_ID,
+ 'port_name': self.INT_PORT_NAME}))
+ self.assertIsNotNone(json_output['id'])
+ self.INT_PORT_ID = json_output['id']
+ self.INT_PORT_ADDRESS = json_output['fixed_ips'][0]['ip_address']
+ output = self.openstack(
+ 'router add subnet ' + self.ROT_ID + ' ' + self.INT_SUB_ID)
+ self.assertEqual('', output)
+
+ def tearDown(self):
+ for ndp_proxy in self.created_ndp_proxies:
+ output = self.openstack(
+ 'router ndp proxy delete ' + ndp_proxy['id'])
+ self.assertEqual('', output)
+ output = self.openstack('port delete ' + self.INT_PORT_ID)
+ self.assertEqual('', output)
+ output = self.openstack(
+ 'router set --disable-ndp-proxy ' + self.ROT_ID)
+ self.assertEqual('', output)
+ output = self.openstack(
+ 'router remove subnet ' + self.ROT_ID + ' ' + self.INT_SUB_ID)
+ self.assertEqual('', output)
+ output = self.openstack('subnet delete ' + self.INT_SUB_ID)
+ self.assertEqual('', output)
+ output = self.openstack('network delete ' + self.INT_NET_ID)
+ self.assertEqual('', output)
+ output = self.openstack(
+ 'router unset ' + self.ROT_ID + ' ' + '--external-gateway')
+ self.assertEqual('', output)
+ output = self.openstack('router delete ' + self.ROT_ID)
+ self.assertEqual('', output)
+ output = self.openstack('subnet delete ' + self.EXT_SUB_ID)
+ self.assertEqual('', output)
+ output = self.openstack('network delete ' + self.EXT_NET_ID)
+ self.assertEqual('', output)
+ super().tearDown()
+
+ def _create_ndp_proxies(self, ndp_proxies):
+ for ndp_proxy in ndp_proxies:
+ output = json.loads(
+ self.openstack(
+ 'router ndp proxy create %(router)s --name %(name)s '
+ '--port %(port)s --ip-address %(address)s -f json' % {
+ 'router': ndp_proxy['router_id'],
+ 'name': ndp_proxy['name'],
+ 'port': ndp_proxy['port_id'],
+ 'address': ndp_proxy['address']}))
+ self.assertEqual(ndp_proxy['router_id'], output['router_id'])
+ self.assertEqual(ndp_proxy['port_id'], output['port_id'])
+ self.assertEqual(ndp_proxy['address'], output['ip_address'])
+ self.created_ndp_proxies.append(output)
+
+ def test_create_ndp_proxy(self):
+ ndp_proxies = [
+ {
+ 'name': self.getUniqueString(),
+ 'router_id': self.ROT_ID,
+ 'port_id': self.INT_PORT_ID,
+ 'address': self.INT_PORT_ADDRESS
+ }
+ ]
+ self._create_ndp_proxies(ndp_proxies)
+
+ def test_ndp_proxy_list(self):
+ ndp_proxies = {
+ 'name': self.getUniqueString(),
+ 'router_id': self.ROT_ID,
+ 'port_id': self.INT_PORT_ID,
+ 'address': self.INT_PORT_ADDRESS}
+ self._create_ndp_proxies([ndp_proxies])
+ ndp_proxy = json.loads(self.openstack(
+ 'router ndp proxy list -f json'))[0]
+ self.assertEqual(ndp_proxies['name'], ndp_proxy['Name'])
+ self.assertEqual(ndp_proxies['router_id'], ndp_proxy['Router ID'])
+ self.assertEqual(ndp_proxies['address'], ndp_proxy['IP Address'])
+
+ def test_ndp_proxy_set_and_show(self):
+ ndp_proxies = {
+ 'name': self.getUniqueString(),
+ 'router_id': self.ROT_ID,
+ 'port_id': self.INT_PORT_ID,
+ 'address': self.INT_PORT_ADDRESS}
+ description = 'balala'
+ self._create_ndp_proxies([ndp_proxies])
+ ndp_proxy_id = self.created_ndp_proxies[0]['id']
+ output = self.openstack(
+ 'router ndp proxy set --description %s %s' % (
+ description, ndp_proxy_id))
+ self.assertEqual('', output)
+ json_output = json.loads(
+ self.openstack('router ndp proxy show -f json ' + ndp_proxy_id))
+ self.assertEqual(ndp_proxies['name'], json_output['name'])
+ self.assertEqual(ndp_proxies['router_id'], json_output['router_id'])
+ self.assertEqual(ndp_proxies['port_id'], json_output['port_id'])
+ self.assertEqual(ndp_proxies['address'], json_output['ip_address'])
+ self.assertEqual(description, json_output['description'])
diff --git a/openstackclient/tests/unit/network/v2/fakes.py b/openstackclient/tests/unit/network/v2/fakes.py
index 06c58220..4d029a0e 100644
--- a/openstackclient/tests/unit/network/v2/fakes.py
+++ b/openstackclient/tests/unit/network/v2/fakes.py
@@ -26,6 +26,7 @@ from openstack.network.v2 import availability_zone as _availability_zone
from openstack.network.v2 import flavor as _flavor
from openstack.network.v2 import local_ip as _local_ip
from openstack.network.v2 import local_ip_association as _local_ip_association
+from openstack.network.v2 import ndp_proxy as _ndp_proxy
from openstack.network.v2 import network as _network
from openstack.network.v2 import network_ip_availability as _ip_availability
from openstack.network.v2 import network_segment_range as _segment_range
@@ -2079,3 +2080,75 @@ def get_local_ip_associations(local_ip_associations=None, count=2):
local_ip_associations = create_local_ip_associations(count)
return mock.Mock(side_effect=local_ip_associations)
+
+
+def create_one_ndp_proxy(attrs=None):
+ """Create a fake NDP proxy.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with router_id, port_id, etc.
+ """
+ attrs = attrs or {}
+ router_id = (
+ attrs.get('router_id') or 'router-id-' + uuid.uuid4().hex
+ )
+ port_id = (
+ attrs.get('port_id') or 'port-id-' + uuid.uuid4().hex
+ )
+ # Set default attributes.
+ np_attrs = {
+ 'id': uuid.uuid4().hex,
+ 'name': 'ndp-proxy-name-' + uuid.uuid4().hex,
+ 'router_id': router_id,
+ 'port_id': port_id,
+ 'ip_address': '2001::1:2',
+ 'description': 'ndp-proxy-description-' + uuid.uuid4().hex,
+ 'project_id': 'project-id-' + uuid.uuid4().hex,
+ 'location': 'MUNCHMUNCHMUNCH',
+ }
+
+ # Overwrite default attributes.
+ np_attrs.update(attrs)
+
+ return _ndp_proxy.NDPProxy(**np_attrs)
+
+
+def create_ndp_proxies(attrs=None, count=2):
+ """Create multiple fake NDP proxies.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of NDP proxxy to fake
+ :return:
+ A list of FakeResource objects faking the NDP proxies
+ """
+ ndp_proxies = []
+ for i in range(0, count):
+ ndp_proxies.append(
+ create_one_ndp_proxy(attrs)
+ )
+ return ndp_proxies
+
+
+def get_ndp_proxies(ndp_proxies=None, count=2):
+ """Get a list of faked NDP proxies.
+
+ If ndp_proxy list is provided, then initialize the Mock object
+ with the list. Otherwise create one.
+
+ :param List ndp_proxies:
+ A list of FakeResource objects faking ndp proxy
+ :param int count:
+ The number of ndp proxy to fake
+ :return:
+ An iterable Mock object with side_effect set to a list of faked
+ ndp proxy
+ """
+ if ndp_proxies is None:
+ ndp_proxies = (
+ create_ndp_proxies(count)
+ )
+ return mock.Mock(side_effect=ndp_proxies)
diff --git a/openstackclient/tests/unit/network/v2/test_ndp_proxy.py b/openstackclient/tests/unit/network/v2/test_ndp_proxy.py
new file mode 100644
index 00000000..48c5deb2
--- /dev/null
+++ b/openstackclient/tests/unit/network/v2/test_ndp_proxy.py
@@ -0,0 +1,454 @@
+# 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 unittest import mock
+from unittest.mock import call
+
+from osc_lib import exceptions
+
+from openstackclient.network.v2 import ndp_proxy
+from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes_v3
+from openstackclient.tests.unit.network.v2 import fakes as network_fakes
+from openstackclient.tests.unit import utils as tests_utils
+
+
+class TestNDPProxy(network_fakes.TestNetworkV2):
+
+ def setUp(self):
+ super(TestNDPProxy, self).setUp()
+ # Get a shortcut to the ProjectManager Mock
+ self.projects_mock = self.app.client_manager.identity.projects
+ # Get a shortcut to the DomainManager Mock
+ self.domains_mock = self.app.client_manager.identity.domains
+ # Get a shortcut to the network client
+ self.network = self.app.client_manager.network
+ self.router = network_fakes.FakeRouter.create_one_router(
+ {'id': 'fake-router-id'})
+ self.network.find_router = mock.Mock(return_value=self.router)
+ self.port = network_fakes.create_one_port()
+ self.network.find_port = mock.Mock(return_value=self.port)
+
+
+class TestCreateNDPProxy(TestNDPProxy):
+ def setUp(self):
+ super(TestCreateNDPProxy, self).setUp()
+ attrs = {'router_id': self.router.id, 'port_id': self.port.id}
+ self.ndp_proxy = (
+ network_fakes.create_one_ndp_proxy(
+ attrs))
+ self.columns = (
+ 'created_at',
+ 'description',
+ 'id',
+ 'ip_address',
+ 'name',
+ 'port_id',
+ 'project_id',
+ 'revision_number',
+ 'router_id',
+ 'updated_at')
+
+ self.data = (
+ self.ndp_proxy.created_at,
+ self.ndp_proxy.description,
+ self.ndp_proxy.id,
+ self.ndp_proxy.ip_address,
+ self.ndp_proxy.name,
+ self.ndp_proxy.port_id,
+ self.ndp_proxy.project_id,
+ self.ndp_proxy.revision_number,
+ self.ndp_proxy.router_id,
+ self.ndp_proxy.updated_at
+ )
+ self.network.create_ndp_proxy = mock.Mock(
+ return_value=self.ndp_proxy)
+
+ # Get the command object to test
+ self.cmd = ndp_proxy.CreateNDPProxy(self.app, self.namespace)
+
+ def test_create_no_options(self):
+ arglist = []
+ verifylist = []
+
+ # Missing required args should bail here
+ self.assertRaises(tests_utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_create_all_options(self):
+ arglist = [
+ self.ndp_proxy.router_id,
+ '--name', self.ndp_proxy.name,
+ '--port', self.ndp_proxy.port_id,
+ '--ip-address', self.ndp_proxy.ip_address,
+ '--description', self.ndp_proxy.description,
+ ]
+ verifylist = [
+ ('name', self.ndp_proxy.name),
+ ('router', self.ndp_proxy.router_id),
+ ('port', self.ndp_proxy.port_id),
+ ('ip_address', self.ndp_proxy.ip_address),
+ ('description', self.ndp_proxy.description),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.network.create_ndp_proxy.assert_called_once_with(
+ **{'name': self.ndp_proxy.name,
+ 'router_id': self.ndp_proxy.router_id,
+ 'ip_address': self.ndp_proxy.ip_address,
+ 'port_id': self.ndp_proxy.port_id,
+ 'description': self.ndp_proxy.description})
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+
+class TestDeleteNDPProxy(TestNDPProxy):
+
+ def setUp(self):
+ super(TestDeleteNDPProxy, self).setUp()
+ attrs = {'router_id': self.router.id, 'port_id': self.port.id}
+ self.ndp_proxies = (
+ network_fakes.create_ndp_proxies(attrs))
+ self.ndp_proxy = self.ndp_proxies[0]
+ self.network.delete_ndp_proxy = mock.Mock(
+ return_value=None)
+ self.network.find_ndp_proxy = mock.Mock(
+ return_value=self.ndp_proxy)
+
+ # Get the command object to test
+ self.cmd = ndp_proxy.DeleteNDPProxy(self.app, self.namespace)
+
+ def test_delete(self):
+ arglist = [
+ self.ndp_proxy.id
+ ]
+ verifylist = [
+ ('ndp_proxy', [self.ndp_proxy.id])
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.network.delete_ndp_proxy.assert_called_once_with(self.ndp_proxy)
+ self.assertIsNone(result)
+
+ def test_delete_error(self):
+ arglist = [
+ self.ndp_proxy.id,
+ ]
+ verifylist = [
+ ('ndp_proxy', [self.ndp_proxy.id])
+ ]
+ self.network.delete_ndp_proxy.side_effect = Exception(
+ 'Error message')
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ self.assertRaises(
+ exceptions.CommandError,
+ self.cmd.take_action, parsed_args)
+
+ def test_multi_ndp_proxies_delete(self):
+ arglist = []
+ np_id = []
+
+ for a in self.ndp_proxies:
+ arglist.append(a.id)
+ np_id.append(a.id)
+
+ verifylist = [
+ ('ndp_proxy', np_id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.network.delete_ndp_proxy.assert_has_calls(
+ [call(self.ndp_proxy), call(self.ndp_proxy)])
+ self.assertIsNone(result)
+
+
+class TestListNDPProxy(TestNDPProxy):
+
+ def setUp(self):
+ super(TestListNDPProxy, self).setUp()
+ attrs = {'router_id': self.router.id, 'port_id': self.port.id}
+ ndp_proxies = (
+ network_fakes.create_ndp_proxies(attrs, count=3))
+ self.columns = (
+ 'ID',
+ 'Name',
+ 'Router ID',
+ 'IP Address',
+ 'Project',
+ )
+ self.data = []
+ for np in ndp_proxies:
+ self.data.append((
+ np.id,
+ np.name,
+ np.router_id,
+ np.ip_address,
+ np.project_id,
+ ))
+
+ self.network.ndp_proxies = mock.Mock(
+ return_value=ndp_proxies)
+
+ # Get the command object to test
+ self.cmd = ndp_proxy.ListNDPProxy(self.app, self.namespace)
+
+ def test_ndp_proxy_list(self):
+ arglist = []
+ verifylist = []
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.network.ndp_proxies.assert_called_once_with()
+ self.assertEqual(self.columns, columns)
+ list_data = list(data)
+ self.assertEqual(len(self.data), len(list_data))
+ for index in range(len(list_data)):
+ self.assertEqual(self.data[index], list_data[index])
+
+ def test_ndp_proxy_list_router(self):
+ arglist = [
+ '--router', 'fake-router-name',
+ ]
+
+ verifylist = [
+ ('router', 'fake-router-name')
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.network.ndp_proxies.assert_called_once_with(**{
+ 'router_id': 'fake-router-id'})
+ self.assertEqual(self.columns, columns)
+ self.assertCountEqual(self.data, list(data))
+
+ def test_ndp_proxy_list_port(self):
+ arglist = [
+ '--port', self.port.id,
+ ]
+
+ verifylist = [
+ ('port', self.port.id)
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.network.ndp_proxies.assert_called_once_with(**{
+ 'port_id': self.port.id})
+ self.assertEqual(self.columns, columns)
+ self.assertCountEqual(self.data, list(data))
+
+ def test_ndp_proxy_list_name(self):
+ arglist = [
+ '--name', 'fake-ndp-proxy-name',
+ ]
+
+ verifylist = [
+ ('name', 'fake-ndp-proxy-name')
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.network.ndp_proxies.assert_called_once_with(**{
+ 'name': 'fake-ndp-proxy-name'})
+ self.assertEqual(self.columns, columns)
+ self.assertCountEqual(self.data, list(data))
+
+ def test_ndp_proxy_list_ip_address(self):
+ arglist = [
+ '--ip-address', '2001::1:2',
+ ]
+
+ verifylist = [
+ ('ip_address', '2001::1:2')
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.network.ndp_proxies.assert_called_once_with(**{
+ 'ip_address': '2001::1:2'})
+ self.assertEqual(self.columns, columns)
+ self.assertCountEqual(self.data, list(data))
+
+ def test_ndp_proxy_list_project(self):
+ project = identity_fakes_v3.FakeProject.create_one_project()
+ self.projects_mock.get.return_value = project
+ arglist = [
+ '--project', project.id,
+ ]
+ verifylist = [
+ ('project', project.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.network.ndp_proxies.assert_called_once_with(
+ **{'project_id': project.id})
+ self.assertEqual(self.columns, columns)
+ self.assertItemsEqual(self.data, list(data))
+
+ def test_ndp_proxy_list_project_domain(self):
+ project = identity_fakes_v3.FakeProject.create_one_project()
+ self.projects_mock.get.return_value = project
+ arglist = [
+ '--project', project.id,
+ '--project-domain', project.domain_id,
+ ]
+ verifylist = [
+ ('project', project.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+ filters = {'project_id': project.id}
+
+ self.network.ndp_proxies.assert_called_once_with(**filters)
+ self.assertEqual(self.columns, columns)
+ self.assertItemsEqual(self.data, list(data))
+
+
+class TestSetNDPProxy(TestNDPProxy):
+
+ def setUp(self):
+ super(TestSetNDPProxy, self).setUp()
+ attrs = {'router_id': self.router.id, 'port_id': self.port.id}
+ self.ndp_proxy = (
+ network_fakes.create_one_ndp_proxy(attrs))
+ self.network.update_ndp_proxy = mock.Mock(return_value=None)
+ self.network.find_ndp_proxy = mock.Mock(
+ return_value=self.ndp_proxy)
+
+ # Get the command object to test
+ self.cmd = ndp_proxy.SetNDPProxy(self.app, self.namespace)
+
+ def test_set_nothing(self):
+ arglist = [
+ self.ndp_proxy.id,
+ ]
+ verifylist = [
+ ('ndp_proxy', self.ndp_proxy.id),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = (self.cmd.take_action(parsed_args))
+
+ self.network.update_ndp_proxy.assert_called_once_with(
+ self.ndp_proxy)
+ self.assertIsNone(result)
+
+ def test_set_name(self):
+ arglist = [
+ self.ndp_proxy.id,
+ '--name', 'fake-name',
+ ]
+ verifylist = [
+ ('ndp_proxy', self.ndp_proxy.id),
+ ('name', 'fake-name'),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = (self.cmd.take_action(parsed_args))
+
+ self.network.update_ndp_proxy.assert_called_once_with(
+ self.ndp_proxy, name='fake-name')
+ self.assertIsNone(result)
+
+ def test_set_description(self):
+ arglist = [
+ self.ndp_proxy.id,
+ '--description', 'balala',
+ ]
+ verifylist = [
+ ('ndp_proxy', self.ndp_proxy.id),
+ ('description', 'balala'),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = (self.cmd.take_action(parsed_args))
+
+ self.network.update_ndp_proxy.assert_called_once_with(
+ self.ndp_proxy, description='balala')
+ self.assertIsNone(result)
+
+
+class TestShowNDPProxy(TestNDPProxy):
+
+ def setUp(self):
+ super(TestShowNDPProxy, self).setUp()
+ attrs = {'router_id': self.router.id, 'port_id': self.port.id}
+ self.ndp_proxy = (
+ network_fakes.create_one_ndp_proxy(attrs))
+
+ self.columns = (
+ 'created_at',
+ 'description',
+ 'id',
+ 'ip_address',
+ 'name',
+ 'port_id',
+ 'project_id',
+ 'revision_number',
+ 'router_id',
+ 'updated_at')
+
+ self.data = (
+ self.ndp_proxy.created_at,
+ self.ndp_proxy.description,
+ self.ndp_proxy.id,
+ self.ndp_proxy.ip_address,
+ self.ndp_proxy.name,
+ self.ndp_proxy.port_id,
+ self.ndp_proxy.project_id,
+ self.ndp_proxy.revision_number,
+ self.ndp_proxy.router_id,
+ self.ndp_proxy.updated_at
+ )
+ self.network.get_ndp_proxy = mock.Mock(return_value=self.ndp_proxy)
+ self.network.find_ndp_proxy = mock.Mock(return_value=self.ndp_proxy)
+
+ # Get the command object to test
+ self.cmd = ndp_proxy.ShowNDPProxy(self.app, self.namespace)
+
+ def test_show_no_options(self):
+ arglist = []
+ verifylist = []
+
+ # Missing required args should bail here
+ self.assertRaises(tests_utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_show_default_options(self):
+ arglist = [
+ self.ndp_proxy.id,
+ ]
+ verifylist = [
+ ('ndp_proxy', self.ndp_proxy.id),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = (self.cmd.take_action(parsed_args))
+
+ self.network.find_ndp_proxy.assert_called_once_with(
+ self.ndp_proxy.id, ignore_missing=False)
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)