diff options
Diffstat (limited to 'openstackclient')
| -rw-r--r-- | openstackclient/compute/v2/server_volume.py | 73 | ||||
| -rw-r--r-- | openstackclient/tests/unit/compute/v2/fakes.py | 60 | ||||
| -rw-r--r-- | openstackclient/tests/unit/compute/v2/test_server_volume.py | 167 |
3 files changed, 300 insertions, 0 deletions
diff --git a/openstackclient/compute/v2/server_volume.py b/openstackclient/compute/v2/server_volume.py new file mode 100644 index 00000000..8a931ae5 --- /dev/null +++ b/openstackclient/compute/v2/server_volume.py @@ -0,0 +1,73 @@ +# Copyright 2020, Red Hat Inc. +# +# 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. + +"""Compute v2 Server action implementations""" + +from novaclient import api_versions +from osc_lib.command import command +from osc_lib import utils + +from openstackclient.i18n import _ + + +class ListServerVolume(command.Lister): + """List all the volumes attached to a server.""" + + def get_parser(self, prog_name): + parser = super().get_parser(prog_name) + parser.add_argument( + 'server', + help=_('Server to list volume attachments for (name or ID)'), + ) + return parser + + def take_action(self, parsed_args): + + compute_client = self.app.client_manager.compute + + server = utils.find_resource( + compute_client.servers, + parsed_args.server, + ) + + volumes = compute_client.volumes.get_server_volumes(server.id) + + columns = ( + 'id', + 'device', + 'serverId', + 'volumeId', + ) + column_headers = ( + 'ID', + 'Device', + 'Server ID', + 'Volume ID', + ) + if compute_client.api_version >= api_versions.APIVersion('2.70'): + columns += ('tag',) + column_headers += ('Tag',) + + if compute_client.api_version >= api_versions.APIVersion('2.79'): + columns += ('delete_on_termination',) + column_headers += ('Delete On Termination?',) + + return ( + column_headers, + ( + utils.get_item_properties( + s, columns, mixed_case_fields=('serverId', 'volumeId') + ) for s in volumes + ), + ) diff --git a/openstackclient/tests/unit/compute/v2/fakes.py b/openstackclient/tests/unit/compute/v2/fakes.py index b667c691..e4cf1045 100644 --- a/openstackclient/tests/unit/compute/v2/fakes.py +++ b/openstackclient/tests/unit/compute/v2/fakes.py @@ -1631,3 +1631,63 @@ class FakeServerMigration(object): attrs, methods)) return migrations + + +class FakeVolumeAttachment(object): + """Fake one or more volume attachments (BDMs).""" + + @staticmethod + def create_one_volume_attachment(attrs=None, methods=None): + """Create a fake volume attachment. + + :param Dictionary attrs: + A dictionary with all attributes + :param Dictionary methods: + A dictionary with all methods + :return: + A FakeResource object, with id, device, and so on + """ + attrs = attrs or {} + methods = methods or {} + + # Set default attributes. + volume_attachment_info = { + "id": uuid.uuid4().hex, + "device": "/dev/sdb", + "serverId": uuid.uuid4().hex, + "volumeId": uuid.uuid4().hex, + # introduced in API microversion 2.70 + "tag": "foo", + # introduced in API microversion 2.79 + "delete_on_termination": True, + } + + # Overwrite default attributes. + volume_attachment_info.update(attrs) + + volume_attachment = fakes.FakeResource( + info=copy.deepcopy(volume_attachment_info), + methods=methods, + loaded=True) + return volume_attachment + + @staticmethod + def create_volume_attachments(attrs=None, methods=None, count=2): + """Create multiple fake volume attachments (BDMs). + + :param Dictionary attrs: + A dictionary with all attributes + :param Dictionary methods: + A dictionary with all methods + :param int count: + The number of server migrations to fake + :return: + A list of FakeResource objects faking the volume attachments. + """ + volume_attachments = [] + for i in range(0, count): + volume_attachments.append( + FakeVolumeAttachment.create_one_volume_attachment( + attrs, methods)) + + return volume_attachments diff --git a/openstackclient/tests/unit/compute/v2/test_server_volume.py b/openstackclient/tests/unit/compute/v2/test_server_volume.py new file mode 100644 index 00000000..d09c2874 --- /dev/null +++ b/openstackclient/tests/unit/compute/v2/test_server_volume.py @@ -0,0 +1,167 @@ +# 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 novaclient import api_versions + +from openstackclient.compute.v2 import server_volume +from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes + + +class TestServerVolume(compute_fakes.TestComputev2): + + def setUp(self): + super().setUp() + + # Get a shortcut to the compute client ServerManager Mock + self.servers_mock = self.app.client_manager.compute.servers + self.servers_mock.reset_mock() + + # Get a shortcut to the compute client VolumeManager mock + self.servers_volumes_mock = self.app.client_manager.compute.volumes + self.servers_volumes_mock.reset_mock() + + +class TestServerVolumeList(TestServerVolume): + + def setUp(self): + super().setUp() + + self.server = compute_fakes.FakeServer.create_one_server() + self.volume_attachments = ( + compute_fakes.FakeVolumeAttachment.create_volume_attachments()) + + self.servers_mock.get.return_value = self.server + self.servers_volumes_mock.get_server_volumes.return_value = ( + self.volume_attachments) + + # Get the command object to test + self.cmd = server_volume.ListServerVolume(self.app, None) + + def test_server_volume_list(self): + self.app.client_manager.compute.api_version = \ + api_versions.APIVersion('2.1') + + arglist = [ + self.server.id, + ] + verifylist = [ + ('server', self.server.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.assertEqual(('ID', 'Device', 'Server ID', 'Volume ID'), columns) + self.assertEqual( + ( + ( + self.volume_attachments[0].id, + self.volume_attachments[0].device, + self.volume_attachments[0].serverId, + self.volume_attachments[0].volumeId, + ), + ( + self.volume_attachments[1].id, + self.volume_attachments[1].device, + self.volume_attachments[1].serverId, + self.volume_attachments[1].volumeId, + ), + ), + tuple(data), + ) + self.servers_volumes_mock.get_server_volumes.assert_called_once_with( + self.server.id) + + def test_server_volume_list_with_tags(self): + self.app.client_manager.compute.api_version = \ + api_versions.APIVersion('2.70') + + arglist = [ + self.server.id, + ] + verifylist = [ + ('server', self.server.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.assertEqual( + ('ID', 'Device', 'Server ID', 'Volume ID', 'Tag',), columns, + ) + self.assertEqual( + ( + ( + self.volume_attachments[0].id, + self.volume_attachments[0].device, + self.volume_attachments[0].serverId, + self.volume_attachments[0].volumeId, + self.volume_attachments[0].tag, + ), + ( + self.volume_attachments[1].id, + self.volume_attachments[1].device, + self.volume_attachments[1].serverId, + self.volume_attachments[1].volumeId, + self.volume_attachments[1].tag, + ), + ), + tuple(data), + ) + self.servers_volumes_mock.get_server_volumes.assert_called_once_with( + self.server.id) + + def test_server_volume_list_with_delete_on_attachment(self): + self.app.client_manager.compute.api_version = \ + api_versions.APIVersion('2.79') + + arglist = [ + self.server.id, + ] + verifylist = [ + ('server', self.server.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.assertEqual( + ( + 'ID', 'Device', 'Server ID', 'Volume ID', 'Tag', + 'Delete On Termination?', + ), + columns, + ) + self.assertEqual( + ( + ( + self.volume_attachments[0].id, + self.volume_attachments[0].device, + self.volume_attachments[0].serverId, + self.volume_attachments[0].volumeId, + self.volume_attachments[0].tag, + self.volume_attachments[0].delete_on_termination, + ), + ( + self.volume_attachments[1].id, + self.volume_attachments[1].device, + self.volume_attachments[1].serverId, + self.volume_attachments[1].volumeId, + self.volume_attachments[1].tag, + self.volume_attachments[1].delete_on_termination, + ), + ), + tuple(data), + ) + self.servers_volumes_mock.get_server_volumes.assert_called_once_with( + self.server.id) |