openstack/python-openstackclient
Code Issues Proposed changes

Merge "volume: Migrate 'snapshot delete' to SDK"

Browse Source
This commit is contained in:
Zuul
2025-05-19 15:45:01 +00:00
committed by Gerrit Code Review
parent 0e47fc4e45 fc42f12eb2
commit 2e0318629d
4 changed files with 128 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

92
openstackclient/tests/unit/volume/v2/test_volume_snapshot.py
View File

@@ -13,9 +13,11 @@
from unittest import mock from unittest import mock
from openstack.block_storage.v2 import snapshot as _snapshot
from openstack import exceptions as sdk_exceptions
from openstack.test import fakes as sdk_fakes
from osc_lib.cli import format_columns from osc_lib.cli import format_columns
from osc_lib import exceptions from osc_lib import exceptions
from osc_lib import utils
from openstackclient.tests.unit.identity.v3 import fakes as project_fakes from openstackclient.tests.unit.identity.v3 import fakes as project_fakes
from openstackclient.tests.unit import utils as test_utils from openstackclient.tests.unit import utils as test_utils
@@ -184,16 +186,16 @@ class TestVolumeSnapshotCreate(TestVolumeSnapshot):
self.assertEqual(self.data, data) self.assertEqual(self.data, data)
class TestVolumeSnapshotDelete(TestVolumeSnapshot): class TestVolumeSnapshotDelete(volume_fakes.TestVolume):
snapshots = volume_fakes.create_snapshots(count=2)
def setUp(self): def setUp(self):
super().setUp() super().setUp()
self.snapshots_mock.get = volume_fakes.get_snapshots(self.snapshots) self.snapshots = list(
self.snapshots_mock.delete.return_value = None sdk_fakes.generate_fake_resources(_snapshot.Snapshot)
)
self.volume_sdk_client.find_snapshot.side_effect = self.snapshots
self.volume_sdk_client.delete_snapshot.return_value = None
# Get the command object to mock
self.cmd = volume_snapshot.DeleteVolumeSnapshot(self.app, None) self.cmd = volume_snapshot.DeleteVolumeSnapshot(self.app, None)
def test_snapshot_delete(self): def test_snapshot_delete(self):
@@ -202,24 +204,30 @@ class TestVolumeSnapshotDelete(TestVolumeSnapshot):
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) result = self.cmd.take_action(parsed_args)
self.snapshots_mock.delete.assert_called_with(
self.snapshots[0].id, False
)
self.assertIsNone(result) self.assertIsNone(result)
self.volume_sdk_client.find_snapshot.assert_called_once_with(
self.snapshots[0].id, ignore_missing=False
)
self.volume_sdk_client.delete_snapshot.assert_called_once_with(
self.snapshots[0].id, force=False
)
def test_snapshot_delete_with_force(self): def test_snapshot_delete_with_force(self):
arglist = ['--force', self.snapshots[0].id] arglist = ['--force', self.snapshots[0].id]
verifylist = [('force', True), ("snapshots", [self.snapshots[0].id])] verifylist = [('force', True), ("snapshots", [self.snapshots[0].id])]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) result = self.cmd.take_action(parsed_args)
self.snapshots_mock.delete.assert_called_with(
self.snapshots[0].id, True
)
self.assertIsNone(result) self.assertIsNone(result)
self.volume_sdk_client.find_snapshot.assert_called_once_with(
self.snapshots[0].id, ignore_missing=False
)
self.volume_sdk_client.delete_snapshot.assert_called_once_with(
self.snapshots[0].id, force=True
)
def test_delete_multiple_snapshots(self): def test_delete_multiple_snapshots(self):
arglist = [] arglist = []
for s in self.snapshots: for s in self.snapshots:
@@ -227,17 +235,24 @@ class TestVolumeSnapshotDelete(TestVolumeSnapshot):
verifylist = [ verifylist = [
('snapshots', arglist), ('snapshots', arglist),
] ]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
calls = [] result = self.cmd.take_action(parsed_args)
for s in self.snapshots:
calls.append(mock.call(s.id, False))
self.snapshots_mock.delete.assert_has_calls(calls)
self.assertIsNone(result) self.assertIsNone(result)
self.volume_sdk_client.find_snapshot.assert_has_calls(
[mock.call(x.id, ignore_missing=False) for x in self.snapshots]
)
self.volume_sdk_client.delete_snapshot.assert_has_calls(
[mock.call(x.id, force=False) for x in self.snapshots]
)
def test_delete_multiple_snapshots_with_exception(self): def test_delete_multiple_snapshots_with_exception(self):
self.volume_sdk_client.find_snapshot.side_effect = [
self.snapshots[0],
sdk_exceptions.NotFoundException(),
]
arglist = [ arglist = [
self.snapshots[0].id, self.snapshots[0].id,
'unexist_snapshot', 'unexist_snapshot',
@@ -248,25 +263,24 @@ class TestVolumeSnapshotDelete(TestVolumeSnapshot):
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
find_mock_result = [self.snapshots[0], exceptions.CommandError] exc = self.assertRaises(
with mock.patch.object( exceptions.CommandError,
utils, 'find_resource', side_effect=find_mock_result self.cmd.take_action,
) as find_mock: parsed_args,
try: )
self.cmd.take_action(parsed_args) self.assertEqual('1 of 2 snapshots failed to delete.', str(exc))
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.volume_sdk_client.find_snapshot.assert_has_calls(
self.snapshots_mock, self.snapshots[0].id [
) mock.call(self.snapshots[0].id, ignore_missing=False),
find_mock.assert_any_call(self.snapshots_mock, 'unexist_snapshot') mock.call('unexist_snapshot', ignore_missing=False),
]
self.assertEqual(2, find_mock.call_count) )
self.snapshots_mock.delete.assert_called_once_with( self.volume_sdk_client.delete_snapshot.assert_has_calls(
self.snapshots[0].id, False [
) mock.call(self.snapshots[0].id, force=False),
]
)
class TestVolumeSnapshotList(TestVolumeSnapshot): class TestVolumeSnapshotList(TestVolumeSnapshot):

106
openstackclient/tests/unit/volume/v3/test_volume_snapshot.py
View File

@@ -13,9 +13,11 @@
from unittest import mock from unittest import mock
from openstack.block_storage.v3 import snapshot as _snapshot
from openstack import exceptions as sdk_exceptions
from openstack.test import fakes as sdk_fakes
from osc_lib.cli import format_columns from osc_lib.cli import format_columns
from osc_lib import exceptions from osc_lib import exceptions
from osc_lib import utils
from openstackclient.tests.unit.identity.v3 import fakes as project_fakes from openstackclient.tests.unit.identity.v3 import fakes as project_fakes
from openstackclient.tests.unit import utils as test_utils from openstackclient.tests.unit import utils as test_utils
@@ -187,16 +189,17 @@ class TestVolumeSnapshotCreate(TestVolumeSnapshot):
self.assertEqual(self.data, data) self.assertEqual(self.data, data)
class TestVolumeSnapshotDelete(TestVolumeSnapshot): class TestVolumeSnapshotDelete(volume_fakes_v3.TestVolume):
snapshots = volume_fakes.create_snapshots(count=2)
def setUp(self): def setUp(self):
super().setUp() super().setUp()
self.snapshots_mock.get = volume_fakes.get_snapshots(self.snapshots) self.snapshots = list(
self.snapshots_mock.delete.return_value = None sdk_fakes.generate_fake_resources(_snapshot.Snapshot)
)
self.volume_sdk_client.find_snapshot.side_effect = self.snapshots
self.volume_sdk_client.delete_snapshot.return_value = None
self.volume_sdk_client.unmanage_snapshot.return_value = None
# Get the command object to mock
self.cmd = volume_snapshot.DeleteVolumeSnapshot(self.app, None) self.cmd = volume_snapshot.DeleteVolumeSnapshot(self.app, None)
def test_snapshot_delete(self): def test_snapshot_delete(self):
@@ -205,24 +208,30 @@ class TestVolumeSnapshotDelete(TestVolumeSnapshot):
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) result = self.cmd.take_action(parsed_args)
self.snapshots_mock.delete.assert_called_with(
self.snapshots[0].id, False
)
self.assertIsNone(result) self.assertIsNone(result)
self.volume_sdk_client.find_snapshot.assert_called_once_with(
self.snapshots[0].id, ignore_missing=False
)
self.volume_sdk_client.delete_snapshot.assert_called_once_with(
self.snapshots[0].id, force=False
)
def test_snapshot_delete_with_force(self): def test_snapshot_delete_with_force(self):
arglist = ['--force', self.snapshots[0].id] arglist = ['--force', self.snapshots[0].id]
verifylist = [('force', True), ("snapshots", [self.snapshots[0].id])] verifylist = [('force', True), ("snapshots", [self.snapshots[0].id])]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) result = self.cmd.take_action(parsed_args)
self.snapshots_mock.delete.assert_called_with(
self.snapshots[0].id, True
)
self.assertIsNone(result) self.assertIsNone(result)
self.volume_sdk_client.find_snapshot.assert_called_once_with(
self.snapshots[0].id, ignore_missing=False
)
self.volume_sdk_client.delete_snapshot.assert_called_once_with(
self.snapshots[0].id, force=True
)
def test_delete_multiple_snapshots(self): def test_delete_multiple_snapshots(self):
arglist = [] arglist = []
for s in self.snapshots: for s in self.snapshots:
@@ -230,17 +239,24 @@ class TestVolumeSnapshotDelete(TestVolumeSnapshot):
verifylist = [ verifylist = [
('snapshots', arglist), ('snapshots', arglist),
] ]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
calls = [] result = self.cmd.take_action(parsed_args)
for s in self.snapshots:
calls.append(mock.call(s.id, False))
self.snapshots_mock.delete.assert_has_calls(calls)
self.assertIsNone(result) self.assertIsNone(result)
self.volume_sdk_client.find_snapshot.assert_has_calls(
[mock.call(x.id, ignore_missing=False) for x in self.snapshots]
)
self.volume_sdk_client.delete_snapshot.assert_has_calls(
[mock.call(x.id, force=False) for x in self.snapshots]
)
def test_delete_multiple_snapshots_with_exception(self): def test_delete_multiple_snapshots_with_exception(self):
self.volume_sdk_client.find_snapshot.side_effect = [
self.snapshots[0],
sdk_exceptions.NotFoundException(),
]
arglist = [ arglist = [
self.snapshots[0].id, self.snapshots[0].id,
'unexist_snapshot', 'unexist_snapshot',
@@ -251,25 +267,24 @@ class TestVolumeSnapshotDelete(TestVolumeSnapshot):
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
find_mock_result = [self.snapshots[0], exceptions.CommandError] exc = self.assertRaises(
with mock.patch.object( exceptions.CommandError,
utils, 'find_resource', side_effect=find_mock_result self.cmd.take_action,
) as find_mock: parsed_args,
try: )
self.cmd.take_action(parsed_args) self.assertEqual('1 of 2 snapshots failed to delete.', str(exc))
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.volume_sdk_client.find_snapshot.assert_has_calls(
self.snapshots_mock, self.snapshots[0].id [
) mock.call(self.snapshots[0].id, ignore_missing=False),
find_mock.assert_any_call(self.snapshots_mock, 'unexist_snapshot') mock.call('unexist_snapshot', ignore_missing=False),
]
self.assertEqual(2, find_mock.call_count) )
self.snapshots_mock.delete.assert_called_once_with( self.volume_sdk_client.delete_snapshot.assert_has_calls(
self.snapshots[0].id, False [
) mock.call(self.snapshots[0].id, force=False),
]
)
def test_snapshot_delete_remote(self): def test_snapshot_delete_remote(self):
arglist = ['--remote', self.snapshots[0].id] arglist = ['--remote', self.snapshots[0].id]
@@ -277,11 +292,11 @@ class TestVolumeSnapshotDelete(TestVolumeSnapshot):
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args) result = self.cmd.take_action(parsed_args)
self.assertIsNone(result)
self.volume_sdk_client.unmanage_snapshot.assert_called_with( self.volume_sdk_client.unmanage_snapshot.assert_called_with(
self.snapshots[0].id self.snapshots[0].id
) )
self.assertIsNone(result)
def test_snapshot_delete_with_remote_force(self): def test_snapshot_delete_with_remote_force(self):
arglist = ['--remote', '--force', self.snapshots[0].id] arglist = ['--remote', '--force', self.snapshots[0].id]
@@ -305,16 +320,15 @@ class TestVolumeSnapshotDelete(TestVolumeSnapshot):
for s in self.snapshots: for s in self.snapshots:
arglist.append(s.id) arglist.append(s.id)
verifylist = [('remote', True), ('snapshots', arglist[1:])] verifylist = [('remote', True), ('snapshots', arglist[1:])]
parsed_args = self.check_parser(self.cmd, arglist, verifylist) parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
calls = [] result = self.cmd.take_action(parsed_args)
for s in self.snapshots:
calls.append(mock.call(s.id))
self.volume_sdk_client.unmanage_snapshot.assert_has_calls(calls)
self.assertIsNone(result) self.assertIsNone(result)
self.volume_sdk_client.unmanage_snapshot.assert_has_calls(
[mock.call(s.id) for s in self.snapshots]
)
class TestVolumeSnapshotList(TestVolumeSnapshot): class TestVolumeSnapshotList(TestVolumeSnapshot):
volume = volume_fakes.create_one_volume() volume = volume_fakes.create_one_volume()

14
openstackclient/volume/v2/volume_snapshot.py
View File

@@ -175,16 +175,16 @@ class DeleteVolumeSnapshot(command.Command):
return parser return parser
def take_action(self, parsed_args): def take_action(self, parsed_args):
volume_client = self.app.client_manager.volume volume_client = self.app.client_manager.sdk_connection.volume
result = 0 result = 0
for i in parsed_args.snapshots: for snapshot in parsed_args.snapshots:
try: try:
snapshot_id = utils.find_resource( snapshot_id = volume_client.find_snapshot(
volume_client.volume_snapshots, i snapshot, ignore_missing=False
).id ).id
volume_client.volume_snapshots.delete( volume_client.delete_snapshot(
snapshot_id, parsed_args.force snapshot_id, force=parsed_args.force
) )
except Exception as e: except Exception as e:
result += 1 result += 1
@@ -193,7 +193,7 @@ class DeleteVolumeSnapshot(command.Command):
"Failed to delete snapshot with " "Failed to delete snapshot with "
"name or ID '%(snapshot)s': %(e)s" "name or ID '%(snapshot)s': %(e)s"
) )
% {'snapshot': i, 'e': e} % {'snapshot': snapshot, 'e': e}
) )
if result > 0: if result > 0:

18
openstackclient/volume/v3/volume_snapshot.py
View File

@@ -182,9 +182,7 @@ class DeleteVolumeSnapshot(command.Command):
return parser return parser
def take_action(self, parsed_args): def take_action(self, parsed_args):
volume_client = self.app.client_manager.volume volume_client = self.app.client_manager.sdk_connection.volume
volume_client_sdk = self.app.client_manager.sdk_connection.volume
result = 0 result = 0
if parsed_args.remote: if parsed_args.remote:
@@ -195,16 +193,16 @@ class DeleteVolumeSnapshot(command.Command):
) )
raise exceptions.CommandError(msg) raise exceptions.CommandError(msg)
for i in parsed_args.snapshots: for snapshot in parsed_args.snapshots:
try: try:
snapshot_id = utils.find_resource( snapshot_id = volume_client.find_snapshot(
volume_client.volume_snapshots, i snapshot, ignore_missing=False
).id ).id
if parsed_args.remote: if parsed_args.remote:
volume_client_sdk.unmanage_snapshot(snapshot_id) volume_client.unmanage_snapshot(snapshot_id)
else: else:
volume_client.volume_snapshots.delete( volume_client.delete_snapshot(
snapshot_id, parsed_args.force snapshot_id, force=parsed_args.force
) )
except Exception as e: except Exception as e:
result += 1 result += 1
@@ -213,7 +211,7 @@ class DeleteVolumeSnapshot(command.Command):
"Failed to delete snapshot with " "Failed to delete snapshot with "
"name or ID '%(snapshot)s': %(e)s" "name or ID '%(snapshot)s': %(e)s"
) )
% {'snapshot': i, 'e': e} % {'snapshot': snapshot, 'e': e}
) )
if result > 0: if result > 0: