openstack/python-openstackclient
Code Issues Proposed changes

Support error handling for delete commands in volumev2

Browse Source 
Some delete conmmands in volumev2 did not support
error handling, this patch add them and also add
the unit tests for bulk deletion

Change-Id: I56ade6f9c7396c78fb989547476c4d94ccd76eae
This commit is contained in:
Huanxuan Ao 2016-07-12 15:53:58 +08:00
parent 63a6789add
commit 9b51127ecc
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):