openstack/python-openstackclient
Code Issues Proposed changes

volume: Migrate 'volume migrate' to SDK

Browse Source 
Change-Id: I99af5fce0c2c184e300dfbf5624b91eeed38c94b
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane
2025-03-07 14:15:56 +00:00
parent 504cbd24e2
commit 5e5f12ba40
4 changed files with 83 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
openstackclient/tests/unit/volume/v2/test_volume.py
View File

@@ -1293,69 +1293,80 @@ class TestVolumeList(TestVolume):
self.assertIn(self.mock_volume.name, each_volume) self.assertIn(self.mock_volume.name, each_volume)
class TestVolumeMigrate(TestVolume): class TestVolumeMigrate(volume_fakes.TestVolume):
_volume = volume_fakes.create_one_volume()
def setUp(self): def setUp(self):
super().setUp() super().setUp()
self.volumes_mock.get.return_value = self._volume self.volume = sdk_fakes.generate_fake_resource(_volume.Volume)
self.volumes_mock.migrate_volume.return_value = None self.volume_sdk_client.find_volume.return_value = self.volume
# Get the command object to test self.volume_sdk_client.migrate_volume.return_value = None
self.cmd = volume.MigrateVolume(self.app, None) self.cmd = volume.MigrateVolume(self.app, None)
def test_volume_migrate(self): def test_volume_migrate(self):
arglist = [ arglist = [
"--host", "--host",
"host@backend-name#pool", "host@backend-name#pool",
self._volume.id, self.volume.id,
] ]
verifylist = [ verifylist = [
("force_host_copy", False), ("force_host_copy", False),
("lock_volume", False), ("lock_volume", False),
("host", "host@backend-name#pool"), ("host", "host@backend-name#pool"),
("volume", self._volume.id), ("volume", self.volume.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.volumes_mock.get.assert_called_once_with(self._volume.id)
self.volumes_mock.migrate_volume.assert_called_once_with(
self._volume.id, "host@backend-name#pool", False, False
)
self.assertIsNone(result) self.assertIsNone(result)
self.volume_sdk_client.find_volume.assert_called_with(
self.volume.id, ignore_missing=False
)
self.volume_sdk_client.migrate_volume.assert_called_once_with(
self.volume.id,
host="host@backend-name#pool",
force_host_copy=False,
lock_volume=False,
)
def test_volume_migrate_with_option(self): def test_volume_migrate_with_option(self):
arglist = [ arglist = [
"--force-host-copy", "--force-host-copy",
"--lock-volume", "--lock-volume",
"--host", "--host",
"host@backend-name#pool", "host@backend-name#pool",
self._volume.id, self.volume.id,
] ]
verifylist = [ verifylist = [
("force_host_copy", True), ("force_host_copy", True),
("lock_volume", True), ("lock_volume", True),
("host", "host@backend-name#pool"), ("host", "host@backend-name#pool"),
("volume", self._volume.id), ("volume", self.volume.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.volumes_mock.get.assert_called_once_with(self._volume.id)
self.volumes_mock.migrate_volume.assert_called_once_with(
self._volume.id, "host@backend-name#pool", True, True
)
self.assertIsNone(result) self.assertIsNone(result)
self.volume_sdk_client.find_volume.assert_called_with(
self.volume.id, ignore_missing=False
)
self.volume_sdk_client.migrate_volume.assert_called_once_with(
self.volume.id,
host="host@backend-name#pool",
force_host_copy=True,
lock_volume=True,
)
def test_volume_migrate_without_host(self): def test_volume_migrate_without_host(self):
arglist = [ arglist = [
self._volume.id, self.volume.id,
] ]
verifylist = [ verifylist = [
("force_host_copy", False), ("force_host_copy", False),
("lock_volume", False), ("lock_volume", False),
("volume", self._volume.id), ("volume", self.volume.id),
] ]
self.assertRaises( self.assertRaises(
@@ -1366,6 +1377,9 @@ class TestVolumeMigrate(TestVolume):
verifylist, verifylist,
) )
self.volume_sdk_client.find_volume.assert_not_called()
self.volume_sdk_client.migrate_volume.assert_not_called()
class TestVolumeSet(TestVolume): class TestVolumeSet(TestVolume):
volume_type = volume_fakes.create_one_volume_type() volume_type = volume_fakes.create_one_volume_type()

53
openstackclient/tests/unit/volume/v3/test_volume.py
View File

@@ -1665,71 +1665,79 @@ class TestVolumeList(volume_fakes.TestVolume):
class TestVolumeMigrate(volume_fakes.TestVolume): class TestVolumeMigrate(volume_fakes.TestVolume):
_volume = volume_fakes.create_one_volume()
def setUp(self): def setUp(self):
super().setUp() super().setUp()
self.volumes_mock = self.volume_client.volumes self.volume = sdk_fakes.generate_fake_resource(_volume.Volume)
self.volumes_mock.reset_mock() self.volume_sdk_client.find_volume.return_value = self.volume
self.volume_sdk_client.migrate_volume.return_value = None
self.volumes_mock.get.return_value = self._volume
self.volumes_mock.migrate_volume.return_value = None
# Get the command object to test
self.cmd = volume.MigrateVolume(self.app, None) self.cmd = volume.MigrateVolume(self.app, None)
def test_volume_migrate(self): def test_volume_migrate(self):
arglist = [ arglist = [
"--host", "--host",
"host@backend-name#pool", "host@backend-name#pool",
self._volume.id, self.volume.id,
] ]
verifylist = [ verifylist = [
("force_host_copy", False), ("force_host_copy", False),
("lock_volume", False), ("lock_volume", False),
("host", "host@backend-name#pool"), ("host", "host@backend-name#pool"),
("volume", self._volume.id), ("volume", self.volume.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.volumes_mock.get.assert_called_once_with(self._volume.id)
self.volumes_mock.migrate_volume.assert_called_once_with(
self._volume.id, "host@backend-name#pool", False, False
)
self.assertIsNone(result) self.assertIsNone(result)
self.volume_sdk_client.find_volume.assert_called_with(
self.volume.id, ignore_missing=False
)
self.volume_sdk_client.migrate_volume.assert_called_once_with(
self.volume.id,
host="host@backend-name#pool",
force_host_copy=False,
lock_volume=False,
)
def test_volume_migrate_with_option(self): def test_volume_migrate_with_option(self):
arglist = [ arglist = [
"--force-host-copy", "--force-host-copy",
"--lock-volume", "--lock-volume",
"--host", "--host",
"host@backend-name#pool", "host@backend-name#pool",
self._volume.id, self.volume.id,
] ]
verifylist = [ verifylist = [
("force_host_copy", True), ("force_host_copy", True),
("lock_volume", True), ("lock_volume", True),
("host", "host@backend-name#pool"), ("host", "host@backend-name#pool"),
("volume", self._volume.id), ("volume", self.volume.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.volumes_mock.get.assert_called_once_with(self._volume.id)
self.volumes_mock.migrate_volume.assert_called_once_with(
self._volume.id, "host@backend-name#pool", True, True
)
self.assertIsNone(result) self.assertIsNone(result)
self.volume_sdk_client.find_volume.assert_called_with(
self.volume.id, ignore_missing=False
)
self.volume_sdk_client.migrate_volume.assert_called_once_with(
self.volume.id,
host="host@backend-name#pool",
force_host_copy=True,
lock_volume=True,
)
def test_volume_migrate_without_host(self): def test_volume_migrate_without_host(self):
arglist = [ arglist = [
self._volume.id, self.volume.id,
] ]
verifylist = [ verifylist = [
("force_host_copy", False), ("force_host_copy", False),
("lock_volume", False), ("lock_volume", False),
("volume", self._volume.id), ("volume", self.volume.id),
] ]
self.assertRaises( self.assertRaises(
@@ -1740,6 +1748,9 @@ class TestVolumeMigrate(volume_fakes.TestVolume):
verifylist, verifylist,
) )
self.volume_sdk_client.find_volume.assert_not_called()
self.volume_sdk_client.migrate_volume.assert_not_called()
class TestVolumeSet(volume_fakes.TestVolume): class TestVolumeSet(volume_fakes.TestVolume):
volume_type = volume_fakes.create_one_volume_type() volume_type = volume_fakes.create_one_volume_type()

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

@@ -613,13 +613,15 @@ class MigrateVolume(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 = utils.find_resource(volume_client.volumes, parsed_args.volume) volume = volume_client.find_volume(
volume_client.volumes.migrate_volume( parsed_args.volume, ignore_missing=False
)
volume_client.migrate_volume(
volume.id, volume.id,
parsed_args.host, host=parsed_args.host,
parsed_args.force_host_copy, force_host_copy=parsed_args.force_host_copy,
parsed_args.lock_volume, lock_volume=parsed_args.lock_volume,
) )

15
openstackclient/volume/v3/volume.py
View File

@@ -761,16 +761,19 @@ class MigrateVolume(command.Command):
"(possibly by another operation)" "(possibly by another operation)"
), ),
) )
# TODO(stephenfin): Add --cluster argument
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 = utils.find_resource(volume_client.volumes, parsed_args.volume) volume = volume_client.find_volume(
volume_client.volumes.migrate_volume( parsed_args.volume, ignore_missing=False
)
volume_client.migrate_volume(
volume.id, volume.id,
parsed_args.host, host=parsed_args.host,
parsed_args.force_host_copy, force_host_copy=parsed_args.force_host_copy,
parsed_args.lock_volume, lock_volume=parsed_args.lock_volume,
) )