summaryrefslogtreecommitdiff
path: root/openstackclient
diff options
context:
space:
mode:
authorZuul <zuul@review.opendev.org>2021-01-11 17:38:26 +0000
committerGerrit Code Review <review@openstack.org>2021-01-11 17:38:26 +0000
commitd828b04c9f69075c1e38ff2c9cb3f26a356f11cc (patch)
treeb31e43914e97043a7b303ef7440b22ed41d6718d /openstackclient
parent8993d32188bac4bd9785576b0db5bad5575b733f (diff)
parentf57e10b903fa71d02a6e104717824d004178bbf5 (diff)
downloadpython-openstackclient-d828b04c9f69075c1e38ff2c9cb3f26a356f11cc.tar.gz
Merge "Support Neutron Address Group CRUD"
Diffstat (limited to 'openstackclient')
-rw-r--r--openstackclient/network/v2/address_group.py296
-rw-r--r--openstackclient/tests/functional/network/v2/test_address_group.py177
-rw-r--r--openstackclient/tests/unit/network/v2/fakes.py73
-rw-r--r--openstackclient/tests/unit/network/v2/test_address_group.py503
4 files changed, 1049 insertions, 0 deletions
diff --git a/openstackclient/network/v2/address_group.py b/openstackclient/network/v2/address_group.py
new file mode 100644
index 00000000..fe1e14a3
--- /dev/null
+++ b/openstackclient/network/v2/address_group.py
@@ -0,0 +1,296 @@
+# 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.
+#
+
+"""Address group action implementations"""
+
+import logging
+
+import netaddr
+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
+from openstackclient.network import sdk_utils
+
+
+LOG = logging.getLogger(__name__)
+
+
+def _get_columns(item):
+ column_map = {
+ 'tenant_id': 'project_id',
+ }
+ return sdk_utils.get_osc_show_columns_for_sdk_resource(item, column_map)
+
+
+def _format_addresses(addresses):
+ ret = []
+ for addr in addresses:
+ ret.append(str(netaddr.IPNetwork(addr)))
+ return ret
+
+
+def _get_attrs(client_manager, parsed_args):
+ attrs = {}
+ attrs['name'] = parsed_args.name
+ if parsed_args.description:
+ attrs['description'] = parsed_args.description
+ attrs['addresses'] = _format_addresses(parsed_args.address)
+ if 'project' in parsed_args and parsed_args.project is not None:
+ identity_client = client_manager.identity
+ project_id = identity_common.find_project(
+ identity_client,
+ parsed_args.project,
+ parsed_args.project_domain,
+ ).id
+ attrs['tenant_id'] = project_id
+
+ return attrs
+
+
+class CreateAddressGroup(command.ShowOne):
+ _description = _("Create a new Address Group")
+
+ def get_parser(self, prog_name):
+ parser = super(CreateAddressGroup, self).get_parser(prog_name)
+ parser.add_argument(
+ 'name',
+ metavar="<name>",
+ help=_("New address group name")
+ )
+ parser.add_argument(
+ '--description',
+ metavar="<description>",
+ help=_("New address group description")
+ )
+ parser.add_argument(
+ "--address",
+ metavar="<ip-address>",
+ action='append',
+ default=[],
+ help=_("IP address or CIDR "
+ "(repeat option to set multiple addresses)"),
+ )
+ parser.add_argument(
+ '--project',
+ metavar="<project>",
+ help=_("Owner's project (name or ID)")
+ )
+ identity_common.add_project_domain_option_to_parser(parser)
+
+ return parser
+
+ def take_action(self, parsed_args):
+ client = self.app.client_manager.network
+ attrs = _get_attrs(self.app.client_manager, parsed_args)
+
+ obj = client.create_address_group(**attrs)
+ display_columns, columns = _get_columns(obj)
+ data = utils.get_item_properties(obj, columns, formatters={})
+
+ return (display_columns, data)
+
+
+class DeleteAddressGroup(command.Command):
+ _description = _("Delete address group(s)")
+
+ def get_parser(self, prog_name):
+ parser = super(DeleteAddressGroup, self).get_parser(prog_name)
+ parser.add_argument(
+ 'address_group',
+ metavar="<address-group>",
+ nargs='+',
+ help=_("Address group(s) to delete (name or ID)")
+ )
+
+ return parser
+
+ def take_action(self, parsed_args):
+ client = self.app.client_manager.network
+ result = 0
+
+ for group in parsed_args.address_group:
+ try:
+ obj = client.find_address_group(group, ignore_missing=False)
+ client.delete_address_group(obj)
+ except Exception as e:
+ result += 1
+ LOG.error(_("Failed to delete address group with "
+ "name or ID '%(group)s': %(e)s"),
+ {'group': group, 'e': e})
+
+ if result > 0:
+ total = len(parsed_args.address_group)
+ msg = (_("%(result)s of %(total)s address groups failed "
+ "to delete.") % {'result': result, 'total': total})
+ raise exceptions.CommandError(msg)
+
+
+class ListAddressGroup(command.Lister):
+ _description = _("List address groups")
+
+ def get_parser(self, prog_name):
+ parser = super(ListAddressGroup, self).get_parser(prog_name)
+
+ parser.add_argument(
+ '--name',
+ metavar='<name>',
+ help=_("List only address groups of given name in output")
+ )
+ parser.add_argument(
+ '--project',
+ metavar="<project>",
+ help=_("List address groups according to their project "
+ "(name or ID)")
+ )
+ identity_common.add_project_domain_option_to_parser(parser)
+
+ return parser
+
+ def take_action(self, parsed_args):
+ client = self.app.client_manager.network
+ columns = (
+ 'id',
+ 'name',
+ 'description',
+ 'project_id',
+ 'addresses',
+ )
+ column_headers = (
+ 'ID',
+ 'Name',
+ 'Description',
+ 'Project',
+ 'Addresses',
+ )
+ attrs = {}
+ if parsed_args.name:
+ attrs['name'] = parsed_args.name
+ if 'project' in parsed_args and parsed_args.project is not None:
+ identity_client = self.app.client_manager.identity
+ project_id = identity_common.find_project(
+ identity_client,
+ parsed_args.project,
+ parsed_args.project_domain,
+ ).id
+ attrs['tenant_id'] = project_id
+ attrs['project_id'] = project_id
+ data = client.address_groups(**attrs)
+
+ return (column_headers,
+ (utils.get_item_properties(
+ s, columns, formatters={},
+ ) for s in data))
+
+
+class SetAddressGroup(command.Command):
+ _description = _("Set address group properties")
+
+ def get_parser(self, prog_name):
+ parser = super(SetAddressGroup, self).get_parser(prog_name)
+ parser.add_argument(
+ 'address_group',
+ metavar="<address-group>",
+ help=_("Address group to modify (name or ID)")
+ )
+ parser.add_argument(
+ '--name',
+ metavar="<name>",
+ help=_('Set address group name')
+ )
+ parser.add_argument(
+ '--description',
+ metavar="<description>",
+ help=_('Set address group description')
+ )
+ parser.add_argument(
+ "--address",
+ metavar="<ip-address>",
+ action='append',
+ default=[],
+ help=_("IP address or CIDR "
+ "(repeat option to set multiple addresses)"),
+ )
+ return parser
+
+ def take_action(self, parsed_args):
+ client = self.app.client_manager.network
+ obj = client.find_address_group(
+ parsed_args.address_group,
+ ignore_missing=False)
+ attrs = {}
+ if parsed_args.name is not None:
+ attrs['name'] = parsed_args.name
+ if parsed_args.description is not None:
+ attrs['description'] = parsed_args.description
+ if attrs:
+ client.update_address_group(obj, **attrs)
+ if parsed_args.address:
+ client.add_addresses_to_address_group(
+ obj, _format_addresses(parsed_args.address))
+
+
+class ShowAddressGroup(command.ShowOne):
+ _description = _("Display address group details")
+
+ def get_parser(self, prog_name):
+ parser = super(ShowAddressGroup, self).get_parser(prog_name)
+ parser.add_argument(
+ 'address_group',
+ metavar="<address-group>",
+ help=_("Address group to display (name or ID)")
+ )
+
+ return parser
+
+ def take_action(self, parsed_args):
+ client = self.app.client_manager.network
+ obj = client.find_address_group(
+ parsed_args.address_group,
+ ignore_missing=False)
+ display_columns, columns = _get_columns(obj)
+ data = utils.get_item_properties(obj, columns, formatters={})
+
+ return (display_columns, data)
+
+
+class UnsetAddressGroup(command.Command):
+ _description = _("Unset address group properties")
+
+ def get_parser(self, prog_name):
+ parser = super(UnsetAddressGroup, self).get_parser(prog_name)
+ parser.add_argument(
+ 'address_group',
+ metavar="<address-group>",
+ help=_("Address group to modify (name or ID)")
+ )
+ parser.add_argument(
+ "--address",
+ metavar="<ip-address>",
+ action='append',
+ default=[],
+ help=_("IP address or CIDR "
+ "(repeat option to unset multiple addresses)"),
+ )
+ return parser
+
+ def take_action(self, parsed_args):
+ client = self.app.client_manager.network
+ obj = client.find_address_group(
+ parsed_args.address_group,
+ ignore_missing=False)
+ if parsed_args.address:
+ client.remove_addresses_from_address_group(
+ obj, _format_addresses(parsed_args.address))
diff --git a/openstackclient/tests/functional/network/v2/test_address_group.py b/openstackclient/tests/functional/network/v2/test_address_group.py
new file mode 100644
index 00000000..52c628a3
--- /dev/null
+++ b/openstackclient/tests/functional/network/v2/test_address_group.py
@@ -0,0 +1,177 @@
+# 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
+import uuid
+
+from openstackclient.tests.functional.network.v2 import common
+
+
+class AddressGroupTests(common.NetworkTests):
+ """Functional tests for address group"""
+
+ def setUp(self):
+ super(AddressGroupTests, self).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('address-group'):
+ self.skipTest("No address-group extension present")
+
+ def test_address_group_create_and_delete(self):
+ """Test create, delete multiple"""
+ name1 = uuid.uuid4().hex
+ cmd_output = json.loads(self.openstack(
+ 'address group create -f json ' +
+ name1
+ ))
+ self.assertEqual(
+ name1,
+ cmd_output['name'],
+ )
+
+ name2 = uuid.uuid4().hex
+ cmd_output = json.loads(self.openstack(
+ 'address group create -f json ' +
+ name2
+ ))
+ self.assertEqual(
+ name2,
+ cmd_output['name'],
+ )
+
+ raw_output = self.openstack(
+ 'address group delete ' + name1 + ' ' + name2,
+ )
+ self.assertOutput('', raw_output)
+
+ def test_address_group_list(self):
+ """Test create, list filters, delete"""
+ # Get project IDs
+ cmd_output = json.loads(self.openstack('token issue -f json '))
+ auth_project_id = cmd_output['project_id']
+
+ cmd_output = json.loads(self.openstack('project list -f json '))
+ admin_project_id = None
+ demo_project_id = None
+ for p in cmd_output:
+ if p['Name'] == 'admin':
+ admin_project_id = p['ID']
+ if p['Name'] == 'demo':
+ demo_project_id = p['ID']
+
+ # Verify assumptions:
+ # * admin and demo projects are present
+ # * demo and admin are distinct projects
+ # * tests run as admin
+ self.assertIsNotNone(admin_project_id)
+ self.assertIsNotNone(demo_project_id)
+ self.assertNotEqual(admin_project_id, demo_project_id)
+ self.assertEqual(admin_project_id, auth_project_id)
+
+ name1 = uuid.uuid4().hex
+ cmd_output = json.loads(self.openstack(
+ 'address group create -f json ' +
+ name1
+ ))
+ self.addCleanup(self.openstack, 'address group delete ' + name1)
+ self.assertEqual(
+ admin_project_id,
+ cmd_output["project_id"],
+ )
+
+ name2 = uuid.uuid4().hex
+ cmd_output = json.loads(self.openstack(
+ 'address group create -f json ' +
+ '--project ' + demo_project_id +
+ ' ' + name2
+ ))
+ self.addCleanup(self.openstack, 'address group delete ' + name2)
+ self.assertEqual(
+ demo_project_id,
+ cmd_output["project_id"],
+ )
+
+ # Test list
+ cmd_output = json.loads(self.openstack(
+ 'address group list -f json ',
+ ))
+ names = [x["Name"] for x in cmd_output]
+ self.assertIn(name1, names)
+ self.assertIn(name2, names)
+
+ # Test list --project
+ cmd_output = json.loads(self.openstack(
+ 'address group list -f json ' +
+ '--project ' + demo_project_id
+ ))
+ names = [x["Name"] for x in cmd_output]
+ self.assertNotIn(name1, names)
+ self.assertIn(name2, names)
+
+ # Test list --name
+ cmd_output = json.loads(self.openstack(
+ 'address group list -f json ' +
+ '--name ' + name1
+ ))
+ names = [x["Name"] for x in cmd_output]
+ self.assertIn(name1, names)
+ self.assertNotIn(name2, names)
+
+ def test_address_group_set_unset_and_show(self):
+ """Tests create options, set, unset, and show"""
+ name = uuid.uuid4().hex
+ newname = name + "_"
+ cmd_output = json.loads(self.openstack(
+ 'address group create -f json ' +
+ '--description aaaa ' +
+ '--address 10.0.0.1 --address 2001::/16 ' +
+ name
+ ))
+ self.addCleanup(self.openstack, 'address group delete ' + newname)
+ self.assertEqual(name, cmd_output['name'])
+ self.assertEqual('aaaa', cmd_output['description'])
+ self.assertEqual(2, len(cmd_output['addresses']))
+
+ # Test set name, description and address
+ raw_output = self.openstack(
+ 'address group set ' +
+ '--name ' + newname + ' ' +
+ '--description bbbb ' +
+ '--address 10.0.0.2 --address 192.0.0.0/8 ' +
+ name,
+ )
+ self.assertOutput('', raw_output)
+
+ # Show the updated address group
+ cmd_output = json.loads(self.openstack(
+ 'address group show -f json ' +
+ newname,
+ ))
+ self.assertEqual(newname, cmd_output['name'])
+ self.assertEqual('bbbb', cmd_output['description'])
+ self.assertEqual(4, len(cmd_output['addresses']))
+
+ # Test unset address
+ raw_output = self.openstack(
+ 'address group unset ' +
+ '--address 10.0.0.1 --address 2001::/16 ' +
+ '--address 10.0.0.2 --address 192.0.0.0/8 ' +
+ newname,
+ )
+ self.assertEqual('', raw_output)
+
+ cmd_output = json.loads(self.openstack(
+ 'address group show -f json ' +
+ newname,
+ ))
+ self.assertEqual(0, len(cmd_output['addresses']))
diff --git a/openstackclient/tests/unit/network/v2/fakes.py b/openstackclient/tests/unit/network/v2/fakes.py
index 2db83d3b..798cfd96 100644
--- a/openstackclient/tests/unit/network/v2/fakes.py
+++ b/openstackclient/tests/unit/network/v2/fakes.py
@@ -83,6 +83,79 @@ class TestNetworkV2(utils.TestCommand):
)
+class FakeAddressGroup(object):
+ """Fake one or more address groups."""
+
+ @staticmethod
+ def create_one_address_group(attrs=None):
+ """Create a fake address group.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with name, id, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ address_group_attrs = {
+ 'name': 'address-group-name-' + uuid.uuid4().hex,
+ 'description': 'address-group-description-' + uuid.uuid4().hex,
+ 'id': 'address-group-id-' + uuid.uuid4().hex,
+ 'tenant_id': 'project-id-' + uuid.uuid4().hex,
+ 'addresses': ['10.0.0.1/32'],
+ }
+
+ # Overwrite default attributes.
+ address_group_attrs.update(attrs)
+
+ address_group = fakes.FakeResource(
+ info=copy.deepcopy(address_group_attrs),
+ loaded=True)
+
+ # Set attributes with special mapping in OpenStack SDK.
+ address_group.project_id = address_group_attrs['tenant_id']
+
+ return address_group
+
+ @staticmethod
+ def create_address_groups(attrs=None, count=2):
+ """Create multiple fake address groups.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of address groups to fake
+ :return:
+ A list of FakeResource objects faking the address groups
+ """
+ address_groups = []
+ for i in range(0, count):
+ address_groups.append(
+ FakeAddressGroup.create_one_address_group(attrs))
+
+ return address_groups
+
+ @staticmethod
+ def get_address_groups(address_groups=None, count=2):
+ """Get an iterable Mock object with a list of faked address groups.
+
+ If address groups list is provided, then initialize the Mock object
+ with the list. Otherwise create one.
+
+ :param List address_groups:
+ A list of FakeResource objects faking address groups
+ :param int count:
+ The number of address groups to fake
+ :return:
+ An iterable Mock object with side_effect set to a list of faked
+ address groups
+ """
+ if address_groups is None:
+ address_groups = FakeAddressGroup.create_address_groups(count)
+ return mock.Mock(side_effect=address_groups)
+
+
class FakeAddressScope(object):
"""Fake one or more address scopes."""
diff --git a/openstackclient/tests/unit/network/v2/test_address_group.py b/openstackclient/tests/unit/network/v2/test_address_group.py
new file mode 100644
index 00000000..3b2b1ab6
--- /dev/null
+++ b/openstackclient/tests/unit/network/v2/test_address_group.py
@@ -0,0 +1,503 @@
+# 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 address_group
+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 TestAddressGroup(network_fakes.TestNetworkV2):
+
+ def setUp(self):
+ super(TestAddressGroup, self).setUp()
+
+ # Get a shortcut to the network client
+ self.network = self.app.client_manager.network
+ # 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
+
+
+class TestCreateAddressGroup(TestAddressGroup):
+
+ project = identity_fakes_v3.FakeProject.create_one_project()
+ domain = identity_fakes_v3.FakeDomain.create_one_domain()
+ # The new address group created.
+ new_address_group = (
+ network_fakes.FakeAddressGroup.create_one_address_group(
+ attrs={
+ 'tenant_id': project.id,
+ }
+ ))
+ columns = (
+ 'addresses',
+ 'description',
+ 'id',
+ 'name',
+ 'project_id',
+ )
+ data = (
+ new_address_group.addresses,
+ new_address_group.description,
+ new_address_group.id,
+ new_address_group.name,
+ new_address_group.project_id,
+ )
+
+ def setUp(self):
+ super(TestCreateAddressGroup, self).setUp()
+ self.network.create_address_group = mock.Mock(
+ return_value=self.new_address_group)
+
+ # Get the command object to test
+ self.cmd = address_group.CreateAddressGroup(self.app, self.namespace)
+
+ self.projects_mock.get.return_value = self.project
+ self.domains_mock.get.return_value = self.domain
+
+ 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_default_options(self):
+ arglist = [
+ self.new_address_group.name,
+ ]
+ verifylist = [
+ ('project', None),
+ ('name', self.new_address_group.name),
+ ('description', None),
+ ('address', []),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = (self.cmd.take_action(parsed_args))
+
+ self.network.create_address_group.assert_called_once_with(**{
+ 'name': self.new_address_group.name,
+ 'addresses': [],
+ })
+ self.assertEqual(self.columns, columns)
+ self.assertItemsEqual(self.data, data)
+
+ def test_create_all_options(self):
+ arglist = [
+ '--project', self.project.name,
+ '--project-domain', self.domain.name,
+ '--address', '10.0.0.1',
+ '--description', self.new_address_group.description,
+ self.new_address_group.name,
+ ]
+ verifylist = [
+ ('project', self.project.name),
+ ('project_domain', self.domain.name),
+ ('address', ['10.0.0.1']),
+ ('description', self.new_address_group.description),
+ ('name', self.new_address_group.name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = (self.cmd.take_action(parsed_args))
+
+ self.network.create_address_group.assert_called_once_with(**{
+ 'addresses': ['10.0.0.1/32'],
+ 'tenant_id': self.project.id,
+ 'name': self.new_address_group.name,
+ 'description': self.new_address_group.description,
+ })
+ self.assertEqual(self.columns, columns)
+ self.assertItemsEqual(self.data, data)
+
+
+class TestDeleteAddressGroup(TestAddressGroup):
+
+ # The address group to delete.
+ _address_groups = (
+ network_fakes.FakeAddressGroup.create_address_groups(count=2))
+
+ def setUp(self):
+ super(TestDeleteAddressGroup, self).setUp()
+ self.network.delete_address_group = mock.Mock(return_value=None)
+ self.network.find_address_group = (
+ network_fakes.FakeAddressGroup.get_address_groups(
+ address_groups=self._address_groups)
+ )
+
+ # Get the command object to test
+ self.cmd = address_group.DeleteAddressGroup(self.app, self.namespace)
+
+ def test_address_group_delete(self):
+ arglist = [
+ self._address_groups[0].name,
+ ]
+ verifylist = [
+ ('address_group', [self._address_groups[0].name]),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.network.find_address_group.assert_called_once_with(
+ self._address_groups[0].name, ignore_missing=False)
+ self.network.delete_address_group.assert_called_once_with(
+ self._address_groups[0])
+ self.assertIsNone(result)
+
+ def test_multi_address_groups_delete(self):
+ arglist = []
+
+ for a in self._address_groups:
+ arglist.append(a.name)
+ verifylist = [
+ ('address_group', arglist),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ calls = []
+ for a in self._address_groups:
+ calls.append(call(a))
+ self.network.delete_address_group.assert_has_calls(calls)
+ self.assertIsNone(result)
+
+ def test_multi_address_groups_delete_with_exception(self):
+ arglist = [
+ self._address_groups[0].name,
+ 'unexist_address_group',
+ ]
+ verifylist = [
+ ('address_group',
+ [self._address_groups[0].name, 'unexist_address_group']),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ find_mock_result = [self._address_groups[0], exceptions.CommandError]
+ self.network.find_address_group = (
+ mock.Mock(side_effect=find_mock_result)
+ )
+
+ try:
+ self.cmd.take_action(parsed_args)
+ self.fail('CommandError should be raised.')
+ except exceptions.CommandError as e:
+ self.assertEqual('1 of 2 address groups failed to delete.', str(e))
+
+ self.network.find_address_group.assert_any_call(
+ self._address_groups[0].name, ignore_missing=False)
+ self.network.find_address_group.assert_any_call(
+ 'unexist_address_group', ignore_missing=False)
+ self.network.delete_address_group.assert_called_once_with(
+ self._address_groups[0]
+ )
+
+
+class TestListAddressGroup(TestAddressGroup):
+
+ # The address groups to list up.
+ address_groups = (
+ network_fakes.FakeAddressGroup.create_address_groups(count=3))
+ columns = (
+ 'ID',
+ 'Name',
+ 'Description',
+ 'Project',
+ 'Addresses',
+ )
+ data = []
+ for group in address_groups:
+ data.append((
+ group.id,
+ group.name,
+ group.description,
+ group.project_id,
+ group.addresses,
+ ))
+
+ def setUp(self):
+ super(TestListAddressGroup, self).setUp()
+ self.network.address_groups = mock.Mock(
+ return_value=self.address_groups)
+
+ # Get the command object to test
+ self.cmd = address_group.ListAddressGroup(self.app, self.namespace)
+
+ def test_address_group_list(self):
+ arglist = []
+ verifylist = []
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.network.address_groups.assert_called_once_with(**{})
+ self.assertEqual(self.columns, columns)
+ self.assertItemsEqual(self.data, list(data))
+
+ def test_address_group_list_name(self):
+ arglist = [
+ '--name', self.address_groups[0].name,
+ ]
+ verifylist = [
+ ('name', self.address_groups[0].name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.network.address_groups.assert_called_once_with(
+ **{'name': self.address_groups[0].name})
+ self.assertEqual(self.columns, columns)
+ self.assertItemsEqual(self.data, list(data))
+
+ def test_address_group_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.address_groups.assert_called_once_with(
+ **{'tenant_id': project.id, 'project_id': project.id})
+ self.assertEqual(self.columns, columns)
+ self.assertItemsEqual(self.data, list(data))
+
+ def test_address_group_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 = {'tenant_id': project.id, 'project_id': project.id}
+
+ self.network.address_groups.assert_called_once_with(**filters)
+ self.assertEqual(self.columns, columns)
+ self.assertItemsEqual(self.data, list(data))
+
+
+class TestSetAddressGroup(TestAddressGroup):
+
+ # The address group to set.
+ _address_group = network_fakes.FakeAddressGroup.create_one_address_group()
+
+ def setUp(self):
+ super(TestSetAddressGroup, self).setUp()
+ self.network.update_address_group = mock.Mock(return_value=None)
+ self.network.find_address_group = mock.Mock(
+ return_value=self._address_group)
+ self.network.add_addresses_to_address_group = mock.Mock(
+ return_value=self._address_group)
+ # Get the command object to test
+ self.cmd = address_group.SetAddressGroup(self.app, self.namespace)
+
+ def test_set_nothing(self):
+ arglist = [self._address_group.name, ]
+ verifylist = [
+ ('address_group', self._address_group.name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ self.network.update_address_group.assert_not_called()
+ self.network.add_addresses_to_address_group.assert_not_called()
+ self.assertIsNone(result)
+
+ def test_set_name_and_description(self):
+ arglist = [
+ '--name', 'new_address_group_name',
+ '--description', 'new_address_group_description',
+ self._address_group.name,
+ ]
+ verifylist = [
+ ('name', 'new_address_group_name'),
+ ('description', 'new_address_group_description'),
+ ('address_group', self._address_group.name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ attrs = {
+ 'name': "new_address_group_name",
+ 'description': 'new_address_group_description',
+ }
+ self.network.update_address_group.assert_called_with(
+ self._address_group, **attrs)
+ self.assertIsNone(result)
+
+ def test_set_one_address(self):
+ arglist = [
+ self._address_group.name,
+ '--address', '10.0.0.2',
+ ]
+ verifylist = [
+ ('address_group', self._address_group.name),
+ ('address', ['10.0.0.2']),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.network.add_addresses_to_address_group.assert_called_once_with(
+ self._address_group, ['10.0.0.2/32'])
+ self.assertIsNone(result)
+
+ def test_set_multiple_addresses(self):
+ arglist = [
+ self._address_group.name,
+ '--address', '10.0.0.2',
+ '--address', '2001::/16',
+ ]
+ verifylist = [
+ ('address_group', self._address_group.name),
+ ('address', ['10.0.0.2', '2001::/16']),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.network.add_addresses_to_address_group.assert_called_once_with(
+ self._address_group, ['10.0.0.2/32', '2001::/16'])
+ self.assertIsNone(result)
+
+
+class TestShowAddressGroup(TestAddressGroup):
+
+ # The address group to show.
+ _address_group = network_fakes.FakeAddressGroup.create_one_address_group()
+ columns = (
+ 'addresses',
+ 'description',
+ 'id',
+ 'name',
+ 'project_id',
+ )
+ data = (
+ _address_group.addresses,
+ _address_group.description,
+ _address_group.id,
+ _address_group.name,
+ _address_group.project_id,
+ )
+
+ def setUp(self):
+ super(TestShowAddressGroup, self).setUp()
+ self.network.find_address_group = mock.Mock(
+ return_value=self._address_group)
+
+ # Get the command object to test
+ self.cmd = address_group.ShowAddressGroup(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_all_options(self):
+ arglist = [
+ self._address_group.name,
+ ]
+ verifylist = [
+ ('address_group', self._address_group.name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.network.find_address_group.assert_called_once_with(
+ self._address_group.name, ignore_missing=False)
+ self.assertEqual(self.columns, columns)
+ self.assertItemsEqual(self.data, list(data))
+
+
+class TestUnsetAddressGroup(TestAddressGroup):
+
+ # The address group to unset.
+ _address_group = network_fakes.FakeAddressGroup.create_one_address_group()
+
+ def setUp(self):
+ super(TestUnsetAddressGroup, self).setUp()
+ self.network.find_address_group = mock.Mock(
+ return_value=self._address_group)
+ self.network.remove_addresses_from_address_group = mock.Mock(
+ return_value=self._address_group)
+ # Get the command object to test
+ self.cmd = address_group.UnsetAddressGroup(self.app, self.namespace)
+
+ def test_unset_nothing(self):
+ arglist = [self._address_group.name, ]
+ verifylist = [
+ ('address_group', self._address_group.name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ self.network.remove_addresses_from_address_group.assert_not_called()
+ self.assertIsNone(result)
+
+ def test_unset_one_address(self):
+ arglist = [
+ self._address_group.name,
+ '--address', '10.0.0.2',
+ ]
+ verifylist = [
+ ('address_group', self._address_group.name),
+ ('address', ['10.0.0.2']),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.network.remove_addresses_from_address_group.\
+ assert_called_once_with(self._address_group, ['10.0.0.2/32'])
+ self.assertIsNone(result)
+
+ def test_unset_multiple_addresses(self):
+ arglist = [
+ self._address_group.name,
+ '--address', '10.0.0.2',
+ '--address', '2001::/16',
+ ]
+ verifylist = [
+ ('address_group', self._address_group.name),
+ ('address', ['10.0.0.2', '2001::/16']),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.network.remove_addresses_from_address_group.\
+ assert_called_once_with(self._address_group,
+ ['10.0.0.2/32', '2001::/16'])
+ self.assertIsNone(result)