openstack/python-openstackclient
Code Issues Proposed changes

Merge "Support error handling for delete commands in volumev2"

Browse Source
This commit is contained in:
Jenkins 2016-07-18 22:41:07 +00:00 committed by Gerrit Code Review
commit 6ce53fed83
9 changed files with 364 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
openstackclient/tests/volume/v2/fakes.py
View File

@ -418,6 +418,26 @@ class FakeBackup(object):
return backups
@staticmethod
def get_backups(backups=None, count=2):
"""Get an iterable MagicMock object with a list of faked backups.
If backups list is provided, then initialize the Mock object with the
list. Otherwise create one.
:param List volumes:
A list of FakeResource objects faking backups
:param Integer count:
The number of backups to be faked
:return
An iterable Mock object with side_effect set to a list of faked
backups
"""
if backups is None:
backups = FakeBackup.create_backups(count)
return mock.MagicMock(side_effect=backups)
class FakeExtension(object):
"""Fake one or more extension."""
@ -529,6 +549,26 @@ class FakeQos(object):
return qoses
@staticmethod
def get_qoses(qoses=None, count=2):
"""Get an iterable MagicMock object with a list of faked qoses.
If qoses list is provided, then initialize the Mock object with the
list. Otherwise create one.
:param List volumes:
A list of FakeResource objects faking qoses
:param Integer count:
The number of qoses to be faked
:return
An iterable Mock object with side_effect set to a list of faked
qoses
"""
if qoses is None:
qoses = FakeQos.create_qoses(count)
return mock.MagicMock(side_effect=qoses)
class FakeSnapshot(object):
"""Fake one or more snapshot."""
@ -582,6 +622,26 @@ class FakeSnapshot(object):
return snapshots
@staticmethod
def get_snapshots(snapshots=None, count=2):
"""Get an iterable MagicMock object with a list of faked snapshots.
If snapshots list is provided, then initialize the Mock object with the
list. Otherwise create one.
:param List volumes:
A list of FakeResource objects faking snapshots
:param Integer count:
The number of snapshots to be faked
:return
An iterable Mock object with side_effect set to a list of faked
snapshots
"""
if snapshots is None:
snapshots = FakeSnapshot.create_snapshots(count)
return mock.MagicMock(side_effect=snapshots)
class FakeType(object):
"""Fake one or more type."""

70
openstackclient/tests/volume/v2/test_backup.py
View File

@ -12,6 +12,12 @@
# under the License.
#
import mock
from mock import call
from osc_lib import exceptions
from osc_lib import utils
from openstackclient.tests.volume.v2 import fakes as volume_fakes
from openstackclient.volume.v2 import backup
@ -138,12 +144,13 @@ class TestBackupCreate(TestBackup):
class TestBackupDelete(TestBackup):
backup = volume_fakes.FakeBackup.create_one_backup()
backups = volume_fakes.FakeBackup.create_backups(count=2)
def setUp(self):
super(TestBackupDelete, self).setUp()
self.backups_mock.get.return_value = self.backup
self.backups_mock.get = (
volume_fakes.FakeBackup.get_backups(self.backups))
self.backups_mock.delete.return_value = None
# Get the command object to mock
@ -151,34 +158,81 @@ class TestBackupDelete(TestBackup):
def test_backup_delete(self):
arglist = [
self.backup.id
self.backups[0].id
]
verifylist = [
("backups", [self.backup.id])
("backups", [self.backups[0].id])
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.backups_mock.delete.assert_called_with(self.backup.id, False)
self.backups_mock.delete.assert_called_with(
self.backups[0].id, False)
self.assertIsNone(result)
def test_backup_delete_with_force(self):
arglist = [
'--force',
self.backup.id,
self.backups[0].id,
]
verifylist = [
('force', True),
("backups", [self.backup.id])
("backups", [self.backups[0].id])
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.backups_mock.delete.assert_called_with(self.backup.id, True)
self.backups_mock.delete.assert_called_with(self.backups[0].id, True)
self.assertIsNone(result)
def test_delete_multiple_backups(self):
arglist = []
for b in self.backups:
arglist.append(b.id)
verifylist = [
('backups', arglist),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
calls = []
for b in self.backups:
calls.append(call(b.id, False))
self.backups_mock.delete.assert_has_calls(calls)
self.assertIsNone(result)
def test_delete_multiple_backups_with_exception(self):
arglist = [
self.backups[0].id,
'unexist_backup',
]
verifylist = [
('backups', arglist),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
find_mock_result = [self.backups[0], exceptions.CommandError]
with mock.patch.object(utils, 'find_resource',
side_effect=find_mock_result) as find_mock:
try:
self.cmd.take_action(parsed_args)
self.fail('CommandError should be raised.')
except exceptions.CommandError as e:
self.assertEqual('1 of 2 backups failed to delete.',
str(e))
find_mock.assert_any_call(self.backups_mock, self.backups[0].id)
find_mock.assert_any_call(self.backups_mock, 'unexist_backup')
self.assertEqual(2, find_mock.call_count)
self.backups_mock.delete.assert_called_once_with(
self.backups[0].id, False
)
class TestBackupList(TestBackup):

69
openstackclient/tests/volume/v2/test_qos_specs.py
View File

@ -13,6 +13,10 @@
# under the License.
#
import mock
from mock import call
from osc_lib import exceptions
from osc_lib import utils
from openstackclient.tests.volume.v2 import fakes as volume_fakes
@ -156,45 +160,94 @@ class TestQosCreate(TestQos):
class TestQosDelete(TestQos):
qos_spec = volume_fakes.FakeQos.create_one_qos()
qos_specs = volume_fakes.FakeQos.create_qoses(count=2)
def setUp(self):
super(TestQosDelete, self).setUp()
self.qos_mock.get.return_value = self.qos_spec
self.qos_mock.get = (
volume_fakes.FakeQos.get_qoses(self.qos_specs))
# Get the command object to test
self.cmd = qos_specs.DeleteQos(self.app, None)
def test_qos_delete(self):
arglist = [
self.qos_spec.id
self.qos_specs[0].id
]
verifylist = [
('qos_specs', [self.qos_spec.id])
('qos_specs', [self.qos_specs[0].id])
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.qos_mock.delete.assert_called_with(self.qos_spec.id, False)
self.qos_mock.delete.assert_called_with(
self.qos_specs[0].id, False)
self.assertIsNone(result)
def test_qos_delete_with_force(self):
arglist = [
'--force',
self.qos_spec.id
self.qos_specs[0].id
]
verifylist = [
('force', True),
('qos_specs', [self.qos_spec.id])
('qos_specs', [self.qos_specs[0].id])
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.qos_mock.delete.assert_called_with(self.qos_spec.id, True)
self.qos_mock.delete.assert_called_with(
self.qos_specs[0].id, True)
self.assertIsNone(result)
def test_delete_multiple_qoses(self):
arglist = []
for q in self.qos_specs:
arglist.append(q.id)
verifylist = [
('qos_specs', arglist),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
calls = []
for q in self.qos_specs:
calls.append(call(q.id, False))
self.qos_mock.delete.assert_has_calls(calls)
self.assertIsNone(result)
def test_delete_multiple_qoses_with_exception(self):
arglist = [
self.qos_specs[0].id,
'unexist_qos',
]
verifylist = [
('qos_specs', arglist),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
find_mock_result = [self.qos_specs[0], exceptions.CommandError]
with mock.patch.object(utils, 'find_resource',
side_effect=find_mock_result) as find_mock:
try:
self.cmd.take_action(parsed_args)
self.fail('CommandError should be raised.')
except exceptions.CommandError as e:
self.assertEqual(
'1 of 2 QoS specifications failed to delete.', str(e))
find_mock.assert_any_call(self.qos_mock, self.qos_specs[0].id)
find_mock.assert_any_call(self.qos_mock, 'unexist_qos')
self.assertEqual(2, find_mock.call_count)
self.qos_mock.delete.assert_called_once_with(
self.qos_specs[0].id, False
)
class TestQosDisassociate(TestQos):

63
openstackclient/tests/volume/v2/test_snapshot.py
View File

@ -12,6 +12,10 @@
# under the License.
#
import mock
from mock import call
from osc_lib import exceptions
from osc_lib import utils
from openstackclient.tests.volume.v2 import fakes as volume_fakes
@ -123,12 +127,13 @@ class TestSnapshotCreate(TestSnapshot):
class TestSnapshotDelete(TestSnapshot):
snapshot = volume_fakes.FakeSnapshot.create_one_snapshot()
snapshots = volume_fakes.FakeSnapshot.create_snapshots(count=2)
def setUp(self):
super(TestSnapshotDelete, self).setUp()
self.snapshots_mock.get.return_value = self.snapshot
self.snapshots_mock.get = (
volume_fakes.FakeSnapshot.get_snapshots(self.snapshots))
self.snapshots_mock.delete.return_value = None
# Get the command object to mock
@ -136,18 +141,66 @@ class TestSnapshotDelete(TestSnapshot):
def test_snapshot_delete(self):
arglist = [
self.snapshot.id
self.snapshots[0].id
]
verifylist = [
("snapshots", [self.snapshot.id])
("snapshots", [self.snapshots[0].id])
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
self.snapshots_mock.delete.assert_called_with(self.snapshot.id)
self.snapshots_mock.delete.assert_called_with(
self.snapshots[0].id)
self.assertIsNone(result)
def test_delete_multiple_snapshots(self):
arglist = []
for s in self.snapshots:
arglist.append(s.id)
verifylist = [
('snapshots', arglist),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
calls = []
for s in self.snapshots:
calls.append(call(s.id))
self.snapshots_mock.delete.assert_has_calls(calls)
self.assertIsNone(result)
def test_delete_multiple_snapshots_with_exception(self):
arglist = [
self.snapshots[0].id,
'unexist_snapshot',
]
verifylist = [
('snapshots', arglist),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
find_mock_result = [self.snapshots[0], exceptions.CommandError]
with mock.patch.object(utils, 'find_resource',
side_effect=find_mock_result) as find_mock:
try:
self.cmd.take_action(parsed_args)
self.fail('CommandError should be raised.')
except exceptions.CommandError as e:
self.assertEqual('1 of 2 snapshots failed to delete.',
str(e))
find_mock.assert_any_call(
self.snapshots_mock, self.snapshots[0].id)
find_mock.assert_any_call(self.snapshots_mock, 'unexist_snapshot')
self.assertEqual(2, find_mock.call_count)
self.snapshots_mock.delete.assert_called_once_with(
self.snapshots[0].id
)
class TestSnapshotList(TestSnapshot):

33
openstackclient/tests/volume/v2/test_volume.py
View File

@ -13,9 +13,10 @@
#
import copy
import mock
from mock import call
from osc_lib import exceptions
from osc_lib import utils
from openstackclient.tests import fakes
@ -458,6 +459,36 @@ class TestVolumeDelete(TestVolume):
self.volumes_mock.delete.assert_has_calls(calls)
self.assertIsNone(result)
def test_volume_delete_multi_volumes_with_exception(self):
volumes = self.setup_volumes_mock(count=2)
arglist = [
volumes[0].id,
'unexist_volume',
]
verifylist = [
('volumes', arglist),
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
find_mock_result = [volumes[0], exceptions.CommandError]
with mock.patch.object(utils, 'find_resource',
side_effect=find_mock_result) as find_mock:
try:
self.cmd.take_action(parsed_args)
self.fail('CommandError should be raised.')
except exceptions.CommandError as e:
self.assertEqual('1 of 2 volumes failed to delete.',
str(e))
find_mock.assert_any_call(self.volumes_mock, volumes[0].id)
find_mock.assert_any_call(self.volumes_mock, 'unexist_volume')
self.assertEqual(2, find_mock.call_count)
self.volumes_mock.delete.assert_called_once_with(
volumes[0].id
)
class TestVolumeList(TestVolume):

27
openstackclient/volume/v2/backup.py
View File

@ -15,14 +15,19 @@
"""Volume v2 Backup action implementations"""
import copy
import logging
from osc_lib.command import command
from osc_lib import exceptions
from osc_lib import utils
import six
from openstackclient.i18n import _
LOG = logging.getLogger(__name__)
class CreateBackup(command.ShowOne):
"""Create new backup"""
@ -109,10 +114,24 @@ class DeleteBackup(command.Command):
def take_action(self, parsed_args):
volume_client = self.app.client_manager.volume
for backup in parsed_args.backups:
backup_id = utils.find_resource(
volume_client.backups, backup).id
volume_client.backups.delete(backup_id, parsed_args.force)
result = 0
for i in parsed_args.backups:
try:
backup_id = utils.find_resource(
volume_client.backups, i).id
volume_client.backups.delete(backup_id, parsed_args.force)
except Exception as e:
result += 1
LOG.error(_("Failed to delete backup with "
"name or ID '%(backup)s': %(e)s")
% {'backup': i, 'e': e})
if result > 0:
total = len(parsed_args.backups)
msg = (_("%(result)s of %(total)s backups failed "
"to delete.") % {'result': result, 'total': total})
raise exceptions.CommandError(msg)
class ListBackup(command.Lister):

26
openstackclient/volume/v2/qos_specs.py
View File

@ -15,14 +15,20 @@
"""Volume v2 QoS action implementations"""
import logging
from osc_lib.cli import parseractions
from osc_lib.command import command
from osc_lib import exceptions
from osc_lib import utils
import six
from openstackclient.i18n import _
LOG = logging.getLogger(__name__)
class AssociateQos(command.Command):
"""Associate a QoS specification to a volume type"""
@ -113,9 +119,23 @@ class DeleteQos(command.Command):
def take_action(self, parsed_args):
volume_client = self.app.client_manager.volume
for qos in parsed_args.qos_specs:
qos_spec = utils.find_resource(volume_client.qos_specs, qos)
volume_client.qos_specs.delete(qos_spec.id, parsed_args.force)
result = 0
for i in parsed_args.qos_specs:
try:
qos_spec = utils.find_resource(volume_client.qos_specs, i)
volume_client.qos_specs.delete(qos_spec.id, parsed_args.force)
except Exception as e:
result += 1
LOG.error(_("Failed to delete QoS specification with "
"name or ID '%(qos)s': %(e)s")
% {'qos': i, 'e': e})
if result > 0:
total = len(parsed_args.qos_specs)
msg = (_("%(result)s of %(total)s QoS specifications failed"
" to delete.") % {'result': result, 'total': total})
raise exceptions.CommandError(msg)
class DisassociateQos(command.Command):

27
openstackclient/volume/v2/snapshot.py
View File

@ -15,15 +15,20 @@
"""Volume v2 snapshot action implementations"""
import copy
import logging
from osc_lib.cli import parseractions
from osc_lib.command import command
from osc_lib import exceptions
from osc_lib import utils
import six
from openstackclient.i18n import _
LOG = logging.getLogger(__name__)
class CreateSnapshot(command.ShowOne):
"""Create new snapshot"""
@ -92,10 +97,24 @@ class DeleteSnapshot(command.Command):
def take_action(self, parsed_args):
volume_client = self.app.client_manager.volume
for snapshot in parsed_args.snapshots:
snapshot_id = utils.find_resource(
volume_client.volume_snapshots, snapshot).id
volume_client.volume_snapshots.delete(snapshot_id)
result = 0
for i in parsed_args.snapshots:
try:
snapshot_id = utils.find_resource(
volume_client.volume_snapshots, i).id
volume_client.volume_snapshots.delete(snapshot_id)
except Exception as e:
result += 1
LOG.error(_("Failed to delete snapshot with "
"name or ID '%(snapshot)s': %(e)s")
% {'snapshot': i, 'e': e})
if result > 0:
total = len(parsed_args.snapshots)
msg = (_("%(result)s of %(total)s snapshots failed "
"to delete.") % {'result': result, 'total': total})
raise exceptions.CommandError(msg)
class ListSnapshot(command.Lister):

29
openstackclient/volume/v2/volume.py
View File

@ -19,6 +19,7 @@ import logging
from osc_lib.cli import parseractions
from osc_lib.command import command
from osc_lib import exceptions
from osc_lib import utils
import six
@ -176,13 +177,27 @@ class DeleteVolume(command.Command):
def take_action(self, parsed_args):
volume_client = self.app.client_manager.volume
for volume in parsed_args.volumes:
volume_obj = utils.find_resource(
volume_client.volumes, volume)
if parsed_args.force:
volume_client.volumes.force_delete(volume_obj.id)
else:
volume_client.volumes.delete(volume_obj.id)
result = 0
for i in parsed_args.volumes:
try:
volume_obj = utils.find_resource(
volume_client.volumes, i)
if parsed_args.force:
volume_client.volumes.force_delete(volume_obj.id)
else:
volume_client.volumes.delete(volume_obj.id)
except Exception as e:
result += 1
LOG.error(_("Failed to delete volume with "
"name or ID '%(volume)s': %(e)s")
% {'volume': i, 'e': e})
if result > 0:
total = len(parsed_args.volumes)
msg = (_("%(result)s of %(total)s volumes failed "
"to delete.") % {'result': result, 'total': total})
raise exceptions.CommandError(msg)
class ListVolume(command.Lister):