diff options
Diffstat (limited to 'openstackclient')
| -rw-r--r-- | openstackclient/compute/v2/server.py | 161 | ||||
| -rw-r--r-- | openstackclient/network/v2/address_group.py | 292 | ||||
| -rw-r--r-- | openstackclient/tests/functional/common/test_help.py | 2 | ||||
| -rw-r--r-- | openstackclient/tests/functional/network/v2/test_address_group.py | 177 | ||||
| -rw-r--r-- | openstackclient/tests/unit/compute/v2/test_server.py | 202 | ||||
| -rw-r--r-- | openstackclient/tests/unit/network/v2/fakes.py | 73 | ||||
| -rw-r--r-- | openstackclient/tests/unit/network/v2/test_address_group.py | 502 |
7 files changed, 1352 insertions, 57 deletions
diff --git a/openstackclient/compute/v2/server.py b/openstackclient/compute/v2/server.py index aa4f2372..bebef897 100644 --- a/openstackclient/compute/v2/server.py +++ b/openstackclient/compute/v2/server.py @@ -3595,25 +3595,115 @@ class SetServer(command.Command): class ShelveServer(command.Command): - _description = _("Shelve server(s)") + """Shelve and optionally offload server(s). + + Shelving a server creates a snapshot of the server and stores this + snapshot before shutting down the server. This shelved server can then be + offloaded or deleted from the host, freeing up remaining resources on the + host, such as network interfaces. Shelved servers can be unshelved, + restoring the server from the snapshot. Shelving is therefore useful where + users wish to retain the UUID and IP of a server, without utilizing other + resources or disks. + + Most clouds are configured to automatically offload shelved servers + immediately or after a small delay. For clouds where this is not + configured, or where the delay is larger, offloading can be manually + specified. This is an admin-only operation by default. + """ def get_parser(self, prog_name): parser = super(ShelveServer, self).get_parser(prog_name) parser.add_argument( - 'server', + 'servers', metavar='<server>', nargs='+', help=_('Server(s) to shelve (name or ID)'), ) + parser.add_argument( + '--offload', + action='store_true', + default=False, + help=_( + 'Remove the shelved server(s) from the host (admin only). ' + 'Invoking this option on an unshelved server(s) will result ' + 'in the server being shelved first' + ), + ) + parser.add_argument( + '--wait', + action='store_true', + default=False, + help=_('Wait for shelve and/or offload operation to complete'), + ) return parser def take_action(self, parsed_args): + + def _show_progress(progress): + if progress: + self.app.stdout.write('\rProgress: %s' % progress) + self.app.stdout.flush() + compute_client = self.app.client_manager.compute - for server in parsed_args.server: - utils.find_resource( + + for server in parsed_args.servers: + server_obj = utils.find_resource( compute_client.servers, server, - ).shelve() + ) + if server_obj.status.lower() in ('shelved', 'shelved_offloaded'): + continue + + server_obj.shelve() + + # if we don't hav to wait, either because it was requested explicitly + # or is required implicitly, then our job is done + if not parsed_args.wait and not parsed_args.offload: + return + + for server in parsed_args.servers: + # TODO(stephenfin): We should wait for these in parallel using e.g. + # https://review.opendev.org/c/openstack/osc-lib/+/762503/ + if not utils.wait_for_status( + compute_client.servers.get, server_obj.id, + success_status=('shelved', 'shelved_offloaded'), + callback=_show_progress, + ): + LOG.error(_('Error shelving server: %s'), server_obj.id) + self.app.stdout.write( + _('Error shelving server: %s\n') % server_obj.id) + raise SystemExit + + if not parsed_args.offload: + return + + for server in parsed_args.servers: + server_obj = utils.find_resource( + compute_client.servers, + server, + ) + if server_obj.status.lower() == 'shelved_offloaded': + continue + + server_obj.shelve_offload() + + if not parsed_args.wait: + return + + for server in parsed_args.servers: + # TODO(stephenfin): We should wait for these in parallel using e.g. + # https://review.opendev.org/c/openstack/osc-lib/+/762503/ + if not utils.wait_for_status( + compute_client.servers.get, server_obj.id, + success_status=('shelved_offloaded',), + callback=_show_progress, + ): + LOG.error( + _('Error offloading shelved server %s'), server_obj.id) + self.app.stdout.write( + _('Error offloading shelved server: %s\n') % ( + server_obj.id)) + raise SystemExit class ShowServer(command.ShowOne): @@ -4051,25 +4141,54 @@ class UnshelveServer(command.Command): 'SHELVED_OFFLOADED server (supported by ' '--os-compute-api-version 2.77 or above)'), ) + parser.add_argument( + '--wait', + action='store_true', + default=False, + help=_('Wait for unshelve operation to complete'), + ) return parser def take_action(self, parsed_args): + + def _show_progress(progress): + if progress: + self.app.stdout.write('\rProgress: %s' % progress) + self.app.stdout.flush() + compute_client = self.app.client_manager.compute - support_az = compute_client.api_version >= api_versions.APIVersion( - '2.77') - if not support_az and parsed_args.availability_zone: - msg = _("--os-compute-api-version 2.77 or greater is required " - "to support the '--availability-zone' option.") - raise exceptions.CommandError(msg) + kwargs = {} + + if parsed_args.availability_zone: + if compute_client.api_version < api_versions.APIVersion('2.77'): + msg = _( + '--os-compute-api-version 2.77 or greater is required ' + 'to support the --availability-zone option' + ) + raise exceptions.CommandError(msg) + + kwargs['availability_zone'] = parsed_args.availability_zone for server in parsed_args.server: - if support_az: - utils.find_resource( - compute_client.servers, - server - ).unshelve(availability_zone=parsed_args.availability_zone) - else: - utils.find_resource( - compute_client.servers, - server, - ).unshelve() + server_obj = utils.find_resource( + compute_client.servers, + server, + ) + + if server_obj.status.lower() not in ( + 'shelved', 'shelved_offloaded', + ): + continue + + server_obj.unshelve(**kwargs) + + if parsed_args.wait: + if not utils.wait_for_status( + compute_client.servers.get, server_obj.id, + success_status=('active', 'shutoff'), + callback=_show_progress, + ): + LOG.error(_('Error unshelving server %s'), server_obj.id) + self.app.stdout.write( + _('Error unshelving server: %s\n') % server_obj.id) + raise SystemExit diff --git a/openstackclient/network/v2/address_group.py b/openstackclient/network/v2/address_group.py new file mode 100644 index 00000000..c5b2f126 --- /dev/null +++ b/openstackclient/network/v2/address_group.py @@ -0,0 +1,292 @@ +# 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): + return [str(netaddr.IPNetwork(addr)) for addr in addresses] + + +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['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/common/test_help.py b/openstackclient/tests/functional/common/test_help.py index 3a9aef9e..c55741f1 100644 --- a/openstackclient/tests/functional/common/test_help.py +++ b/openstackclient/tests/functional/common/test_help.py @@ -43,7 +43,7 @@ class HelpTests(base.TestCase): ('server resize', 'Scale server to a new flavor'), ('server resume', 'Resume server(s)'), ('server set', 'Set server properties'), - ('server shelve', 'Shelve server(s)'), + ('server shelve', 'Shelve and optionally offload server(s)'), ('server show', 'Show server details'), ('server ssh', 'SSH to server'), ('server start', 'Start server(s).'), 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/compute/v2/test_server.py b/openstackclient/tests/unit/compute/v2/test_server.py index bde0699b..6cec6ad5 100644 --- a/openstackclient/tests/unit/compute/v2/test_server.py +++ b/openstackclient/tests/unit/compute/v2/test_server.py @@ -6455,16 +6455,126 @@ class TestServerShelve(TestServer): # Get the command object to test self.cmd = server.ShelveServer(self.app, None) - # Set shelve method to be tested. - self.methods = { + def test_shelve(self): + server_info = {'status': 'ACTIVE'} + server_methods = { 'shelve': None, + 'shelve_offload': None, } - def test_shelve_one_server(self): - self.run_method_with_servers('shelve', 1) + server = compute_fakes.FakeServer.create_one_server( + attrs=server_info, methods=server_methods) + self.servers_mock.get.return_value = server + + arglist = [server.name] + verifylist = [ + ('servers', [server.name]), + ('wait', False), + ('offload', False), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.assertIsNone(result) - def test_shelve_multi_servers(self): - self.run_method_with_servers('shelve', 3) + self.servers_mock.get.assert_called_once_with(server.name) + server.shelve.assert_called_once_with() + server.shelve_offload.assert_not_called() + + def test_shelve_already_shelved(self): + server_info = {'status': 'SHELVED'} + server_methods = { + 'shelve': None, + 'shelve_offload': None, + } + + server = compute_fakes.FakeServer.create_one_server( + attrs=server_info, methods=server_methods) + self.servers_mock.get.return_value = server + + arglist = [server.name] + verifylist = [ + ('servers', [server.name]), + ('wait', False), + ('offload', False), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.assertIsNone(result) + + self.servers_mock.get.assert_called_once_with(server.name) + server.shelve.assert_not_called() + server.shelve_offload.assert_not_called() + + @mock.patch.object(common_utils, 'wait_for_status', return_value=True) + def test_shelve_with_wait(self, mock_wait_for_status): + server_info = {'status': 'ACTIVE'} + server_methods = { + 'shelve': None, + 'shelve_offload': None, + } + + server = compute_fakes.FakeServer.create_one_server( + attrs=server_info, methods=server_methods) + self.servers_mock.get.return_value = server + + arglist = ['--wait', server.name] + verifylist = [ + ('servers', [server.name]), + ('wait', True), + ('offload', False), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.assertIsNone(result) + + self.servers_mock.get.assert_called_once_with(server.name) + server.shelve.assert_called_once_with() + server.shelve_offload.assert_not_called() + mock_wait_for_status.assert_called_once_with( + self.servers_mock.get, + server.id, + callback=mock.ANY, + success_status=('shelved', 'shelved_offloaded'), + ) + + @mock.patch.object(common_utils, 'wait_for_status', return_value=True) + def test_shelve_offload(self, mock_wait_for_status): + server_info = {'status': 'ACTIVE'} + server_methods = { + 'shelve': None, + 'shelve_offload': None, + } + + server = compute_fakes.FakeServer.create_one_server( + attrs=server_info, methods=server_methods) + self.servers_mock.get.return_value = server + + arglist = ['--offload', server.name] + verifylist = [ + ('servers', [server.name]), + ('wait', False), + ('offload', True), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.assertIsNone(result) + + self.servers_mock.get.assert_has_calls([ + mock.call(server.name), + mock.call(server.name), + ]) + server.shelve.assert_called_once_with() + server.shelve_offload.assert_called_once_with() + mock_wait_for_status.assert_called_once_with( + self.servers_mock.get, + server.id, + callback=mock.ANY, + success_status=('shelved', 'shelved_offloaded'), + ) class TestServerShow(TestServer): @@ -6880,6 +6990,9 @@ class TestServerUnshelve(TestServer): self.methods = { 'unshelve': None, } + self.attrs = { + 'status': 'SHELVED', + } def test_unshelve_one_server(self): self.run_method_with_servers('unshelve', 1) @@ -6887,55 +7000,74 @@ class TestServerUnshelve(TestServer): def test_unshelve_multi_servers(self): self.run_method_with_servers('unshelve', 3) - def test_unshelve_server_with_specified_az(self): - server = compute_fakes.FakeServer.create_one_server() + def test_unshelve_with_specified_az(self): + self.app.client_manager.compute.api_version = \ + api_versions.APIVersion('2.77') + + server = compute_fakes.FakeServer.create_one_server( + attrs=self.attrs, methods=self.methods) + self.servers_mock.get.return_value = server arglist = [ - server.id, '--availability-zone', "foo-az", + server.id, ] verifylist = [ ('availability_zone', "foo-az"), ('server', [server.id]) ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) - ex = self.assertRaises(exceptions.CommandError, - self.cmd.take_action, - parsed_args) - self.assertIn( - '--os-compute-api-version 2.77 or greater is required', str(ex)) - -class TestServerUnshelveV277(TestServerUnshelve): - - def setUp(self): - super(TestServerUnshelveV277, self).setUp() - - self.server = compute_fakes.FakeServer.create_one_server( - methods=self.methods) - - # This is the return value for utils.find_resource() - self.servers_mock.get.return_value = self.server + self.cmd.take_action(parsed_args) - # Get the command object to test - self.cmd = server.UnshelveServer(self.app, None) + self.servers_mock.get.assert_called_with(server.id) + server.unshelve.assert_called_with(availability_zone="foo-az") - def test_specified_az_to_unshelve_with_v277(self): - self.app.client_manager.compute.api_version = api_versions.APIVersion( - '2.77') + def test_unshelve_with_specified_az_pre_v277(self): + self.app.client_manager.compute.api_version = \ + api_versions.APIVersion('2.76') + server = compute_fakes.FakeServer.create_one_server( + attrs=self.attrs, methods=self.methods) arglist = [ + server.id, '--availability-zone', "foo-az", - self.server.id, ] verifylist = [ ('availability_zone', "foo-az"), - ('server', [self.server.id]) + ('server', [server.id]) ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) + ex = self.assertRaises( + exceptions.CommandError, + self.cmd.take_action, + parsed_args) + self.assertIn( + '--os-compute-api-version 2.77 or greater is required', str(ex)) - self.cmd.take_action(parsed_args) - self.servers_mock.get.assert_called_with(self.server.id) - self.server.unshelve.assert_called_with(availability_zone="foo-az") + @mock.patch.object(common_utils, 'wait_for_status', return_value=True) + def test_unshelve_with_wait(self, mock_wait_for_status): + server = compute_fakes.FakeServer.create_one_server( + attrs=self.attrs, methods=self.methods) + self.servers_mock.get.return_value = server + + arglist = ['--wait', server.name] + verifylist = [ + ('server', [server.name]), + ('wait', True), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.assertIsNone(result) + + self.servers_mock.get.assert_called_once_with(server.name) + server.unshelve.assert_called_once_with() + mock_wait_for_status.assert_called_once_with( + self.servers_mock.get, + server.id, + callback=mock.ANY, + success_status=('active', 'shutoff'), + ) class TestServerGeneral(TestServer): 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..e4fa8ab3 --- /dev/null +++ b/openstackclient/tests/unit/network/v2/test_address_group.py @@ -0,0 +1,502 @@ +# 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( + 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) + + self.network.address_groups.assert_called_once_with( + project_id=project.id) + 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) |