# 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 cinderclient import api_versions from osc_lib.cli import format_columns from osc_lib import exceptions from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes from openstackclient.tests.unit.volume.v3 import fakes as volume_fakes from openstackclient.volume.v3 import volume_attachment class TestVolumeAttachment(volume_fakes.TestVolume): def setUp(self): super().setUp() self.volumes_mock = self.app.client_manager.volume.volumes self.volumes_mock.reset_mock() self.volume_attachments_mock = \ self.app.client_manager.volume.attachments self.volume_attachments_mock.reset_mock() self.projects_mock = self.app.client_manager.identity.projects self.projects_mock.reset_mock() self.servers_mock = self.app.client_manager.compute.servers self.servers_mock.reset_mock() class TestVolumeAttachmentCreate(TestVolumeAttachment): volume = volume_fakes.FakeVolume.create_one_volume() server = compute_fakes.FakeServer.create_one_server() volume_attachment = \ volume_fakes.FakeVolumeAttachment.create_one_volume_attachment( attrs={'instance': server.id, 'volume_id': volume.id}) columns = ( 'ID', 'Volume ID', 'Instance ID', 'Status', 'Attach Mode', 'Attached At', 'Detached At', 'Properties', ) data = ( volume_attachment.id, volume_attachment.volume_id, volume_attachment.instance, volume_attachment.status, volume_attachment.attach_mode, volume_attachment.attached_at, volume_attachment.detached_at, format_columns.DictColumn(volume_attachment.connection_info), ) def setUp(self): super().setUp() self.volumes_mock.get.return_value = self.volume self.servers_mock.get.return_value = self.server # VolumeAttachmentManager.create returns a dict self.volume_attachments_mock.create.return_value = \ self.volume_attachment.to_dict() self.cmd = volume_attachment.CreateVolumeAttachment(self.app, None) def test_volume_attachment_create(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.27') arglist = [ self.volume.id, self.server.id, ] verifylist = [ ('volume', self.volume.id), ('server', self.server.id), ('connect', False), ('initiator', None), ('ip', None), ('host', None), ('platform', None), ('os_type', None), ('multipath', False), ('mountpoint', None), ('mode', None), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = self.cmd.take_action(parsed_args) self.volumes_mock.get.assert_called_once_with(self.volume.id) self.servers_mock.get.assert_called_once_with(self.server.id) self.volume_attachments_mock.create.assert_called_once_with( self.volume.id, {}, self.server.id, None, ) self.assertEqual(self.columns, columns) self.assertCountEqual(self.data, data) def test_volume_attachment_create_with_connect(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.54') arglist = [ self.volume.id, self.server.id, '--connect', '--initiator', 'iqn.1993-08.org.debian:01:cad181614cec', '--ip', '192.168.1.20', '--host', 'my-host', '--platform', 'x86_64', '--os-type', 'linux2', '--multipath', '--mountpoint', '/dev/vdb', '--mode', 'null', ] verifylist = [ ('volume', self.volume.id), ('server', self.server.id), ('connect', True), ('initiator', 'iqn.1993-08.org.debian:01:cad181614cec'), ('ip', '192.168.1.20'), ('host', 'my-host'), ('platform', 'x86_64'), ('os_type', 'linux2'), ('multipath', True), ('mountpoint', '/dev/vdb'), ('mode', 'null'), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = self.cmd.take_action(parsed_args) connect_info = dict([ ('initiator', 'iqn.1993-08.org.debian:01:cad181614cec'), ('ip', '192.168.1.20'), ('host', 'my-host'), ('platform', 'x86_64'), ('os_type', 'linux2'), ('multipath', True), ('mountpoint', '/dev/vdb'), ]) self.volumes_mock.get.assert_called_once_with(self.volume.id) self.servers_mock.get.assert_called_once_with(self.server.id) self.volume_attachments_mock.create.assert_called_once_with( self.volume.id, connect_info, self.server.id, 'null', ) self.assertEqual(self.columns, columns) self.assertCountEqual(self.data, data) def test_volume_attachment_create_pre_v327(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.26') arglist = [ self.volume.id, self.server.id, ] verifylist = [ ('volume', self.volume.id), ('server', self.server.id), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) exc = self.assertRaises( exceptions.CommandError, self.cmd.take_action, parsed_args) self.assertIn( '--os-volume-api-version 3.27 or greater is required', str(exc)) def test_volume_attachment_create_with_mode_pre_v354(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.53') arglist = [ self.volume.id, self.server.id, '--mode', 'rw', ] verifylist = [ ('volume', self.volume.id), ('server', self.server.id), ('mode', 'rw'), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) exc = self.assertRaises( exceptions.CommandError, self.cmd.take_action, parsed_args) self.assertIn( '--os-volume-api-version 3.54 or greater is required', str(exc)) def test_volume_attachment_create_with_connect_missing_arg(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.54') arglist = [ self.volume.id, self.server.id, '--initiator', 'iqn.1993-08.org.debian:01:cad181614cec', ] verifylist = [ ('volume', self.volume.id), ('server', self.server.id), ('connect', False), ('initiator', 'iqn.1993-08.org.debian:01:cad181614cec'), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) exc = self.assertRaises( exceptions.CommandError, self.cmd.take_action, parsed_args) self.assertIn( 'You must specify the --connect option for any', str(exc)) class TestVolumeAttachmentDelete(TestVolumeAttachment): volume_attachment = \ volume_fakes.FakeVolumeAttachment.create_one_volume_attachment() def setUp(self): super().setUp() self.volume_attachments_mock.delete.return_value = None self.cmd = volume_attachment.DeleteVolumeAttachment(self.app, None) def test_volume_attachment_delete(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.27') arglist = [ self.volume_attachment.id, ] verifylist = [ ('attachment', self.volume_attachment.id), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) self.volume_attachments_mock.delete.assert_called_once_with( self.volume_attachment.id, ) self.assertIsNone(result) def test_volume_attachment_delete_pre_v327(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.26') arglist = [ self.volume_attachment.id, ] verifylist = [ ('attachment', self.volume_attachment.id), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) exc = self.assertRaises( exceptions.CommandError, self.cmd.take_action, parsed_args) self.assertIn( '--os-volume-api-version 3.27 or greater is required', str(exc)) class TestVolumeAttachmentSet(TestVolumeAttachment): volume_attachment = \ volume_fakes.FakeVolumeAttachment.create_one_volume_attachment() columns = ( 'ID', 'Volume ID', 'Instance ID', 'Status', 'Attach Mode', 'Attached At', 'Detached At', 'Properties', ) data = ( volume_attachment.id, volume_attachment.volume_id, volume_attachment.instance, volume_attachment.status, volume_attachment.attach_mode, volume_attachment.attached_at, volume_attachment.detached_at, format_columns.DictColumn(volume_attachment.connection_info), ) def setUp(self): super().setUp() self.volume_attachments_mock.update.return_value = \ self.volume_attachment self.cmd = volume_attachment.SetVolumeAttachment(self.app, None) def test_volume_attachment_set(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.27') arglist = [ self.volume_attachment.id, '--initiator', 'iqn.1993-08.org.debian:01:cad181614cec', '--ip', '192.168.1.20', '--host', 'my-host', '--platform', 'x86_64', '--os-type', 'linux2', '--multipath', '--mountpoint', '/dev/vdb', ] verifylist = [ ('attachment', self.volume_attachment.id), ('initiator', 'iqn.1993-08.org.debian:01:cad181614cec'), ('ip', '192.168.1.20'), ('host', 'my-host'), ('platform', 'x86_64'), ('os_type', 'linux2'), ('multipath', True), ('mountpoint', '/dev/vdb'), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = self.cmd.take_action(parsed_args) connect_info = dict([ ('initiator', 'iqn.1993-08.org.debian:01:cad181614cec'), ('ip', '192.168.1.20'), ('host', 'my-host'), ('platform', 'x86_64'), ('os_type', 'linux2'), ('multipath', True), ('mountpoint', '/dev/vdb'), ]) self.volume_attachments_mock.update.assert_called_once_with( self.volume_attachment.id, connect_info, ) self.assertEqual(self.columns, columns) self.assertCountEqual(self.data, data) def test_volume_attachment_set_pre_v327(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.26') arglist = [ self.volume_attachment.id, '--initiator', 'iqn.1993-08.org.debian:01:cad181614cec', ] verifylist = [ ('attachment', self.volume_attachment.id), ('initiator', 'iqn.1993-08.org.debian:01:cad181614cec'), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) exc = self.assertRaises( exceptions.CommandError, self.cmd.take_action, parsed_args) self.assertIn( '--os-volume-api-version 3.27 or greater is required', str(exc)) class TestVolumeAttachmentComplete(TestVolumeAttachment): volume_attachment = \ volume_fakes.FakeVolumeAttachment.create_one_volume_attachment() def setUp(self): super().setUp() self.volume_attachments_mock.complete.return_value = None self.cmd = volume_attachment.CompleteVolumeAttachment(self.app, None) def test_volume_attachment_complete(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.44') arglist = [ self.volume_attachment.id, ] verifylist = [ ('attachment', self.volume_attachment.id), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) self.volume_attachments_mock.complete.assert_called_once_with( self.volume_attachment.id, ) self.assertIsNone(result) def test_volume_attachment_complete_pre_v344(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.43') arglist = [ self.volume_attachment.id, ] verifylist = [ ('attachment', self.volume_attachment.id), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) exc = self.assertRaises( exceptions.CommandError, self.cmd.take_action, parsed_args) self.assertIn( '--os-volume-api-version 3.44 or greater is required', str(exc)) class TestVolumeAttachmentList(TestVolumeAttachment): project = identity_fakes.FakeProject.create_one_project() volume_attachments = \ volume_fakes.FakeVolumeAttachment.create_volume_attachments() columns = ( 'ID', 'Volume ID', 'Server ID', 'Status', ) data = [ ( volume_attachment.id, volume_attachment.volume_id, volume_attachment.instance, volume_attachment.status, ) for volume_attachment in volume_attachments ] def setUp(self): super().setUp() self.projects_mock.get.return_value = self.project self.volume_attachments_mock.list.return_value = \ self.volume_attachments self.cmd = volume_attachment.ListVolumeAttachment(self.app, None) def test_volume_attachment_list(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.27') arglist = [] verifylist = [ ('project', None), ('all_projects', False), ('volume_id', None), ('status', None), ('marker', None), ('limit', None), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = self.cmd.take_action(parsed_args) self.volume_attachments_mock.list.assert_called_once_with( search_opts={ 'all_tenants': False, 'project_id': None, 'status': None, 'volume_id': None, }, marker=None, limit=None, ) self.assertEqual(self.columns, columns) self.assertCountEqual(tuple(self.data), data) def test_volume_attachment_list_with_options(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.27') arglist = [ '--project', self.project.name, '--volume-id', 'volume-id', '--status', 'attached', '--marker', 'volume-attachment-id', '--limit', '2', ] verifylist = [ ('project', self.project.name), ('all_projects', False), ('volume_id', 'volume-id'), ('status', 'attached'), ('marker', 'volume-attachment-id'), ('limit', 2), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = self.cmd.take_action(parsed_args) self.volume_attachments_mock.list.assert_called_once_with( search_opts={ 'all_tenants': True, 'project_id': self.project.id, 'status': 'attached', 'volume_id': 'volume-id', }, marker='volume-attachment-id', limit=2, ) self.assertEqual(self.columns, columns) self.assertCountEqual(tuple(self.data), data) def test_volume_attachment_list_pre_v327(self): self.app.client_manager.volume.api_version = \ api_versions.APIVersion('3.26') arglist = [] verifylist = [ ('project', None), ('all_projects', False), ('volume_id', None), ('status', None), ('marker', None), ('limit', None), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) exc = self.assertRaises( exceptions.CommandError, self.cmd.take_action, parsed_args) self.assertIn( '--os-volume-api-version 3.27 or greater is required', str(exc))