From e76e10c0bac9cf87c564a7f0201df189f7cd8b52 Mon Sep 17 00:00:00 2001 From: Sean McGinnis Date: Tue, 23 Oct 2018 11:34:07 -0500 Subject: Remove deprecated volume commands and args The following were deprecated over two years ago and can now be removed: * Remove ``backup`` commands in favor of ``volume backup`` * Remove ``snapshot`` commands in favor of ``volume snapshot`` * Remove ``volume create`` options ``--project``, ``--user`` and ``--multi-attach`` * Use of an auth-key positional argument in volume transfers * ``volume transfer request`` no longer accepts 'auth_key' as a positional arg, ``--auth-key`` is now required Internal (non-user-visible) * Rename backup.py to volume_backup.py for Volume v1 and v2, update tests These are backwards incompatible changes and will require a major version bump after they are merged. Change-Id: I94aa7a9824e44f9585ffb45e5e7637b9588539b4 Signed-off-by: Sean McGinnis Signed-off-by: Dean Troyer --- .../tests/functional/volume/v2/test_backup.py | 59 -- .../tests/functional/volume/v2/test_snapshot.py | 254 ------- .../functional/volume/v2/test_volume_backup.py | 59 ++ .../functional/volume/v2/test_volume_snapshot.py | 254 +++++++ .../tests/functional/volume/v3/test_snapshot.py | 18 - .../functional/volume/v3/test_volume_snapshot.py | 18 + .../tests/unit/volume/v1/test_backup.py | 394 ----------- .../tests/unit/volume/v1/test_snapshot.py | 580 ---------------- .../tests/unit/volume/v1/test_transfer_request.py | 20 - .../tests/unit/volume/v1/test_volume_backup.py | 394 +++++++++++ .../tests/unit/volume/v2/test_backup.py | 535 --------------- .../tests/unit/volume/v2/test_snapshot.py | 741 --------------------- .../tests/unit/volume/v2/test_transfer_request.py | 30 +- .../tests/unit/volume/v2/test_volume.py | 34 - .../tests/unit/volume/v2/test_volume_backup.py | 535 +++++++++++++++ openstackclient/volume/v1/backup.py | 332 --------- openstackclient/volume/v1/snapshot.py | 318 --------- openstackclient/volume/v1/volume_backup.py | 247 +++++++ .../volume/v1/volume_transfer_request.py | 22 +- openstackclient/volume/v2/backup.py | 440 ------------ openstackclient/volume/v2/snapshot.py | 351 ---------- openstackclient/volume/v2/volume.py | 36 - openstackclient/volume/v2/volume_backup.py | 355 ++++++++++ .../volume/v2/volume_transfer_request.py | 23 +- 24 files changed, 1871 insertions(+), 4178 deletions(-) delete mode 100644 openstackclient/tests/functional/volume/v2/test_backup.py delete mode 100644 openstackclient/tests/functional/volume/v2/test_snapshot.py create mode 100644 openstackclient/tests/functional/volume/v2/test_volume_backup.py create mode 100644 openstackclient/tests/functional/volume/v2/test_volume_snapshot.py delete mode 100644 openstackclient/tests/functional/volume/v3/test_snapshot.py create mode 100644 openstackclient/tests/functional/volume/v3/test_volume_snapshot.py delete mode 100644 openstackclient/tests/unit/volume/v1/test_backup.py delete mode 100644 openstackclient/tests/unit/volume/v1/test_snapshot.py create mode 100644 openstackclient/tests/unit/volume/v1/test_volume_backup.py delete mode 100644 openstackclient/tests/unit/volume/v2/test_backup.py delete mode 100644 openstackclient/tests/unit/volume/v2/test_snapshot.py create mode 100644 openstackclient/tests/unit/volume/v2/test_volume_backup.py delete mode 100644 openstackclient/volume/v1/backup.py delete mode 100644 openstackclient/volume/v1/snapshot.py create mode 100644 openstackclient/volume/v1/volume_backup.py delete mode 100644 openstackclient/volume/v2/backup.py delete mode 100644 openstackclient/volume/v2/snapshot.py create mode 100644 openstackclient/volume/v2/volume_backup.py (limited to 'openstackclient') diff --git a/openstackclient/tests/functional/volume/v2/test_backup.py b/openstackclient/tests/functional/volume/v2/test_backup.py deleted file mode 100644 index 8f5c032c..00000000 --- a/openstackclient/tests/functional/volume/v2/test_backup.py +++ /dev/null @@ -1,59 +0,0 @@ -# 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. - -import json -import uuid - -from openstackclient.tests.functional.volume.v2 import common - - -class VolumeBackupTests(common.BaseVolumeTests): - """Functional tests for volume backups. """ - - def setUp(self): - super(VolumeBackupTests, self).setUp() - self.backup_enabled = False - serv_list = json.loads(self.openstack('volume service list -f json')) - for service in serv_list: - if service['Binary'] == 'cinder-backup': - if service['Status'] == 'enabled': - self.backup_enabled = True - - def test_volume_backup_restore(self): - """Test restore backup""" - if not self.backup_enabled: - self.skipTest('Backup service is not enabled') - vol_id = uuid.uuid4().hex - # create a volume - json.loads(self.openstack( - 'volume create -f json ' + - '--size 1 ' + - vol_id - )) - self.wait_for_status("volume", vol_id, "available") - - # create a backup - backup = json.loads(self.openstack( - 'volume backup create -f json ' + - vol_id - )) - self.wait_for_status("backup", backup['id'], "available") - - # restore the backup - backup_restored = json.loads(self.openstack( - 'volume backup restore -f json %s %s' - % (backup['id'], vol_id))) - self.assertEqual(backup_restored['backup_id'], backup['id']) - self.wait_for_status("backup", backup['id'], "available") - self.wait_for_status("volume", backup_restored['volume_id'], - "available") - self.addCleanup(self.openstack, 'volume delete %s' % vol_id) diff --git a/openstackclient/tests/functional/volume/v2/test_snapshot.py b/openstackclient/tests/functional/volume/v2/test_snapshot.py deleted file mode 100644 index 264f4adb..00000000 --- a/openstackclient/tests/functional/volume/v2/test_snapshot.py +++ /dev/null @@ -1,254 +0,0 @@ -# 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. - -import json -import uuid - -from openstackclient.tests.functional.volume.v2 import common - - -class VolumeSnapshotTests(common.BaseVolumeTests): - """Functional tests for volume snapshot. """ - - VOLLY = uuid.uuid4().hex - - @classmethod - def setUpClass(cls): - super(VolumeSnapshotTests, cls).setUpClass() - # create a volume for all tests to create snapshot - cmd_output = json.loads(cls.openstack( - 'volume create -f json ' + - '--size 1 ' + - cls.VOLLY - )) - cls.wait_for_status('volume', cls.VOLLY, 'available') - cls.VOLUME_ID = cmd_output['id'] - - @classmethod - def tearDownClass(cls): - try: - cls.wait_for_status('volume', cls.VOLLY, 'available') - raw_output = cls.openstack( - 'volume delete --force ' + cls.VOLLY) - cls.assertOutput('', raw_output) - finally: - super(VolumeSnapshotTests, cls).tearDownClass() - - def test_volume_snapshot_delete(self): - """Test create, delete multiple""" - name1 = uuid.uuid4().hex - cmd_output = json.loads(self.openstack( - 'volume snapshot create -f json ' + - name1 + - ' --volume ' + self.VOLLY - )) - self.assertEqual( - name1, - cmd_output["name"], - ) - - name2 = uuid.uuid4().hex - cmd_output = json.loads(self.openstack( - 'volume snapshot create -f json ' + - name2 + - ' --volume ' + self.VOLLY - )) - self.assertEqual( - name2, - cmd_output["name"], - ) - - self.wait_for_status('volume snapshot', name1, 'available') - self.wait_for_status('volume snapshot', name2, 'available') - - del_output = self.openstack( - 'volume snapshot delete ' + name1 + ' ' + name2) - self.assertOutput('', del_output) - self.wait_for_delete('volume snapshot', name1) - self.wait_for_delete('volume snapshot', name2) - - def test_volume_snapshot_list(self): - """Test create, list filter""" - name1 = uuid.uuid4().hex - cmd_output = json.loads(self.openstack( - 'volume snapshot create -f json ' + - name1 + - ' --volume ' + self.VOLLY - )) - self.addCleanup(self.wait_for_delete, 'volume snapshot', name1) - self.addCleanup(self.openstack, 'volume snapshot delete ' + name1) - self.assertEqual( - name1, - cmd_output["name"], - ) - self.assertEqual( - self.VOLUME_ID, - cmd_output["volume_id"], - ) - self.assertEqual( - 1, - cmd_output["size"], - ) - self.wait_for_status('volume snapshot', name1, 'available') - - name2 = uuid.uuid4().hex - cmd_output = json.loads(self.openstack( - 'volume snapshot create -f json ' + - name2 + - ' --volume ' + self.VOLLY - )) - self.addCleanup(self.wait_for_delete, 'volume snapshot', name2) - self.addCleanup(self.openstack, 'volume snapshot delete ' + name2) - self.assertEqual( - name2, - cmd_output["name"], - ) - self.assertEqual( - self.VOLUME_ID, - cmd_output["volume_id"], - ) - self.assertEqual( - 1, - cmd_output["size"], - ) - self.wait_for_status('volume snapshot', name2, 'available') - raw_output = self.openstack( - 'volume snapshot set ' + - '--state error ' + - name2 - ) - self.assertOutput('', raw_output) - - # Test list --long, --status - cmd_output = json.loads(self.openstack( - 'volume snapshot list -f json ' + - '--long ' + - '--status error' - )) - names = [x["Name"] for x in cmd_output] - self.assertNotIn(name1, names) - self.assertIn(name2, names) - - # Test list --volume - cmd_output = json.loads(self.openstack( - 'volume snapshot list -f json ' + - '--volume ' + self.VOLLY - )) - names = [x["Name"] for x in cmd_output] - self.assertIn(name1, names) - self.assertIn(name2, names) - - # Test list --name - cmd_output = json.loads(self.openstack( - 'volume snapshot list -f json ' + - '--name ' + name1 - )) - names = [x["Name"] for x in cmd_output] - self.assertIn(name1, names) - self.assertNotIn(name2, names) - - def test_volume_snapshot_set(self): - """Test create, set, unset, show, delete volume snapshot""" - name = uuid.uuid4().hex - new_name = name + "_" - cmd_output = json.loads(self.openstack( - 'volume snapshot create -f json ' + - '--volume ' + self.VOLLY + - ' --description aaaa ' + - '--property Alpha=a ' + - name - )) - self.addCleanup(self.wait_for_delete, 'volume snapshot', new_name) - self.addCleanup(self.openstack, 'volume snapshot delete ' + new_name) - self.assertEqual( - name, - cmd_output["name"], - ) - self.assertEqual( - 1, - cmd_output["size"], - ) - self.assertEqual( - 'aaaa', - cmd_output["description"], - ) - self.assertEqual( - "Alpha='a'", - cmd_output["properties"], - ) - self.wait_for_status('volume snapshot', name, 'available') - - # Test volume snapshot set - raw_output = self.openstack( - 'volume snapshot set ' + - '--name ' + new_name + - ' --description bbbb ' + - '--property Alpha=c ' + - '--property Beta=b ' + - name, - ) - self.assertOutput('', raw_output) - - # Show snapshot set result - cmd_output = json.loads(self.openstack( - 'volume snapshot show -f json ' + - new_name - )) - self.assertEqual( - new_name, - cmd_output["name"], - ) - self.assertEqual( - 1, - cmd_output["size"], - ) - self.assertEqual( - 'bbbb', - cmd_output["description"], - ) - self.assertEqual( - "Alpha='c', Beta='b'", - cmd_output["properties"], - ) - - # Test volume snapshot unset - raw_output = self.openstack( - 'volume snapshot unset ' + - '--property Alpha ' + - new_name, - ) - self.assertOutput('', raw_output) - - cmd_output = json.loads(self.openstack( - 'volume snapshot show -f json ' + - new_name - )) - self.assertEqual( - "Beta='b'", - cmd_output["properties"], - ) - - # Test volume snapshot set --no-property - raw_output = self.openstack( - 'volume snapshot set ' + - '--no-property ' + - new_name, - ) - self.assertOutput('', raw_output) - cmd_output = json.loads(self.openstack( - 'volume snapshot show -f json ' + - new_name - )) - self.assertNotIn( - "Beta='b'", - cmd_output["properties"], - ) diff --git a/openstackclient/tests/functional/volume/v2/test_volume_backup.py b/openstackclient/tests/functional/volume/v2/test_volume_backup.py new file mode 100644 index 00000000..6868bd40 --- /dev/null +++ b/openstackclient/tests/functional/volume/v2/test_volume_backup.py @@ -0,0 +1,59 @@ +# 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. + +import json +import uuid + +from openstackclient.tests.functional.volume.v2 import common + + +class VolumeBackupTests(common.BaseVolumeTests): + """Functional tests for volume backups. """ + + def setUp(self): + super(VolumeBackupTests, self).setUp() + self.backup_enabled = False + serv_list = json.loads(self.openstack('volume service list -f json')) + for service in serv_list: + if service['Binary'] == 'cinder-backup': + if service['Status'] == 'enabled': + self.backup_enabled = True + + def test_volume_backup_restore(self): + """Test restore backup""" + if not self.backup_enabled: + self.skipTest('Backup service is not enabled') + vol_id = uuid.uuid4().hex + # create a volume + json.loads(self.openstack( + 'volume create -f json ' + + '--size 1 ' + + vol_id + )) + self.wait_for_status("volume", vol_id, "available") + + # create a backup + backup = json.loads(self.openstack( + 'volume backup create -f json ' + + vol_id + )) + self.wait_for_status("volume backup", backup['id'], "available") + + # restore the backup + backup_restored = json.loads(self.openstack( + 'volume backup restore -f json %s %s' + % (backup['id'], vol_id))) + self.assertEqual(backup_restored['backup_id'], backup['id']) + self.wait_for_status("volume backup", backup['id'], "available") + self.wait_for_status("volume", backup_restored['volume_id'], + "available") + self.addCleanup(self.openstack, 'volume delete %s' % vol_id) diff --git a/openstackclient/tests/functional/volume/v2/test_volume_snapshot.py b/openstackclient/tests/functional/volume/v2/test_volume_snapshot.py new file mode 100644 index 00000000..264f4adb --- /dev/null +++ b/openstackclient/tests/functional/volume/v2/test_volume_snapshot.py @@ -0,0 +1,254 @@ +# 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. + +import json +import uuid + +from openstackclient.tests.functional.volume.v2 import common + + +class VolumeSnapshotTests(common.BaseVolumeTests): + """Functional tests for volume snapshot. """ + + VOLLY = uuid.uuid4().hex + + @classmethod + def setUpClass(cls): + super(VolumeSnapshotTests, cls).setUpClass() + # create a volume for all tests to create snapshot + cmd_output = json.loads(cls.openstack( + 'volume create -f json ' + + '--size 1 ' + + cls.VOLLY + )) + cls.wait_for_status('volume', cls.VOLLY, 'available') + cls.VOLUME_ID = cmd_output['id'] + + @classmethod + def tearDownClass(cls): + try: + cls.wait_for_status('volume', cls.VOLLY, 'available') + raw_output = cls.openstack( + 'volume delete --force ' + cls.VOLLY) + cls.assertOutput('', raw_output) + finally: + super(VolumeSnapshotTests, cls).tearDownClass() + + def test_volume_snapshot_delete(self): + """Test create, delete multiple""" + name1 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume snapshot create -f json ' + + name1 + + ' --volume ' + self.VOLLY + )) + self.assertEqual( + name1, + cmd_output["name"], + ) + + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume snapshot create -f json ' + + name2 + + ' --volume ' + self.VOLLY + )) + self.assertEqual( + name2, + cmd_output["name"], + ) + + self.wait_for_status('volume snapshot', name1, 'available') + self.wait_for_status('volume snapshot', name2, 'available') + + del_output = self.openstack( + 'volume snapshot delete ' + name1 + ' ' + name2) + self.assertOutput('', del_output) + self.wait_for_delete('volume snapshot', name1) + self.wait_for_delete('volume snapshot', name2) + + def test_volume_snapshot_list(self): + """Test create, list filter""" + name1 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume snapshot create -f json ' + + name1 + + ' --volume ' + self.VOLLY + )) + self.addCleanup(self.wait_for_delete, 'volume snapshot', name1) + self.addCleanup(self.openstack, 'volume snapshot delete ' + name1) + self.assertEqual( + name1, + cmd_output["name"], + ) + self.assertEqual( + self.VOLUME_ID, + cmd_output["volume_id"], + ) + self.assertEqual( + 1, + cmd_output["size"], + ) + self.wait_for_status('volume snapshot', name1, 'available') + + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume snapshot create -f json ' + + name2 + + ' --volume ' + self.VOLLY + )) + self.addCleanup(self.wait_for_delete, 'volume snapshot', name2) + self.addCleanup(self.openstack, 'volume snapshot delete ' + name2) + self.assertEqual( + name2, + cmd_output["name"], + ) + self.assertEqual( + self.VOLUME_ID, + cmd_output["volume_id"], + ) + self.assertEqual( + 1, + cmd_output["size"], + ) + self.wait_for_status('volume snapshot', name2, 'available') + raw_output = self.openstack( + 'volume snapshot set ' + + '--state error ' + + name2 + ) + self.assertOutput('', raw_output) + + # Test list --long, --status + cmd_output = json.loads(self.openstack( + 'volume snapshot list -f json ' + + '--long ' + + '--status error' + )) + names = [x["Name"] for x in cmd_output] + self.assertNotIn(name1, names) + self.assertIn(name2, names) + + # Test list --volume + cmd_output = json.loads(self.openstack( + 'volume snapshot list -f json ' + + '--volume ' + self.VOLLY + )) + names = [x["Name"] for x in cmd_output] + self.assertIn(name1, names) + self.assertIn(name2, names) + + # Test list --name + cmd_output = json.loads(self.openstack( + 'volume snapshot list -f json ' + + '--name ' + name1 + )) + names = [x["Name"] for x in cmd_output] + self.assertIn(name1, names) + self.assertNotIn(name2, names) + + def test_volume_snapshot_set(self): + """Test create, set, unset, show, delete volume snapshot""" + name = uuid.uuid4().hex + new_name = name + "_" + cmd_output = json.loads(self.openstack( + 'volume snapshot create -f json ' + + '--volume ' + self.VOLLY + + ' --description aaaa ' + + '--property Alpha=a ' + + name + )) + self.addCleanup(self.wait_for_delete, 'volume snapshot', new_name) + self.addCleanup(self.openstack, 'volume snapshot delete ' + new_name) + self.assertEqual( + name, + cmd_output["name"], + ) + self.assertEqual( + 1, + cmd_output["size"], + ) + self.assertEqual( + 'aaaa', + cmd_output["description"], + ) + self.assertEqual( + "Alpha='a'", + cmd_output["properties"], + ) + self.wait_for_status('volume snapshot', name, 'available') + + # Test volume snapshot set + raw_output = self.openstack( + 'volume snapshot set ' + + '--name ' + new_name + + ' --description bbbb ' + + '--property Alpha=c ' + + '--property Beta=b ' + + name, + ) + self.assertOutput('', raw_output) + + # Show snapshot set result + cmd_output = json.loads(self.openstack( + 'volume snapshot show -f json ' + + new_name + )) + self.assertEqual( + new_name, + cmd_output["name"], + ) + self.assertEqual( + 1, + cmd_output["size"], + ) + self.assertEqual( + 'bbbb', + cmd_output["description"], + ) + self.assertEqual( + "Alpha='c', Beta='b'", + cmd_output["properties"], + ) + + # Test volume snapshot unset + raw_output = self.openstack( + 'volume snapshot unset ' + + '--property Alpha ' + + new_name, + ) + self.assertOutput('', raw_output) + + cmd_output = json.loads(self.openstack( + 'volume snapshot show -f json ' + + new_name + )) + self.assertEqual( + "Beta='b'", + cmd_output["properties"], + ) + + # Test volume snapshot set --no-property + raw_output = self.openstack( + 'volume snapshot set ' + + '--no-property ' + + new_name, + ) + self.assertOutput('', raw_output) + cmd_output = json.loads(self.openstack( + 'volume snapshot show -f json ' + + new_name + )) + self.assertNotIn( + "Beta='b'", + cmd_output["properties"], + ) diff --git a/openstackclient/tests/functional/volume/v3/test_snapshot.py b/openstackclient/tests/functional/volume/v3/test_snapshot.py deleted file mode 100644 index 38e0563a..00000000 --- a/openstackclient/tests/functional/volume/v3/test_snapshot.py +++ /dev/null @@ -1,18 +0,0 @@ -# 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 openstackclient.tests.functional.volume.v2 import test_snapshot as v2 -from openstackclient.tests.functional.volume.v3 import common - - -class VolumeSnapshotTests(common.BaseVolumeTests, v2.VolumeSnapshotTests): - """Functional tests for volume snapshot. """ diff --git a/openstackclient/tests/functional/volume/v3/test_volume_snapshot.py b/openstackclient/tests/functional/volume/v3/test_volume_snapshot.py new file mode 100644 index 00000000..28eee6d2 --- /dev/null +++ b/openstackclient/tests/functional/volume/v3/test_volume_snapshot.py @@ -0,0 +1,18 @@ +# 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 openstackclient.tests.functional.volume.v2 import test_volume_snapshot as v2 # noqa +from openstackclient.tests.functional.volume.v3 import common + + +class VolumeSnapshotTests(common.BaseVolumeTests, v2.VolumeSnapshotTests): + """Functional tests for volume snapshot. """ diff --git a/openstackclient/tests/unit/volume/v1/test_backup.py b/openstackclient/tests/unit/volume/v1/test_backup.py deleted file mode 100644 index 1097d3f1..00000000 --- a/openstackclient/tests/unit/volume/v1/test_backup.py +++ /dev/null @@ -1,394 +0,0 @@ -# -# 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. -# - -import mock -from mock import call - -from osc_lib import exceptions -from osc_lib import utils - -from openstackclient.tests.unit.volume.v1 import fakes as volume_fakes -from openstackclient.volume.v1 import backup - - -class TestBackup(volume_fakes.TestVolumev1): - - def setUp(self): - super(TestBackup, self).setUp() - - self.backups_mock = self.app.client_manager.volume.backups - self.backups_mock.reset_mock() - self.volumes_mock = self.app.client_manager.volume.volumes - self.volumes_mock.reset_mock() - self.snapshots_mock = self.app.client_manager.volume.volume_snapshots - self.snapshots_mock.reset_mock() - self.restores_mock = self.app.client_manager.volume.restores - self.restores_mock.reset_mock() - - -class TestBackupCreate(TestBackup): - - volume = volume_fakes.FakeVolume.create_one_volume() - - columns = ( - 'availability_zone', - 'container', - 'description', - 'id', - 'name', - 'object_count', - 'size', - 'snapshot_id', - 'status', - 'volume_id', - ) - - def setUp(self): - super(TestBackupCreate, self).setUp() - self.new_backup = volume_fakes.FakeBackup.create_one_backup( - attrs={'volume_id': self.volume.id}) - self.data = ( - self.new_backup.availability_zone, - self.new_backup.container, - self.new_backup.description, - self.new_backup.id, - self.new_backup.name, - self.new_backup.object_count, - self.new_backup.size, - self.new_backup.snapshot_id, - self.new_backup.status, - self.new_backup.volume_id, - ) - self.volumes_mock.get.return_value = self.volume - self.backups_mock.create.return_value = self.new_backup - - # Get the command object to test - self.cmd = backup.CreateVolumeBackup(self.app, None) - - def test_backup_create(self): - arglist = [ - "--name", self.new_backup.name, - "--description", self.new_backup.description, - "--container", self.new_backup.container, - self.new_backup.volume_id, - ] - verifylist = [ - ("name", self.new_backup.name), - ("description", self.new_backup.description), - ("container", self.new_backup.container), - ("volume", self.new_backup.volume_id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.backups_mock.create.assert_called_with( - self.new_backup.volume_id, - self.new_backup.container, - self.new_backup.name, - self.new_backup.description, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - def test_backup_create_without_name(self): - arglist = [ - "--description", self.new_backup.description, - "--container", self.new_backup.container, - self.new_backup.volume_id, - ] - verifylist = [ - ("description", self.new_backup.description), - ("container", self.new_backup.container), - ("volume", self.new_backup.volume_id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.backups_mock.create.assert_called_with( - self.new_backup.volume_id, - self.new_backup.container, - None, - self.new_backup.description, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - -class TestBackupDelete(TestBackup): - - backups = volume_fakes.FakeBackup.create_backups(count=2) - - def setUp(self): - super(TestBackupDelete, self).setUp() - - self.backups_mock.get = ( - volume_fakes.FakeBackup.get_backups(self.backups)) - self.backups_mock.delete.return_value = None - - # Get the command object to mock - self.cmd = backup.DeleteVolumeBackup(self.app, None) - - def test_backup_delete(self): - arglist = [ - self.backups[0].id - ] - verifylist = [ - ("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.backups[0].id) - 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)) - 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, - ) - - -class TestBackupList(TestBackup): - - volume = volume_fakes.FakeVolume.create_one_volume() - backups = volume_fakes.FakeBackup.create_backups( - attrs={'volume_id': volume.display_name}, count=3) - - columns = [ - 'ID', - 'Name', - 'Description', - 'Status', - 'Size', - ] - columns_long = columns + [ - 'Availability Zone', - 'Volume', - 'Container', - ] - - data = [] - for b in backups: - data.append(( - b.id, - b.name, - b.description, - b.status, - b.size, - )) - data_long = [] - for b in backups: - data_long.append(( - b.id, - b.name, - b.description, - b.status, - b.size, - b.availability_zone, - b.volume_id, - b.container, - )) - - def setUp(self): - super(TestBackupList, self).setUp() - - self.volumes_mock.list.return_value = [self.volume] - self.backups_mock.list.return_value = self.backups - self.volumes_mock.get.return_value = self.volume - # Get the command to test - self.cmd = backup.ListVolumeBackup(self.app, None) - - def test_backup_list_without_options(self): - arglist = [] - verifylist = [ - ("long", False), - ("name", None), - ("status", None), - ("volume", None), - ('all_projects', False), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - columns, data = self.cmd.take_action(parsed_args) - - search_opts = { - "name": None, - "status": None, - "volume_id": None, - "all_tenants": False, - } - self.volumes_mock.get.assert_not_called() - self.backups_mock.list.assert_called_with( - search_opts=search_opts, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_backup_list_with_options(self): - arglist = [ - "--long", - "--name", self.backups[0].name, - "--status", "error", - "--volume", self.volume.id, - "--all-projects" - ] - verifylist = [ - ("long", True), - ("name", self.backups[0].name), - ("status", "error"), - ("volume", self.volume.id), - ('all_projects', True), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - columns, data = self.cmd.take_action(parsed_args) - - search_opts = { - "name": self.backups[0].name, - "status": "error", - "volume_id": self.volume.id, - "all_tenants": True, - } - self.volumes_mock.get.assert_called_once_with(self.volume.id) - self.backups_mock.list.assert_called_with( - search_opts=search_opts, - ) - self.assertEqual(self.columns_long, columns) - self.assertEqual(self.data_long, list(data)) - - -class TestBackupRestore(TestBackup): - - volume = volume_fakes.FakeVolume.create_one_volume() - backup = volume_fakes.FakeBackup.create_one_backup( - attrs={'volume_id': volume.id}) - - def setUp(self): - super(TestBackupRestore, self).setUp() - - self.backups_mock.get.return_value = self.backup - self.volumes_mock.get.return_value = self.volume - self.restores_mock.restore.return_value = None - # Get the command object to mock - self.cmd = backup.RestoreVolumeBackup(self.app, None) - - def test_backup_restore(self): - arglist = [ - self.backup.id, - self.backup.volume_id - ] - verifylist = [ - ("backup", self.backup.id), - ("volume", self.backup.volume_id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.restores_mock.restore.assert_called_with(self.backup.id, - self.backup.volume_id) - self.assertIsNone(result) - - -class TestBackupShow(TestBackup): - - columns = ( - 'availability_zone', - 'container', - 'description', - 'id', - 'name', - 'object_count', - 'size', - 'snapshot_id', - 'status', - 'volume_id', - ) - - def setUp(self): - super(TestBackupShow, self).setUp() - self.backup = volume_fakes.FakeBackup.create_one_backup() - self.data = ( - self.backup.availability_zone, - self.backup.container, - self.backup.description, - self.backup.id, - self.backup.name, - self.backup.object_count, - self.backup.size, - self.backup.snapshot_id, - self.backup.status, - self.backup.volume_id, - ) - self.backups_mock.get.return_value = self.backup - # Get the command object to test - self.cmd = backup.ShowVolumeBackup(self.app, None) - - def test_backup_show(self): - arglist = [ - self.backup.id - ] - verifylist = [ - ("backup", self.backup.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.backups_mock.get.assert_called_with(self.backup.id) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) diff --git a/openstackclient/tests/unit/volume/v1/test_snapshot.py b/openstackclient/tests/unit/volume/v1/test_snapshot.py deleted file mode 100644 index 70b55ce2..00000000 --- a/openstackclient/tests/unit/volume/v1/test_snapshot.py +++ /dev/null @@ -1,580 +0,0 @@ -# -# 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. -# - -import mock -from mock import call - -from osc_lib import exceptions -from osc_lib import utils - -from openstackclient.tests.unit import utils as tests_utils -from openstackclient.tests.unit.volume.v1 import fakes as volume_fakes -from openstackclient.volume.v1 import volume_snapshot - - -class TestSnapshot(volume_fakes.TestVolumev1): - - def setUp(self): - super(TestSnapshot, self).setUp() - - self.snapshots_mock = self.app.client_manager.volume.volume_snapshots - self.snapshots_mock.reset_mock() - self.volumes_mock = self.app.client_manager.volume.volumes - self.volumes_mock.reset_mock() - - -class TestSnapshotCreate(TestSnapshot): - - columns = ( - 'created_at', - 'display_description', - 'display_name', - 'id', - 'properties', - 'size', - 'status', - 'volume_id', - ) - - def setUp(self): - super(TestSnapshotCreate, self).setUp() - - self.volume = volume_fakes.FakeVolume.create_one_volume() - self.new_snapshot = volume_fakes.FakeSnapshot.create_one_snapshot( - attrs={'volume_id': self.volume.id}) - - self.data = ( - self.new_snapshot.created_at, - self.new_snapshot.display_description, - self.new_snapshot.display_name, - self.new_snapshot.id, - utils.format_dict(self.new_snapshot.metadata), - self.new_snapshot.size, - self.new_snapshot.status, - self.new_snapshot.volume_id, - ) - - self.volumes_mock.get.return_value = self.volume - self.snapshots_mock.create.return_value = self.new_snapshot - # Get the command object to test - self.cmd = volume_snapshot.CreateVolumeSnapshot(self.app, None) - - def test_snapshot_create(self): - arglist = [ - "--volume", self.new_snapshot.volume_id, - "--description", self.new_snapshot.display_description, - "--force", - self.new_snapshot.display_name, - ] - verifylist = [ - ("volume", self.new_snapshot.volume_id), - ("description", self.new_snapshot.display_description), - ("force", True), - ("snapshot_name", self.new_snapshot.display_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.create.assert_called_with( - self.new_snapshot.volume_id, - True, - self.new_snapshot.display_name, - self.new_snapshot.display_description, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - def test_snapshot_create_without_name(self): - arglist = [ - "--volume", self.new_snapshot.volume_id, - ] - verifylist = [ - ("volume", self.new_snapshot.volume_id), - ] - self.assertRaises( - tests_utils.ParserException, - self.check_parser, - self.cmd, - arglist, - verifylist, - ) - - def test_snapshot_create_without_volume(self): - arglist = [ - "--description", self.new_snapshot.display_description, - "--force", - self.new_snapshot.display_name - ] - verifylist = [ - ("description", self.new_snapshot.display_description), - ("force", True), - ("snapshot_name", self.new_snapshot.display_name) - ] - 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.new_snapshot.display_name) - self.snapshots_mock.create.assert_called_once_with( - self.new_snapshot.volume_id, - True, - self.new_snapshot.display_name, - self.new_snapshot.display_description, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - -class TestSnapshotDelete(TestSnapshot): - - snapshots = volume_fakes.FakeSnapshot.create_snapshots(count=2) - - def setUp(self): - super(TestSnapshotDelete, self).setUp() - - self.snapshots_mock.get = ( - volume_fakes.FakeSnapshot.get_snapshots(self.snapshots)) - self.snapshots_mock.delete.return_value = None - - # Get the command object to mock - self.cmd = volume_snapshot.DeleteVolumeSnapshot(self.app, None) - - def test_snapshot_delete(self): - arglist = [ - self.snapshots[0].id - ] - verifylist = [ - ("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.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): - - volume = volume_fakes.FakeVolume.create_one_volume() - snapshots = volume_fakes.FakeSnapshot.create_snapshots( - attrs={'volume_id': volume.display_name}, count=3) - - columns = [ - "ID", - "Name", - "Description", - "Status", - "Size" - ] - columns_long = columns + [ - "Created At", - "Volume", - "Properties" - ] - - data = [] - for s in snapshots: - data.append(( - s.id, - s.display_name, - s.display_description, - s.status, - s.size, - )) - data_long = [] - for s in snapshots: - data_long.append(( - s.id, - s.display_name, - s.display_description, - s.status, - s.size, - s.created_at, - s.volume_id, - utils.format_dict(s.metadata), - )) - - def setUp(self): - super(TestSnapshotList, self).setUp() - - self.volumes_mock.list.return_value = [self.volume] - self.volumes_mock.get.return_value = self.volume - self.snapshots_mock.list.return_value = self.snapshots - # Get the command to test - self.cmd = volume_snapshot.ListVolumeSnapshot(self.app, None) - - def test_snapshot_list_without_options(self): - arglist = [] - verifylist = [ - ('all_projects', False), - ("long", False) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - search_opts={ - 'all_tenants': False, - 'display_name': None, - 'status': None, - 'volume_id': None - } - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_snapshot_list_with_long(self): - arglist = [ - "--long", - ] - verifylist = [ - ("long", True), - ('all_projects', False), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - search_opts={ - 'all_tenants': False, - 'display_name': None, - 'status': None, - 'volume_id': None - } - ) - self.assertEqual(self.columns_long, columns) - self.assertEqual(self.data_long, list(data)) - - def test_snapshot_list_name_option(self): - arglist = [ - '--name', self.snapshots[0].display_name, - ] - verifylist = [ - ('all_projects', False), - ('long', False), - ('name', self.snapshots[0].display_name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - search_opts={ - 'all_tenants': False, - 'display_name': self.snapshots[0].display_name, - 'status': None, - 'volume_id': None - } - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_snapshot_list_status_option(self): - arglist = [ - '--status', self.snapshots[0].status, - ] - verifylist = [ - ('all_projects', False), - ('long', False), - ('status', self.snapshots[0].status), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - search_opts={ - 'all_tenants': False, - 'display_name': None, - 'status': self.snapshots[0].status, - 'volume_id': None - } - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_snapshot_list_volumeid_option(self): - arglist = [ - '--volume', self.volume.id, - ] - verifylist = [ - ('all_projects', False), - ('long', False), - ('volume', self.volume.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - search_opts={ - 'all_tenants': False, - 'display_name': None, - 'status': None, - 'volume_id': self.volume.id - } - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_snapshot_list_all_projects(self): - arglist = [ - '--all-projects', - ] - verifylist = [ - ('long', False), - ('all_projects', True) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - search_opts={ - 'all_tenants': True, - 'display_name': None, - 'status': None, - 'volume_id': None - } - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - -class TestSnapshotSet(TestSnapshot): - - snapshot = volume_fakes.FakeSnapshot.create_one_snapshot() - - def setUp(self): - super(TestSnapshotSet, self).setUp() - - self.snapshots_mock.get.return_value = self.snapshot - self.snapshots_mock.set_metadata.return_value = None - # Get the command object to mock - self.cmd = volume_snapshot.SetVolumeSnapshot(self.app, None) - - def test_snapshot_set_all(self): - arglist = [ - "--name", "new_snapshot", - "--description", "new_description", - "--property", "foo_1=foo_1", - "--property", "foo_2=foo_2", - "--no-property", - self.snapshot.id, - ] - new_property = {"foo_1": "foo_1", "foo_2": "foo_2"} - verifylist = [ - ("name", "new_snapshot"), - ("description", "new_description"), - ("property", new_property), - ("no_property", True), - ("snapshot", self.snapshot.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - kwargs = { - "display_name": "new_snapshot", - "display_description": "new_description", - } - self.snapshot.update.assert_called_with(**kwargs) - self.snapshots_mock.delete_metadata.assert_called_with( - self.snapshot.id, ["foo"] - ) - self.snapshots_mock.set_metadata.assert_called_with( - self.snapshot.id, {"foo_2": "foo_2", "foo_1": "foo_1"} - ) - self.assertIsNone(result) - - def test_snapshot_set_nothing(self): - arglist = [ - self.snapshot.id, - ] - verifylist = [ - ("snapshot", self.snapshot.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.assertIsNone(result) - - def test_snapshot_set_fail(self): - self.snapshots_mock.set_metadata.side_effect = ( - exceptions.CommandError()) - arglist = [ - "--name", "new_snapshot", - "--description", "new_description", - "--property", "x=y", - "--property", "foo=foo", - self.snapshot.id, - ] - new_property = {"x": "y", "foo": "foo"} - verifylist = [ - ("name", "new_snapshot"), - ("description", "new_description"), - ("property", new_property), - ("snapshot", self.snapshot.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - self.assertRaises(exceptions.CommandError, - self.cmd.take_action, parsed_args) - - -class TestSnapshotShow(TestSnapshot): - - columns = ( - 'created_at', - 'display_description', - 'display_name', - 'id', - 'properties', - 'size', - 'status', - 'volume_id', - ) - - def setUp(self): - super(TestSnapshotShow, self).setUp() - - self.snapshot = volume_fakes.FakeSnapshot.create_one_snapshot() - - self.data = ( - self.snapshot.created_at, - self.snapshot.display_description, - self.snapshot.display_name, - self.snapshot.id, - utils.format_dict(self.snapshot.metadata), - self.snapshot.size, - self.snapshot.status, - self.snapshot.volume_id, - ) - - self.snapshots_mock.get.return_value = self.snapshot - # Get the command object to test - self.cmd = volume_snapshot.ShowVolumeSnapshot(self.app, None) - - def test_snapshot_show(self): - arglist = [ - self.snapshot.id - ] - verifylist = [ - ("snapshot", self.snapshot.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.snapshots_mock.get.assert_called_with(self.snapshot.id) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - -class TestSnapshotUnset(TestSnapshot): - - snapshot = volume_fakes.FakeSnapshot.create_one_snapshot() - - def setUp(self): - super(TestSnapshotUnset, self).setUp() - - self.snapshots_mock.get.return_value = self.snapshot - self.snapshots_mock.delete_metadata.return_value = None - # Get the command object to mock - self.cmd = volume_snapshot.UnsetVolumeSnapshot(self.app, None) - - def test_snapshot_unset(self): - arglist = [ - "--property", "foo", - self.snapshot.id, - ] - verifylist = [ - ("property", ["foo"]), - ("snapshot", self.snapshot.id), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.snapshots_mock.delete_metadata.assert_called_with( - self.snapshot.id, ["foo"] - ) - self.assertIsNone(result) - - def test_snapshot_unset_nothing(self): - arglist = [ - self.snapshot.id, - ] - verifylist = [ - ("snapshot", self.snapshot.id), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.assertIsNone(result) diff --git a/openstackclient/tests/unit/volume/v1/test_transfer_request.py b/openstackclient/tests/unit/volume/v1/test_transfer_request.py index 4c013dc0..680561d5 100644 --- a/openstackclient/tests/unit/volume/v1/test_transfer_request.py +++ b/openstackclient/tests/unit/volume/v1/test_transfer_request.py @@ -85,26 +85,6 @@ class TestTransferAccept(TestTransfer): self.assertEqual(self.columns, columns) self.assertEqual(self.data, data) - def test_transfer_accept_deprecated(self): - arglist = [ - self.volume_transfer.id, - 'key_value', - ] - verifylist = [ - ('transfer_request', self.volume_transfer.id), - ('old_auth_key', 'key_value'), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.transfer_mock.accept.assert_called_once_with( - self.volume_transfer.id, - 'key_value', - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - def test_transfer_accept_no_option(self): arglist = [ self.volume_transfer.id, diff --git a/openstackclient/tests/unit/volume/v1/test_volume_backup.py b/openstackclient/tests/unit/volume/v1/test_volume_backup.py new file mode 100644 index 00000000..22c1f8cf --- /dev/null +++ b/openstackclient/tests/unit/volume/v1/test_volume_backup.py @@ -0,0 +1,394 @@ +# +# 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. +# + +import mock +from mock import call + +from osc_lib import exceptions +from osc_lib import utils + +from openstackclient.tests.unit.volume.v1 import fakes as volume_fakes +from openstackclient.volume.v1 import volume_backup + + +class TestBackup(volume_fakes.TestVolumev1): + + def setUp(self): + super(TestBackup, self).setUp() + + self.backups_mock = self.app.client_manager.volume.backups + self.backups_mock.reset_mock() + self.volumes_mock = self.app.client_manager.volume.volumes + self.volumes_mock.reset_mock() + self.snapshots_mock = self.app.client_manager.volume.volume_snapshots + self.snapshots_mock.reset_mock() + self.restores_mock = self.app.client_manager.volume.restores + self.restores_mock.reset_mock() + + +class TestBackupCreate(TestBackup): + + volume = volume_fakes.FakeVolume.create_one_volume() + + columns = ( + 'availability_zone', + 'container', + 'description', + 'id', + 'name', + 'object_count', + 'size', + 'snapshot_id', + 'status', + 'volume_id', + ) + + def setUp(self): + super(TestBackupCreate, self).setUp() + self.new_backup = volume_fakes.FakeBackup.create_one_backup( + attrs={'volume_id': self.volume.id}) + self.data = ( + self.new_backup.availability_zone, + self.new_backup.container, + self.new_backup.description, + self.new_backup.id, + self.new_backup.name, + self.new_backup.object_count, + self.new_backup.size, + self.new_backup.snapshot_id, + self.new_backup.status, + self.new_backup.volume_id, + ) + self.volumes_mock.get.return_value = self.volume + self.backups_mock.create.return_value = self.new_backup + + # Get the command object to test + self.cmd = volume_backup.CreateVolumeBackup(self.app, None) + + def test_backup_create(self): + arglist = [ + "--name", self.new_backup.name, + "--description", self.new_backup.description, + "--container", self.new_backup.container, + self.new_backup.volume_id, + ] + verifylist = [ + ("name", self.new_backup.name), + ("description", self.new_backup.description), + ("container", self.new_backup.container), + ("volume", self.new_backup.volume_id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.backups_mock.create.assert_called_with( + self.new_backup.volume_id, + self.new_backup.container, + self.new_backup.name, + self.new_backup.description, + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + + def test_backup_create_without_name(self): + arglist = [ + "--description", self.new_backup.description, + "--container", self.new_backup.container, + self.new_backup.volume_id, + ] + verifylist = [ + ("description", self.new_backup.description), + ("container", self.new_backup.container), + ("volume", self.new_backup.volume_id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.backups_mock.create.assert_called_with( + self.new_backup.volume_id, + self.new_backup.container, + None, + self.new_backup.description, + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + + +class TestBackupDelete(TestBackup): + + backups = volume_fakes.FakeBackup.create_backups(count=2) + + def setUp(self): + super(TestBackupDelete, self).setUp() + + self.backups_mock.get = ( + volume_fakes.FakeBackup.get_backups(self.backups)) + self.backups_mock.delete.return_value = None + + # Get the command object to mock + self.cmd = volume_backup.DeleteVolumeBackup(self.app, None) + + def test_backup_delete(self): + arglist = [ + self.backups[0].id + ] + verifylist = [ + ("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.backups[0].id) + 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)) + 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, + ) + + +class TestBackupList(TestBackup): + + volume = volume_fakes.FakeVolume.create_one_volume() + backups = volume_fakes.FakeBackup.create_backups( + attrs={'volume_id': volume.display_name}, count=3) + + columns = [ + 'ID', + 'Name', + 'Description', + 'Status', + 'Size', + ] + columns_long = columns + [ + 'Availability Zone', + 'Volume', + 'Container', + ] + + data = [] + for b in backups: + data.append(( + b.id, + b.name, + b.description, + b.status, + b.size, + )) + data_long = [] + for b in backups: + data_long.append(( + b.id, + b.name, + b.description, + b.status, + b.size, + b.availability_zone, + b.volume_id, + b.container, + )) + + def setUp(self): + super(TestBackupList, self).setUp() + + self.volumes_mock.list.return_value = [self.volume] + self.backups_mock.list.return_value = self.backups + self.volumes_mock.get.return_value = self.volume + # Get the command to test + self.cmd = volume_backup.ListVolumeBackup(self.app, None) + + def test_backup_list_without_options(self): + arglist = [] + verifylist = [ + ("long", False), + ("name", None), + ("status", None), + ("volume", None), + ('all_projects', False), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = self.cmd.take_action(parsed_args) + + search_opts = { + "name": None, + "status": None, + "volume_id": None, + "all_tenants": False, + } + self.volumes_mock.get.assert_not_called() + self.backups_mock.list.assert_called_with( + search_opts=search_opts, + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + + def test_backup_list_with_options(self): + arglist = [ + "--long", + "--name", self.backups[0].name, + "--status", "error", + "--volume", self.volume.id, + "--all-projects" + ] + verifylist = [ + ("long", True), + ("name", self.backups[0].name), + ("status", "error"), + ("volume", self.volume.id), + ('all_projects', True), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = self.cmd.take_action(parsed_args) + + search_opts = { + "name": self.backups[0].name, + "status": "error", + "volume_id": self.volume.id, + "all_tenants": True, + } + self.volumes_mock.get.assert_called_once_with(self.volume.id) + self.backups_mock.list.assert_called_with( + search_opts=search_opts, + ) + self.assertEqual(self.columns_long, columns) + self.assertEqual(self.data_long, list(data)) + + +class TestBackupRestore(TestBackup): + + volume = volume_fakes.FakeVolume.create_one_volume() + backup = volume_fakes.FakeBackup.create_one_backup( + attrs={'volume_id': volume.id}) + + def setUp(self): + super(TestBackupRestore, self).setUp() + + self.backups_mock.get.return_value = self.backup + self.volumes_mock.get.return_value = self.volume + self.restores_mock.restore.return_value = None + # Get the command object to mock + self.cmd = volume_backup.RestoreVolumeBackup(self.app, None) + + def test_backup_restore(self): + arglist = [ + self.backup.id, + self.backup.volume_id + ] + verifylist = [ + ("backup", self.backup.id), + ("volume", self.backup.volume_id) + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.restores_mock.restore.assert_called_with(self.backup.id, + self.backup.volume_id) + self.assertIsNone(result) + + +class TestBackupShow(TestBackup): + + columns = ( + 'availability_zone', + 'container', + 'description', + 'id', + 'name', + 'object_count', + 'size', + 'snapshot_id', + 'status', + 'volume_id', + ) + + def setUp(self): + super(TestBackupShow, self).setUp() + self.backup = volume_fakes.FakeBackup.create_one_backup() + self.data = ( + self.backup.availability_zone, + self.backup.container, + self.backup.description, + self.backup.id, + self.backup.name, + self.backup.object_count, + self.backup.size, + self.backup.snapshot_id, + self.backup.status, + self.backup.volume_id, + ) + self.backups_mock.get.return_value = self.backup + # Get the command object to test + self.cmd = volume_backup.ShowVolumeBackup(self.app, None) + + def test_backup_show(self): + arglist = [ + self.backup.id + ] + verifylist = [ + ("backup", self.backup.id) + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + self.backups_mock.get.assert_called_with(self.backup.id) + + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) diff --git a/openstackclient/tests/unit/volume/v2/test_backup.py b/openstackclient/tests/unit/volume/v2/test_backup.py deleted file mode 100644 index 9a2ce718..00000000 --- a/openstackclient/tests/unit/volume/v2/test_backup.py +++ /dev/null @@ -1,535 +0,0 @@ -# -# 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. -# - -import mock -from mock import call - -from osc_lib import exceptions -from osc_lib import utils - -from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes -from openstackclient.volume.v2 import backup - - -class TestBackup(volume_fakes.TestVolume): - - def setUp(self): - super(TestBackup, self).setUp() - - self.backups_mock = self.app.client_manager.volume.backups - self.backups_mock.reset_mock() - self.volumes_mock = self.app.client_manager.volume.volumes - self.volumes_mock.reset_mock() - self.snapshots_mock = self.app.client_manager.volume.volume_snapshots - self.snapshots_mock.reset_mock() - self.restores_mock = self.app.client_manager.volume.restores - self.restores_mock.reset_mock() - - -class TestBackupCreate(TestBackup): - - volume = volume_fakes.FakeVolume.create_one_volume() - snapshot = volume_fakes.FakeSnapshot.create_one_snapshot() - new_backup = volume_fakes.FakeBackup.create_one_backup( - attrs={'volume_id': volume.id, 'snapshot_id': snapshot.id}) - - columns = ( - 'availability_zone', - 'container', - 'description', - 'id', - 'name', - 'object_count', - 'size', - 'snapshot_id', - 'status', - 'volume_id', - ) - data = ( - new_backup.availability_zone, - new_backup.container, - new_backup.description, - new_backup.id, - new_backup.name, - new_backup.object_count, - new_backup.size, - new_backup.snapshot_id, - new_backup.status, - new_backup.volume_id, - ) - - def setUp(self): - super(TestBackupCreate, self).setUp() - - self.volumes_mock.get.return_value = self.volume - self.snapshots_mock.get.return_value = self.snapshot - self.backups_mock.create.return_value = self.new_backup - - # Get the command object to test - self.cmd = backup.CreateVolumeBackup(self.app, None) - - def test_backup_create(self): - arglist = [ - "--name", self.new_backup.name, - "--description", self.new_backup.description, - "--container", self.new_backup.container, - "--force", - "--incremental", - "--snapshot", self.new_backup.snapshot_id, - self.new_backup.volume_id, - ] - verifylist = [ - ("name", self.new_backup.name), - ("description", self.new_backup.description), - ("container", self.new_backup.container), - ("force", True), - ("incremental", True), - ("snapshot", self.new_backup.snapshot_id), - ("volume", self.new_backup.volume_id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.backups_mock.create.assert_called_with( - self.new_backup.volume_id, - container=self.new_backup.container, - name=self.new_backup.name, - description=self.new_backup.description, - force=True, - incremental=True, - snapshot_id=self.new_backup.snapshot_id, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - def test_backup_create_without_name(self): - arglist = [ - "--description", self.new_backup.description, - "--container", self.new_backup.container, - self.new_backup.volume_id, - ] - verifylist = [ - ("description", self.new_backup.description), - ("container", self.new_backup.container), - ("volume", self.new_backup.volume_id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.backups_mock.create.assert_called_with( - self.new_backup.volume_id, - container=self.new_backup.container, - name=None, - description=self.new_backup.description, - force=False, - incremental=False, - snapshot_id=None, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - -class TestBackupDelete(TestBackup): - - backups = volume_fakes.FakeBackup.create_backups(count=2) - - def setUp(self): - super(TestBackupDelete, self).setUp() - - self.backups_mock.get = ( - volume_fakes.FakeBackup.get_backups(self.backups)) - self.backups_mock.delete.return_value = None - - # Get the command object to mock - self.cmd = backup.DeleteVolumeBackup(self.app, None) - - def test_backup_delete(self): - arglist = [ - self.backups[0].id - ] - verifylist = [ - ("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.backups[0].id, False) - self.assertIsNone(result) - - def test_backup_delete_with_force(self): - arglist = [ - '--force', - self.backups[0].id, - ] - verifylist = [ - ('force', True), - ("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.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): - - volume = volume_fakes.FakeVolume.create_one_volume() - backups = volume_fakes.FakeBackup.create_backups( - attrs={'volume_id': volume.name}, count=3) - - columns = [ - 'ID', - 'Name', - 'Description', - 'Status', - 'Size', - ] - columns_long = columns + [ - 'Availability Zone', - 'Volume', - 'Container', - ] - - data = [] - for b in backups: - data.append(( - b.id, - b.name, - b.description, - b.status, - b.size, - )) - data_long = [] - for b in backups: - data_long.append(( - b.id, - b.name, - b.description, - b.status, - b.size, - b.availability_zone, - b.volume_id, - b.container, - )) - - def setUp(self): - super(TestBackupList, self).setUp() - - self.volumes_mock.list.return_value = [self.volume] - self.backups_mock.list.return_value = self.backups - self.volumes_mock.get.return_value = self.volume - self.backups_mock.get.return_value = self.backups[0] - # Get the command to test - self.cmd = backup.ListVolumeBackup(self.app, None) - - def test_backup_list_without_options(self): - arglist = [] - verifylist = [ - ("long", False), - ("name", None), - ("status", None), - ("volume", None), - ("marker", None), - ("limit", None), - ('all_projects', False), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - columns, data = self.cmd.take_action(parsed_args) - - search_opts = { - "name": None, - "status": None, - "volume_id": None, - 'all_tenants': False, - } - self.volumes_mock.get.assert_not_called() - self.backups_mock.get.assert_not_called() - self.backups_mock.list.assert_called_with( - search_opts=search_opts, - marker=None, - limit=None, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_backup_list_with_options(self): - arglist = [ - "--long", - "--name", self.backups[0].name, - "--status", "error", - "--volume", self.volume.id, - "--marker", self.backups[0].id, - "--all-projects", - "--limit", "3", - ] - verifylist = [ - ("long", True), - ("name", self.backups[0].name), - ("status", "error"), - ("volume", self.volume.id), - ("marker", self.backups[0].id), - ('all_projects', True), - ("limit", 3), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - columns, data = self.cmd.take_action(parsed_args) - - search_opts = { - "name": self.backups[0].name, - "status": "error", - "volume_id": self.volume.id, - 'all_tenants': True, - } - self.volumes_mock.get.assert_called_once_with(self.volume.id) - self.backups_mock.get.assert_called_once_with(self.backups[0].id) - self.backups_mock.list.assert_called_with( - search_opts=search_opts, - marker=self.backups[0].id, - limit=3, - ) - self.assertEqual(self.columns_long, columns) - self.assertEqual(self.data_long, list(data)) - - -class TestBackupRestore(TestBackup): - - volume = volume_fakes.FakeVolume.create_one_volume() - backup = volume_fakes.FakeBackup.create_one_backup( - attrs={'volume_id': volume.id}) - - def setUp(self): - super(TestBackupRestore, self).setUp() - - self.backups_mock.get.return_value = self.backup - self.volumes_mock.get.return_value = self.volume - self.restores_mock.restore.return_value = ( - volume_fakes.FakeVolume.create_one_volume( - {'id': self.volume['id']})) - # Get the command object to mock - self.cmd = backup.RestoreVolumeBackup(self.app, None) - - def test_backup_restore(self): - arglist = [ - self.backup.id, - self.backup.volume_id - ] - verifylist = [ - ("backup", self.backup.id), - ("volume", self.backup.volume_id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.restores_mock.restore.assert_called_with(self.backup.id, - self.backup.volume_id) - self.assertIsNotNone(result) - - -class TestBackupSet(TestBackup): - - backup = volume_fakes.FakeBackup.create_one_backup() - - def setUp(self): - super(TestBackupSet, self).setUp() - - self.backups_mock.get.return_value = self.backup - - # Get the command object to test - self.cmd = backup.SetVolumeBackup(self.app, None) - - def test_backup_set_name(self): - arglist = [ - '--name', 'new_name', - self.backup.id, - ] - verifylist = [ - ('name', 'new_name'), - ('backup', self.backup.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - # In base command class ShowOne in cliff, abstract method take_action() - # returns nothing - result = self.cmd.take_action(parsed_args) - self.backups_mock.update.assert_called_once_with( - self.backup.id, **{'name': 'new_name'}) - self.assertIsNone(result) - - def test_backup_set_description(self): - arglist = [ - '--description', 'new_description', - self.backup.id, - ] - verifylist = [ - ('name', None), - ('description', 'new_description'), - ('backup', self.backup.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - # Set expected values - kwargs = { - 'description': 'new_description' - } - self.backups_mock.update.assert_called_once_with( - self.backup.id, - **kwargs - ) - self.assertIsNone(result) - - def test_backup_set_state(self): - arglist = [ - '--state', 'error', - self.backup.id - ] - verifylist = [ - ('state', 'error'), - ('backup', self.backup.id) - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.backups_mock.reset_state.assert_called_once_with( - self.backup.id, 'error') - self.assertIsNone(result) - - def test_backup_set_state_failed(self): - self.backups_mock.reset_state.side_effect = exceptions.CommandError() - arglist = [ - '--state', 'error', - self.backup.id - ] - verifylist = [ - ('state', 'error'), - ('backup', self.backup.id) - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - try: - self.cmd.take_action(parsed_args) - self.fail('CommandError should be raised.') - except exceptions.CommandError as e: - self.assertEqual('One or more of the set operations failed', - str(e)) - self.backups_mock.reset_state.assert_called_with( - self.backup.id, 'error') - - -class TestBackupShow(TestBackup): - - backup = volume_fakes.FakeBackup.create_one_backup() - - columns = ( - 'availability_zone', - 'container', - 'description', - 'id', - 'name', - 'object_count', - 'size', - 'snapshot_id', - 'status', - 'volume_id', - ) - data = ( - backup.availability_zone, - backup.container, - backup.description, - backup.id, - backup.name, - backup.object_count, - backup.size, - backup.snapshot_id, - backup.status, - backup.volume_id, - ) - - def setUp(self): - super(TestBackupShow, self).setUp() - - self.backups_mock.get.return_value = self.backup - # Get the command object to test - self.cmd = backup.ShowVolumeBackup(self.app, None) - - def test_backup_show(self): - arglist = [ - self.backup.id - ] - verifylist = [ - ("backup", self.backup.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.backups_mock.get.assert_called_with(self.backup.id) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) diff --git a/openstackclient/tests/unit/volume/v2/test_snapshot.py b/openstackclient/tests/unit/volume/v2/test_snapshot.py deleted file mode 100644 index e8f4ae5a..00000000 --- a/openstackclient/tests/unit/volume/v2/test_snapshot.py +++ /dev/null @@ -1,741 +0,0 @@ -# -# 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. -# - -import argparse - -import mock -from mock import call -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 import utils as tests_utils -from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes -from openstackclient.volume.v2 import volume_snapshot - - -class TestSnapshot(volume_fakes.TestVolume): - - def setUp(self): - super(TestSnapshot, self).setUp() - - self.snapshots_mock = self.app.client_manager.volume.volume_snapshots - self.snapshots_mock.reset_mock() - self.volumes_mock = self.app.client_manager.volume.volumes - self.volumes_mock.reset_mock() - self.project_mock = self.app.client_manager.identity.projects - self.project_mock.reset_mock() - - -class TestSnapshotCreate(TestSnapshot): - - columns = ( - 'created_at', - 'description', - 'id', - 'name', - 'properties', - 'size', - 'status', - 'volume_id', - ) - - def setUp(self): - super(TestSnapshotCreate, self).setUp() - - self.volume = volume_fakes.FakeVolume.create_one_volume() - self.new_snapshot = volume_fakes.FakeSnapshot.create_one_snapshot( - attrs={'volume_id': self.volume.id}) - - self.data = ( - self.new_snapshot.created_at, - self.new_snapshot.description, - self.new_snapshot.id, - self.new_snapshot.name, - utils.format_dict(self.new_snapshot.metadata), - self.new_snapshot.size, - self.new_snapshot.status, - self.new_snapshot.volume_id, - ) - - self.volumes_mock.get.return_value = self.volume - self.snapshots_mock.create.return_value = self.new_snapshot - self.snapshots_mock.manage.return_value = self.new_snapshot - # Get the command object to test - self.cmd = volume_snapshot.CreateVolumeSnapshot(self.app, None) - - def test_snapshot_create(self): - arglist = [ - "--volume", self.new_snapshot.volume_id, - "--description", self.new_snapshot.description, - "--force", - '--property', 'Alpha=a', - '--property', 'Beta=b', - self.new_snapshot.name, - ] - verifylist = [ - ("volume", self.new_snapshot.volume_id), - ("description", self.new_snapshot.description), - ("force", True), - ('property', {'Alpha': 'a', 'Beta': 'b'}), - ("snapshot_name", self.new_snapshot.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.create.assert_called_with( - self.new_snapshot.volume_id, - force=True, - name=self.new_snapshot.name, - description=self.new_snapshot.description, - metadata={'Alpha': 'a', 'Beta': 'b'}, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - def test_snapshot_create_without_name(self): - arglist = [ - "--volume", self.new_snapshot.volume_id, - ] - verifylist = [ - ("volume", self.new_snapshot.volume_id), - ] - self.assertRaises( - tests_utils.ParserException, - self.check_parser, - self.cmd, - arglist, - verifylist, - ) - - def test_snapshot_create_without_volume(self): - arglist = [ - "--description", self.new_snapshot.description, - "--force", - self.new_snapshot.name - ] - verifylist = [ - ("description", self.new_snapshot.description), - ("force", True), - ("snapshot_name", self.new_snapshot.name) - ] - 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.new_snapshot.name) - self.snapshots_mock.create.assert_called_once_with( - self.new_snapshot.volume_id, - force=True, - name=self.new_snapshot.name, - description=self.new_snapshot.description, - metadata=None, - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - def test_snapshot_create_with_remote_source(self): - arglist = [ - '--remote-source', 'source-name=test_source_name', - '--remote-source', 'source-id=test_source_id', - '--volume', self.new_snapshot.volume_id, - self.new_snapshot.name, - ] - ref_dict = {'source-name': 'test_source_name', - 'source-id': 'test_source_id'} - verifylist = [ - ('remote_source', ref_dict), - ('volume', self.new_snapshot.volume_id), - ("snapshot_name", self.new_snapshot.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.manage.assert_called_with( - volume_id=self.new_snapshot.volume_id, - ref=ref_dict, - name=self.new_snapshot.name, - description=None, - metadata=None, - ) - self.snapshots_mock.create.assert_not_called() - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - -class TestSnapshotDelete(TestSnapshot): - - snapshots = volume_fakes.FakeSnapshot.create_snapshots(count=2) - - def setUp(self): - super(TestSnapshotDelete, self).setUp() - - self.snapshots_mock.get = ( - volume_fakes.FakeSnapshot.get_snapshots(self.snapshots)) - self.snapshots_mock.delete.return_value = None - - # Get the command object to mock - self.cmd = volume_snapshot.DeleteVolumeSnapshot(self.app, None) - - def test_snapshot_delete(self): - arglist = [ - self.snapshots[0].id - ] - verifylist = [ - ("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.snapshots[0].id, False) - self.assertIsNone(result) - - def test_snapshot_delete_with_force(self): - arglist = [ - '--force', - self.snapshots[0].id - ] - verifylist = [ - ('force', True), - ("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.snapshots[0].id, True) - 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, False)) - 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, False - ) - - -class TestSnapshotList(TestSnapshot): - - volume = volume_fakes.FakeVolume.create_one_volume() - project = project_fakes.FakeProject.create_one_project() - snapshots = volume_fakes.FakeSnapshot.create_snapshots( - attrs={'volume_id': volume.name}, count=3) - - columns = [ - "ID", - "Name", - "Description", - "Status", - "Size" - ] - columns_long = columns + [ - "Created At", - "Volume", - "Properties" - ] - - data = [] - for s in snapshots: - data.append(( - s.id, - s.name, - s.description, - s.status, - s.size, - )) - data_long = [] - for s in snapshots: - data_long.append(( - s.id, - s.name, - s.description, - s.status, - s.size, - s.created_at, - s.volume_id, - utils.format_dict(s.metadata), - )) - - def setUp(self): - super(TestSnapshotList, self).setUp() - - self.volumes_mock.list.return_value = [self.volume] - self.volumes_mock.get.return_value = self.volume - self.project_mock.get.return_value = self.project - self.snapshots_mock.list.return_value = self.snapshots - # Get the command to test - self.cmd = volume_snapshot.ListVolumeSnapshot(self.app, None) - - def test_snapshot_list_without_options(self): - arglist = [] - verifylist = [ - ('all_projects', False), - ('long', False) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - limit=None, marker=None, - search_opts={ - 'all_tenants': False, - 'name': None, - 'status': None, - 'project_id': None, - 'volume_id': None - } - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_snapshot_list_with_options(self): - arglist = [ - "--long", - "--limit", "2", - "--project", self.project.id, - "--marker", self.snapshots[0].id, - ] - verifylist = [ - ("long", True), - ("limit", 2), - ("project", self.project.id), - ("marker", self.snapshots[0].id), - ('all_projects', False), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - limit=2, - marker=self.snapshots[0].id, - search_opts={ - 'all_tenants': True, - 'project_id': self.project.id, - 'name': None, - 'status': None, - 'volume_id': None - } - ) - self.assertEqual(self.columns_long, columns) - self.assertEqual(self.data_long, list(data)) - - def test_snapshot_list_all_projects(self): - arglist = [ - '--all-projects', - ] - verifylist = [ - ('long', False), - ('all_projects', True) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - limit=None, marker=None, - search_opts={ - 'all_tenants': True, - 'name': None, - 'status': None, - 'project_id': None, - 'volume_id': None - } - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_snapshot_list_name_option(self): - arglist = [ - '--name', self.snapshots[0].name, - ] - verifylist = [ - ('all_projects', False), - ('long', False), - ('name', self.snapshots[0].name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - limit=None, marker=None, - search_opts={ - 'all_tenants': False, - 'name': self.snapshots[0].name, - 'status': None, - 'project_id': None, - 'volume_id': None - } - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_snapshot_list_status_option(self): - arglist = [ - '--status', self.snapshots[0].status, - ] - verifylist = [ - ('all_projects', False), - ('long', False), - ('status', self.snapshots[0].status), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - limit=None, marker=None, - search_opts={ - 'all_tenants': False, - 'name': None, - 'status': self.snapshots[0].status, - 'project_id': None, - 'volume_id': None - } - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_snapshot_list_volumeid_option(self): - arglist = [ - '--volume', self.volume.id, - ] - verifylist = [ - ('all_projects', False), - ('long', False), - ('volume', self.volume.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.snapshots_mock.list.assert_called_once_with( - limit=None, marker=None, - search_opts={ - 'all_tenants': False, - 'name': None, - 'status': None, - 'project_id': None, - 'volume_id': self.volume.id - } - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, list(data)) - - def test_snapshot_list_negative_limit(self): - arglist = [ - "--limit", "-2", - ] - verifylist = [ - ("limit", -2), - ] - self.assertRaises(argparse.ArgumentTypeError, self.check_parser, - self.cmd, arglist, verifylist) - - -class TestSnapshotSet(TestSnapshot): - - snapshot = volume_fakes.FakeSnapshot.create_one_snapshot() - - def setUp(self): - super(TestSnapshotSet, self).setUp() - - self.snapshots_mock.get.return_value = self.snapshot - self.snapshots_mock.set_metadata.return_value = None - self.snapshots_mock.update.return_value = None - # Get the command object to mock - self.cmd = volume_snapshot.SetVolumeSnapshot(self.app, None) - - def test_snapshot_set_no_option(self): - arglist = [ - self.snapshot.id, - ] - verifylist = [ - ("snapshot", self.snapshot.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.snapshots_mock.get.assert_called_once_with(parsed_args.snapshot) - self.assertNotCalled(self.snapshots_mock.reset_state) - self.assertNotCalled(self.snapshots_mock.update) - self.assertNotCalled(self.snapshots_mock.set_metadata) - self.assertIsNone(result) - - def test_snapshot_set_name_and_property(self): - arglist = [ - "--name", "new_snapshot", - "--property", "x=y", - "--property", "foo=foo", - self.snapshot.id, - ] - new_property = {"x": "y", "foo": "foo"} - verifylist = [ - ("name", "new_snapshot"), - ("property", new_property), - ("snapshot", self.snapshot.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - kwargs = { - "name": "new_snapshot", - } - self.snapshots_mock.update.assert_called_with( - self.snapshot.id, **kwargs) - self.snapshots_mock.set_metadata.assert_called_with( - self.snapshot.id, new_property - ) - self.assertIsNone(result) - - def test_snapshot_set_with_no_property(self): - arglist = [ - "--no-property", - self.snapshot.id, - ] - verifylist = [ - ("no_property", True), - ("snapshot", self.snapshot.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.snapshots_mock.get.assert_called_once_with(parsed_args.snapshot) - self.assertNotCalled(self.snapshots_mock.reset_state) - self.assertNotCalled(self.snapshots_mock.update) - self.assertNotCalled(self.snapshots_mock.set_metadata) - self.snapshots_mock.delete_metadata.assert_called_with( - self.snapshot.id, ["foo"] - ) - self.assertIsNone(result) - - def test_snapshot_set_with_no_property_and_property(self): - arglist = [ - "--no-property", - "--property", "foo_1=bar_1", - self.snapshot.id, - ] - verifylist = [ - ("no_property", True), - ("property", {"foo_1": "bar_1"}), - ("snapshot", self.snapshot.id), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - self.snapshots_mock.get.assert_called_once_with(parsed_args.snapshot) - self.assertNotCalled(self.snapshots_mock.reset_state) - self.assertNotCalled(self.snapshots_mock.update) - self.snapshots_mock.delete_metadata.assert_called_with( - self.snapshot.id, ["foo"] - ) - self.snapshots_mock.set_metadata.assert_called_once_with( - self.snapshot.id, {"foo_1": "bar_1"}) - self.assertIsNone(result) - - def test_snapshot_set_state_to_error(self): - arglist = [ - "--state", "error", - self.snapshot.id - ] - verifylist = [ - ("state", "error"), - ("snapshot", self.snapshot.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.snapshots_mock.reset_state.assert_called_with( - self.snapshot.id, "error") - self.assertIsNone(result) - - def test_volume_set_state_failed(self): - self.snapshots_mock.reset_state.side_effect = exceptions.CommandError() - arglist = [ - '--state', 'error', - self.snapshot.id - ] - verifylist = [ - ('state', 'error'), - ('snapshot', self.snapshot.id) - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - try: - self.cmd.take_action(parsed_args) - self.fail('CommandError should be raised.') - except exceptions.CommandError as e: - self.assertEqual('One or more of the set operations failed', - str(e)) - self.snapshots_mock.reset_state.assert_called_once_with( - self.snapshot.id, 'error') - - def test_volume_set_name_and_state_failed(self): - self.snapshots_mock.reset_state.side_effect = exceptions.CommandError() - arglist = [ - '--state', 'error', - "--name", "new_snapshot", - self.snapshot.id - ] - verifylist = [ - ('state', 'error'), - ("name", "new_snapshot"), - ('snapshot', self.snapshot.id) - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - try: - self.cmd.take_action(parsed_args) - self.fail('CommandError should be raised.') - except exceptions.CommandError as e: - self.assertEqual('One or more of the set operations failed', - str(e)) - kwargs = { - "name": "new_snapshot", - } - self.snapshots_mock.update.assert_called_once_with( - self.snapshot.id, **kwargs) - self.snapshots_mock.reset_state.assert_called_once_with( - self.snapshot.id, 'error') - - -class TestSnapshotShow(TestSnapshot): - - columns = ( - 'created_at', - 'description', - 'id', - 'name', - 'properties', - 'size', - 'status', - 'volume_id', - ) - - def setUp(self): - super(TestSnapshotShow, self).setUp() - - self.snapshot = volume_fakes.FakeSnapshot.create_one_snapshot() - - self.data = ( - self.snapshot.created_at, - self.snapshot.description, - self.snapshot.id, - self.snapshot.name, - utils.format_dict(self.snapshot.metadata), - self.snapshot.size, - self.snapshot.status, - self.snapshot.volume_id, - ) - - self.snapshots_mock.get.return_value = self.snapshot - # Get the command object to test - self.cmd = volume_snapshot.ShowVolumeSnapshot(self.app, None) - - def test_snapshot_show(self): - arglist = [ - self.snapshot.id - ] - verifylist = [ - ("snapshot", self.snapshot.id) - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - self.snapshots_mock.get.assert_called_with(self.snapshot.id) - - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - - -class TestSnapshotUnset(TestSnapshot): - - snapshot = volume_fakes.FakeSnapshot.create_one_snapshot() - - def setUp(self): - super(TestSnapshotUnset, self).setUp() - - self.snapshots_mock.get.return_value = self.snapshot - self.snapshots_mock.delete_metadata.return_value = None - # Get the command object to mock - self.cmd = volume_snapshot.UnsetVolumeSnapshot(self.app, None) - - def test_snapshot_unset(self): - arglist = [ - "--property", "foo", - self.snapshot.id, - ] - verifylist = [ - ("property", ["foo"]), - ("snapshot", self.snapshot.id), - ] - - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - result = self.cmd.take_action(parsed_args) - - self.snapshots_mock.delete_metadata.assert_called_with( - self.snapshot.id, ["foo"] - ) - self.assertIsNone(result) diff --git a/openstackclient/tests/unit/volume/v2/test_transfer_request.py b/openstackclient/tests/unit/volume/v2/test_transfer_request.py index 37eed11e..1ea6648f 100644 --- a/openstackclient/tests/unit/volume/v2/test_transfer_request.py +++ b/openstackclient/tests/unit/volume/v2/test_transfer_request.py @@ -18,6 +18,7 @@ from mock import call from osc_lib import exceptions from osc_lib import utils +from openstackclient.tests.unit import utils as test_utils from openstackclient.tests.unit.volume.v2 import fakes as transfer_fakes from openstackclient.volume.v2 import volume_transfer_request @@ -85,26 +86,6 @@ class TestTransferAccept(TestTransfer): self.assertEqual(self.columns, columns) self.assertEqual(self.data, data) - def test_transfer_accept_deprecated(self): - arglist = [ - self.volume_transfer.id, - 'key_value', - ] - verifylist = [ - ('transfer_request', self.volume_transfer.id), - ('old_auth_key', 'key_value'), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - columns, data = self.cmd.take_action(parsed_args) - - self.transfer_mock.accept.assert_called_once_with( - self.volume_transfer.id, - 'key_value', - ) - self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) - def test_transfer_accept_no_option(self): arglist = [ self.volume_transfer.id, @@ -112,12 +93,13 @@ class TestTransferAccept(TestTransfer): verifylist = [ ('transfer_request', self.volume_transfer.id), ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) self.assertRaises( - exceptions.CommandError, - self.cmd.take_action, - parsed_args, + test_utils.ParserException, + self.check_parser, + self.cmd, + arglist, + verifylist, ) diff --git a/openstackclient/tests/unit/volume/v2/test_volume.py b/openstackclient/tests/unit/volume/v2/test_volume.py index dbe69ea0..97da1601 100644 --- a/openstackclient/tests/unit/volume/v2/test_volume.py +++ b/openstackclient/tests/unit/volume/v2/test_volume.py @@ -183,40 +183,6 @@ class TestVolumeCreate(TestVolume): self.assertEqual(self.columns, columns) self.assertEqual(self.datalist, data) - def test_volume_create_user(self): - arglist = [ - '--size', str(self.new_volume.size), - '--user', self.user.id, - self.new_volume.name, - ] - verifylist = [ - ('size', self.new_volume.size), - ('user', self.user.id), - ('name', self.new_volume.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - self.assertRaises(exceptions.CommandError, self.cmd.take_action, - parsed_args) - self.volumes_mock.create.assert_not_called() - - def test_volume_create_project(self): - arglist = [ - '--size', str(self.new_volume.size), - '--project', self.project.id, - self.new_volume.name, - ] - verifylist = [ - ('size', self.new_volume.size), - ('project', self.project.id), - ('name', self.new_volume.name), - ] - parsed_args = self.check_parser(self.cmd, arglist, verifylist) - - self.assertRaises(exceptions.CommandError, self.cmd.take_action, - parsed_args) - self.volumes_mock.create.assert_not_called() - def test_volume_create_properties(self): arglist = [ '--property', 'Alpha=a', diff --git a/openstackclient/tests/unit/volume/v2/test_volume_backup.py b/openstackclient/tests/unit/volume/v2/test_volume_backup.py new file mode 100644 index 00000000..d0d1da3b --- /dev/null +++ b/openstackclient/tests/unit/volume/v2/test_volume_backup.py @@ -0,0 +1,535 @@ +# +# 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. +# + +import mock +from mock import call + +from osc_lib import exceptions +from osc_lib import utils + +from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes +from openstackclient.volume.v2 import volume_backup + + +class TestBackup(volume_fakes.TestVolume): + + def setUp(self): + super(TestBackup, self).setUp() + + self.backups_mock = self.app.client_manager.volume.backups + self.backups_mock.reset_mock() + self.volumes_mock = self.app.client_manager.volume.volumes + self.volumes_mock.reset_mock() + self.snapshots_mock = self.app.client_manager.volume.volume_snapshots + self.snapshots_mock.reset_mock() + self.restores_mock = self.app.client_manager.volume.restores + self.restores_mock.reset_mock() + + +class TestBackupCreate(TestBackup): + + volume = volume_fakes.FakeVolume.create_one_volume() + snapshot = volume_fakes.FakeSnapshot.create_one_snapshot() + new_backup = volume_fakes.FakeBackup.create_one_backup( + attrs={'volume_id': volume.id, 'snapshot_id': snapshot.id}) + + columns = ( + 'availability_zone', + 'container', + 'description', + 'id', + 'name', + 'object_count', + 'size', + 'snapshot_id', + 'status', + 'volume_id', + ) + data = ( + new_backup.availability_zone, + new_backup.container, + new_backup.description, + new_backup.id, + new_backup.name, + new_backup.object_count, + new_backup.size, + new_backup.snapshot_id, + new_backup.status, + new_backup.volume_id, + ) + + def setUp(self): + super(TestBackupCreate, self).setUp() + + self.volumes_mock.get.return_value = self.volume + self.snapshots_mock.get.return_value = self.snapshot + self.backups_mock.create.return_value = self.new_backup + + # Get the command object to test + self.cmd = volume_backup.CreateVolumeBackup(self.app, None) + + def test_backup_create(self): + arglist = [ + "--name", self.new_backup.name, + "--description", self.new_backup.description, + "--container", self.new_backup.container, + "--force", + "--incremental", + "--snapshot", self.new_backup.snapshot_id, + self.new_backup.volume_id, + ] + verifylist = [ + ("name", self.new_backup.name), + ("description", self.new_backup.description), + ("container", self.new_backup.container), + ("force", True), + ("incremental", True), + ("snapshot", self.new_backup.snapshot_id), + ("volume", self.new_backup.volume_id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.backups_mock.create.assert_called_with( + self.new_backup.volume_id, + container=self.new_backup.container, + name=self.new_backup.name, + description=self.new_backup.description, + force=True, + incremental=True, + snapshot_id=self.new_backup.snapshot_id, + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + + def test_backup_create_without_name(self): + arglist = [ + "--description", self.new_backup.description, + "--container", self.new_backup.container, + self.new_backup.volume_id, + ] + verifylist = [ + ("description", self.new_backup.description), + ("container", self.new_backup.container), + ("volume", self.new_backup.volume_id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.backups_mock.create.assert_called_with( + self.new_backup.volume_id, + container=self.new_backup.container, + name=None, + description=self.new_backup.description, + force=False, + incremental=False, + snapshot_id=None, + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + + +class TestBackupDelete(TestBackup): + + backups = volume_fakes.FakeBackup.create_backups(count=2) + + def setUp(self): + super(TestBackupDelete, self).setUp() + + self.backups_mock.get = ( + volume_fakes.FakeBackup.get_backups(self.backups)) + self.backups_mock.delete.return_value = None + + # Get the command object to mock + self.cmd = volume_backup.DeleteVolumeBackup(self.app, None) + + def test_backup_delete(self): + arglist = [ + self.backups[0].id + ] + verifylist = [ + ("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.backups[0].id, False) + self.assertIsNone(result) + + def test_backup_delete_with_force(self): + arglist = [ + '--force', + self.backups[0].id, + ] + verifylist = [ + ('force', True), + ("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.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): + + volume = volume_fakes.FakeVolume.create_one_volume() + backups = volume_fakes.FakeBackup.create_backups( + attrs={'volume_id': volume.name}, count=3) + + columns = [ + 'ID', + 'Name', + 'Description', + 'Status', + 'Size', + ] + columns_long = columns + [ + 'Availability Zone', + 'Volume', + 'Container', + ] + + data = [] + for b in backups: + data.append(( + b.id, + b.name, + b.description, + b.status, + b.size, + )) + data_long = [] + for b in backups: + data_long.append(( + b.id, + b.name, + b.description, + b.status, + b.size, + b.availability_zone, + b.volume_id, + b.container, + )) + + def setUp(self): + super(TestBackupList, self).setUp() + + self.volumes_mock.list.return_value = [self.volume] + self.backups_mock.list.return_value = self.backups + self.volumes_mock.get.return_value = self.volume + self.backups_mock.get.return_value = self.backups[0] + # Get the command to test + self.cmd = volume_backup.ListVolumeBackup(self.app, None) + + def test_backup_list_without_options(self): + arglist = [] + verifylist = [ + ("long", False), + ("name", None), + ("status", None), + ("volume", None), + ("marker", None), + ("limit", None), + ('all_projects', False), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = self.cmd.take_action(parsed_args) + + search_opts = { + "name": None, + "status": None, + "volume_id": None, + 'all_tenants': False, + } + self.volumes_mock.get.assert_not_called() + self.backups_mock.get.assert_not_called() + self.backups_mock.list.assert_called_with( + search_opts=search_opts, + marker=None, + limit=None, + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + + def test_backup_list_with_options(self): + arglist = [ + "--long", + "--name", self.backups[0].name, + "--status", "error", + "--volume", self.volume.id, + "--marker", self.backups[0].id, + "--all-projects", + "--limit", "3", + ] + verifylist = [ + ("long", True), + ("name", self.backups[0].name), + ("status", "error"), + ("volume", self.volume.id), + ("marker", self.backups[0].id), + ('all_projects', True), + ("limit", 3), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = self.cmd.take_action(parsed_args) + + search_opts = { + "name": self.backups[0].name, + "status": "error", + "volume_id": self.volume.id, + 'all_tenants': True, + } + self.volumes_mock.get.assert_called_once_with(self.volume.id) + self.backups_mock.get.assert_called_once_with(self.backups[0].id) + self.backups_mock.list.assert_called_with( + search_opts=search_opts, + marker=self.backups[0].id, + limit=3, + ) + self.assertEqual(self.columns_long, columns) + self.assertEqual(self.data_long, list(data)) + + +class TestBackupRestore(TestBackup): + + volume = volume_fakes.FakeVolume.create_one_volume() + backup = volume_fakes.FakeBackup.create_one_backup( + attrs={'volume_id': volume.id}) + + def setUp(self): + super(TestBackupRestore, self).setUp() + + self.backups_mock.get.return_value = self.backup + self.volumes_mock.get.return_value = self.volume + self.restores_mock.restore.return_value = ( + volume_fakes.FakeVolume.create_one_volume( + {'id': self.volume['id']})) + # Get the command object to mock + self.cmd = volume_backup.RestoreVolumeBackup(self.app, None) + + def test_backup_restore(self): + arglist = [ + self.backup.id, + self.backup.volume_id + ] + verifylist = [ + ("backup", self.backup.id), + ("volume", self.backup.volume_id) + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.restores_mock.restore.assert_called_with(self.backup.id, + self.backup.volume_id) + self.assertIsNotNone(result) + + +class TestBackupSet(TestBackup): + + backup = volume_fakes.FakeBackup.create_one_backup() + + def setUp(self): + super(TestBackupSet, self).setUp() + + self.backups_mock.get.return_value = self.backup + + # Get the command object to test + self.cmd = volume_backup.SetVolumeBackup(self.app, None) + + def test_backup_set_name(self): + arglist = [ + '--name', 'new_name', + self.backup.id, + ] + verifylist = [ + ('name', 'new_name'), + ('backup', self.backup.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + # In base command class ShowOne in cliff, abstract method take_action() + # returns nothing + result = self.cmd.take_action(parsed_args) + self.backups_mock.update.assert_called_once_with( + self.backup.id, **{'name': 'new_name'}) + self.assertIsNone(result) + + def test_backup_set_description(self): + arglist = [ + '--description', 'new_description', + self.backup.id, + ] + verifylist = [ + ('name', None), + ('description', 'new_description'), + ('backup', self.backup.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + # Set expected values + kwargs = { + 'description': 'new_description' + } + self.backups_mock.update.assert_called_once_with( + self.backup.id, + **kwargs + ) + self.assertIsNone(result) + + def test_backup_set_state(self): + arglist = [ + '--state', 'error', + self.backup.id + ] + verifylist = [ + ('state', 'error'), + ('backup', self.backup.id) + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + self.backups_mock.reset_state.assert_called_once_with( + self.backup.id, 'error') + self.assertIsNone(result) + + def test_backup_set_state_failed(self): + self.backups_mock.reset_state.side_effect = exceptions.CommandError() + arglist = [ + '--state', 'error', + self.backup.id + ] + verifylist = [ + ('state', 'error'), + ('backup', self.backup.id) + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('One or more of the set operations failed', + str(e)) + self.backups_mock.reset_state.assert_called_with( + self.backup.id, 'error') + + +class TestBackupShow(TestBackup): + + backup = volume_fakes.FakeBackup.create_one_backup() + + columns = ( + 'availability_zone', + 'container', + 'description', + 'id', + 'name', + 'object_count', + 'size', + 'snapshot_id', + 'status', + 'volume_id', + ) + data = ( + backup.availability_zone, + backup.container, + backup.description, + backup.id, + backup.name, + backup.object_count, + backup.size, + backup.snapshot_id, + backup.status, + backup.volume_id, + ) + + def setUp(self): + super(TestBackupShow, self).setUp() + + self.backups_mock.get.return_value = self.backup + # Get the command object to test + self.cmd = volume_backup.ShowVolumeBackup(self.app, None) + + def test_backup_show(self): + arglist = [ + self.backup.id + ] + verifylist = [ + ("backup", self.backup.id) + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + self.backups_mock.get.assert_called_with(self.backup.id) + + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) diff --git a/openstackclient/volume/v1/backup.py b/openstackclient/volume/v1/backup.py deleted file mode 100644 index 9ac1302a..00000000 --- a/openstackclient/volume/v1/backup.py +++ /dev/null @@ -1,332 +0,0 @@ -# Copyright 2012-2013 OpenStack Foundation -# -# 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. -# - -"""Volume v1 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 CreateVolumeBackup(command.ShowOne): - _description = _("Create new volume backup") - - def get_parser(self, prog_name): - parser = super(CreateVolumeBackup, self).get_parser(prog_name) - parser.add_argument( - 'volume', - metavar='', - help=_('Volume to backup (name or ID)'), - ) - parser.add_argument( - '--container', - metavar='', - required=False, - help=_('Optional backup container name'), - ) - parser.add_argument( - '--name', - metavar='', - help=_('Name of the backup'), - ) - parser.add_argument( - '--description', - metavar='', - help=_('Description of the backup'), - ) - return parser - - def take_action(self, parsed_args): - volume_client = self.app.client_manager.volume - volume_id = utils.find_resource(volume_client.volumes, - parsed_args.volume).id - backup = volume_client.backups.create( - volume_id, - parsed_args.container, - parsed_args.name, - parsed_args.description - ) - - backup._info.pop('links') - return zip(*sorted(six.iteritems(backup._info))) - - -class CreateBackup(CreateVolumeBackup): - _description = _("Create new backup") - - # TODO(Huanxuan Ao): Remove this class and ``backup create`` command - # two cycles after Newton. - - # This notifies cliff to not display the help for this command - deprecated = True - - log = logging.getLogger('deprecated') - - def take_action(self, parsed_args): - self.log.warning(_('This command has been deprecated. ' - 'Please use "volume backup create" instead.')) - return super(CreateBackup, self).take_action(parsed_args) - - -class DeleteVolumeBackup(command.Command): - _description = _("Delete volume backup(s)") - - def get_parser(self, prog_name): - parser = super(DeleteVolumeBackup, self).get_parser(prog_name) - parser.add_argument( - 'backups', - metavar='', - nargs="+", - help=_('Backup(s) to delete (name or ID)'), - ) - return parser - - def take_action(self, parsed_args): - volume_client = self.app.client_manager.volume - 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) - 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 DeleteBackup(DeleteVolumeBackup): - _description = _("Delete backup(s)") - - # TODO(Huanxuan Ao): Remove this class and ``backup delete`` command - # two cycles after Newton. - - # This notifies cliff to not display the help for this command - deprecated = True - - log = logging.getLogger('deprecated') - - def take_action(self, parsed_args): - self.log.warning(_('This command has been deprecated. ' - 'Please use "volume backup delete" instead.')) - return super(DeleteBackup, self).take_action(parsed_args) - - -class ListVolumeBackup(command.Lister): - _description = _("List volume backups") - - def get_parser(self, prog_name): - parser = super(ListVolumeBackup, self).get_parser(prog_name) - parser.add_argument( - '--long', - action='store_true', - default=False, - help=_('List additional fields in output'), - ) - parser.add_argument( - "--name", - metavar="", - help=_("Filters results by the backup name") - ) - parser.add_argument( - "--status", - metavar="", - choices=['creating', 'available', 'deleting', - 'error', 'restoring', 'error_restoring'], - help=_("Filters results by the backup status " - "('creating', 'available', 'deleting', " - "'error', 'restoring' or 'error_restoring')") - ) - parser.add_argument( - "--volume", - metavar="", - help=_("Filters results by the volume which they " - "backup (name or ID)") - ) - parser.add_argument( - '--all-projects', - action='store_true', - default=False, - help=_('Include all projects (admin only)'), - ) - return parser - - def take_action(self, parsed_args): - volume_client = self.app.client_manager.volume - - def _format_volume_id(volume_id): - """Return a volume name if available - - :param volume_id: a volume ID - :rtype: either the volume ID or name - """ - - volume = volume_id - if volume_id in volume_cache.keys(): - volume = volume_cache[volume_id].display_name - return volume - - if parsed_args.long: - columns = ['ID', 'Name', 'Description', 'Status', 'Size', - 'Availability Zone', 'Volume ID', 'Container'] - column_headers = copy.deepcopy(columns) - column_headers[6] = 'Volume' - else: - columns = ['ID', 'Name', 'Description', 'Status', 'Size'] - column_headers = columns - - # Cache the volume list - volume_cache = {} - try: - for s in volume_client.volumes.list(): - volume_cache[s.id] = s - except Exception: - # Just forget it if there's any trouble - pass - - filter_volume_id = None - if parsed_args.volume: - filter_volume_id = utils.find_resource(volume_client.volumes, - parsed_args.volume).id - search_opts = { - 'name': parsed_args.name, - 'status': parsed_args.status, - 'volume_id': filter_volume_id, - 'all_tenants': parsed_args.all_projects, - } - data = volume_client.backups.list( - search_opts=search_opts, - ) - - return (column_headers, - (utils.get_item_properties( - s, columns, - formatters={'Volume ID': _format_volume_id}, - ) for s in data)) - - -class ListBackup(ListVolumeBackup): - _description = _("List backups") - - # TODO(Huanxuan Ao): Remove this class and ``backup list`` command - # two cycles after Newton. - - # This notifies cliff to not display the help for this command - deprecated = True - - log = logging.getLogger('deprecated') - - def take_action(self, parsed_args): - self.log.warning(_('This command has been deprecated. ' - 'Please use "volume backup list" instead.')) - return super(ListBackup, self).take_action(parsed_args) - - -class RestoreVolumeBackup(command.Command): - _description = _("Restore volume backup") - - def get_parser(self, prog_name): - parser = super(RestoreVolumeBackup, self).get_parser(prog_name) - parser.add_argument( - 'backup', - metavar='', - help=_('Backup to restore (name or ID)') - ) - parser.add_argument( - 'volume', - metavar='', - help=_('Volume to restore to (name or ID)') - ) - return parser - - def take_action(self, parsed_args): - volume_client = self.app.client_manager.volume - backup = utils.find_resource(volume_client.backups, - parsed_args.backup) - destination_volume = utils.find_resource(volume_client.volumes, - parsed_args.volume) - return volume_client.restores.restore(backup.id, - destination_volume.id) - - -class RestoreBackup(RestoreVolumeBackup): - _description = _("Restore backup") - - # TODO(Huanxuan Ao): Remove this class and ``backup restore`` command - # two cycles after Newton. - - # This notifies cliff to not display the help for this command - deprecated = True - - log = logging.getLogger('deprecated') - - def take_action(self, parsed_args): - self.log.warning(_('This command has been deprecated. ' - 'Please use "volume backup restore" instead.')) - return super(RestoreBackup, self).take_action(parsed_args) - - -class ShowVolumeBackup(command.ShowOne): - _description = _("Display volume backup details") - - def get_parser(self, prog_name): - parser = super(ShowVolumeBackup, self).get_parser(prog_name) - parser.add_argument( - 'backup', - metavar='', - help=_('Backup to display (name or ID)') - ) - return parser - - def take_action(self, parsed_args): - volume_client = self.app.client_manager.volume - backup = utils.find_resource(volume_client.backups, - parsed_args.backup) - backup._info.pop('links') - return zip(*sorted(six.iteritems(backup._info))) - - -class ShowBackup(ShowVolumeBackup): - _description = _("Display backup details") - - # TODO(Huanxuan Ao): Remove this class and ``backup show`` command - # two cycles after Newton. - - # This notifies cliff to not display the help for this command - deprecated = True - - log = logging.getLogger('deprecated') - - def take_action(self, parsed_args): - self.log.warning(_('This command has been deprecated. ' - 'Please use "volume backup show" instead.')) - return super(ShowBackup, self).take_action(parsed_args) diff --git a/openstackclient/volume/v1/snapshot.py b/openstackclient/volume/v1/snapshot.py deleted file mode 100644 index e9e3894b..00000000 --- a/openstackclient/volume/v1/snapshot.py +++ /dev/null @@ -1,318 +0,0 @@ -# Copyright 2012-2013 OpenStack Foundation -# -# 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. -# - -# TODO(Huanxuan Ao): Remove this file and "snapshot create", "snapshot delete", -# "snapshot set", "snapshot show" and "snapshot unset" -# commands two cycles after Ocata. - -"""Volume v1 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 _ - - -deprecated = True -LOG_DEP = logging.getLogger('deprecated') -LOG = logging.getLogger(__name__) - - -class CreateSnapshot(command.ShowOne): - _description = _("Create new snapshot") - - def get_parser(self, prog_name): - parser = super(CreateSnapshot, self).get_parser(prog_name) - parser.add_argument( - 'volume', - metavar='', - help=_('Volume to snapshot (name or ID)'), - ) - parser.add_argument( - '--name', - metavar='', - help=_('Name of the snapshot'), - ) - parser.add_argument( - '--description', - metavar='', - help=_('Description of the snapshot'), - ) - parser.add_argument( - '--force', - dest='force', - action='store_true', - default=False, - help=_('Create a snapshot attached to an instance. ' - 'Default is False'), - ) - return parser - - def take_action(self, parsed_args): - LOG_DEP.warning(_('This command has been deprecated. ' - 'Please use "volume snapshot create" instead.')) - volume_client = self.app.client_manager.volume - volume_id = utils.find_resource(volume_client.volumes, - parsed_args.volume).id - snapshot = volume_client.volume_snapshots.create( - volume_id, - parsed_args.force, - parsed_args.name, - parsed_args.description - ) - - snapshot._info.update( - {'properties': utils.format_dict(snapshot._info.pop('metadata'))} - ) - - return zip(*sorted(six.iteritems(snapshot._info))) - - -class DeleteSnapshot(command.Command): - _description = _("Delete snapshot(s)") - - def get_parser(self, prog_name): - parser = super(DeleteSnapshot, self).get_parser(prog_name) - parser.add_argument( - 'snapshots', - metavar='', - nargs="+", - help=_('Snapshot(s) to delete (name or ID)'), - ) - return parser - - def take_action(self, parsed_args): - LOG_DEP.warning(_('This command has been deprecated. ' - 'Please use "volume snapshot delete" instead.')) - volume_client = self.app.client_manager.volume - 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): - _description = _("List snapshots") - - def get_parser(self, prog_name): - parser = super(ListSnapshot, self).get_parser(prog_name) - parser.add_argument( - '--all-projects', - action='store_true', - default=False, - help=_('Include all projects (admin only)'), - ) - parser.add_argument( - '--long', - action='store_true', - default=False, - help=_('List additional fields in output'), - ) - return parser - - def take_action(self, parsed_args): - LOG_DEP.warning(_('This command has been deprecated. ' - 'Please use "volume snapshot list" instead.')) - - def _format_volume_id(volume_id): - """Return a volume name if available - - :param volume_id: a volume ID - :rtype: either the volume ID or name - """ - - volume = volume_id - if volume_id in volume_cache.keys(): - volume = volume_cache[volume_id].display_name - return volume - - if parsed_args.long: - columns = ['ID', 'Display Name', 'Display Description', 'Status', - 'Size', 'Created At', 'Volume ID', 'Metadata'] - column_headers = copy.deepcopy(columns) - column_headers[6] = 'Volume' - column_headers[7] = 'Properties' - else: - columns = ['ID', 'Display Name', 'Display Description', 'Status', - 'Size'] - column_headers = copy.deepcopy(columns) - - # Always update Name and Description - column_headers[1] = 'Name' - column_headers[2] = 'Description' - - # Cache the volume list - volume_cache = {} - try: - for s in self.app.client_manager.volume.volumes.list(): - volume_cache[s.id] = s - except Exception: - # Just forget it if there's any trouble - pass - - search_opts = { - 'all_tenants': parsed_args.all_projects, - } - - data = self.app.client_manager.volume.volume_snapshots.list( - search_opts=search_opts) - return (column_headers, - (utils.get_item_properties( - s, columns, - formatters={'Metadata': utils.format_dict, - 'Volume ID': _format_volume_id}, - ) for s in data)) - - -class SetSnapshot(command.Command): - _description = _("Set snapshot properties") - - def get_parser(self, prog_name): - parser = super(SetSnapshot, self).get_parser(prog_name) - parser.add_argument( - 'snapshot', - metavar='', - help=_('Snapshot to modify (name or ID)') - ) - parser.add_argument( - '--name', - metavar='', - help=_('New snapshot name') - ) - parser.add_argument( - '--description', - metavar='', - help=_('New snapshot description') - ) - parser.add_argument( - '--property', - metavar='', - action=parseractions.KeyValueAction, - help=_('Property to add/change for this snapshot ' - '(repeat option to set multiple properties)'), - ) - return parser - - def take_action(self, parsed_args): - LOG_DEP.warning(_('This command has been deprecated. ' - 'Please use "volume snapshot set" instead.')) - volume_client = self.app.client_manager.volume - snapshot = utils.find_resource(volume_client.volume_snapshots, - parsed_args.snapshot) - - result = 0 - if parsed_args.property: - try: - volume_client.volume_snapshots.set_metadata( - snapshot.id, parsed_args.property) - except Exception as e: - LOG.error(_("Failed to set snapshot property: %s"), e) - result += 1 - - kwargs = {} - if parsed_args.name: - kwargs['display_name'] = parsed_args.name - if parsed_args.description: - kwargs['display_description'] = parsed_args.description - if kwargs: - try: - snapshot.update(**kwargs) - except Exception as e: - LOG.error(_("Failed to update snapshot display name " - "or display description: %s"), e) - result += 1 - - if result > 0: - raise exceptions.CommandError(_("One or more of the " - "set operations failed")) - - -class ShowSnapshot(command.ShowOne): - _description = _("Display snapshot details") - - def get_parser(self, prog_name): - parser = super(ShowSnapshot, self).get_parser(prog_name) - parser.add_argument( - 'snapshot', - metavar='', - help=_('Snapshot to display (name or ID)') - ) - return parser - - def take_action(self, parsed_args): - LOG_DEP.warning(_('This command has been deprecated. ' - 'Please use "volume snapshot show" instead.')) - volume_client = self.app.client_manager.volume - snapshot = utils.find_resource(volume_client.volume_snapshots, - parsed_args.snapshot) - - snapshot._info.update( - {'properties': utils.format_dict(snapshot._info.pop('metadata'))} - ) - - return zip(*sorted(six.iteritems(snapshot._info))) - - -class UnsetSnapshot(command.Command): - _description = _("Unset snapshot properties") - - def get_parser(self, prog_name): - parser = super(UnsetSnapshot, self).get_parser(prog_name) - parser.add_argument( - 'snapshot', - metavar='', - help=_('Snapshot to modify (name or ID)'), - ) - parser.add_argument( - '--property', - metavar='', - action='append', - help=_('Property to remove from snapshot ' - '(repeat option to remove multiple properties)'), - ) - return parser - - def take_action(self, parsed_args): - LOG_DEP.warning(_('This command has been deprecated. ' - 'Please use "volume snapshot unset" instead.')) - volume_client = self.app.client_manager.volume - snapshot = utils.find_resource( - volume_client.volume_snapshots, parsed_args.snapshot) - - if parsed_args.property: - volume_client.volume_snapshots.delete_metadata( - snapshot.id, - parsed_args.property, - ) diff --git a/openstackclient/volume/v1/volume_backup.py b/openstackclient/volume/v1/volume_backup.py new file mode 100644 index 00000000..9af32705 --- /dev/null +++ b/openstackclient/volume/v1/volume_backup.py @@ -0,0 +1,247 @@ +# Copyright 2012-2013 OpenStack Foundation +# +# 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. +# + +"""Volume v1 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 CreateVolumeBackup(command.ShowOne): + _description = _("Create new volume backup") + + def get_parser(self, prog_name): + parser = super(CreateVolumeBackup, self).get_parser(prog_name) + parser.add_argument( + 'volume', + metavar='', + help=_('Volume to backup (name or ID)'), + ) + parser.add_argument( + '--container', + metavar='', + required=False, + help=_('Optional backup container name'), + ) + parser.add_argument( + '--name', + metavar='', + help=_('Name of the backup'), + ) + parser.add_argument( + '--description', + metavar='', + help=_('Description of the backup'), + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + volume_id = utils.find_resource(volume_client.volumes, + parsed_args.volume).id + backup = volume_client.backups.create( + volume_id, + parsed_args.container, + parsed_args.name, + parsed_args.description + ) + + backup._info.pop('links') + return zip(*sorted(six.iteritems(backup._info))) + + +class DeleteVolumeBackup(command.Command): + _description = _("Delete volume backup(s)") + + def get_parser(self, prog_name): + parser = super(DeleteVolumeBackup, self).get_parser(prog_name) + parser.add_argument( + 'backups', + metavar='', + nargs="+", + help=_('Backup(s) to delete (name or ID)'), + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + 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) + 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 ListVolumeBackup(command.Lister): + _description = _("List volume backups") + + def get_parser(self, prog_name): + parser = super(ListVolumeBackup, self).get_parser(prog_name) + parser.add_argument( + '--long', + action='store_true', + default=False, + help=_('List additional fields in output'), + ) + parser.add_argument( + "--name", + metavar="", + help=_("Filters results by the backup name") + ) + parser.add_argument( + "--status", + metavar="", + choices=['creating', 'available', 'deleting', + 'error', 'restoring', 'error_restoring'], + help=_("Filters results by the backup status " + "('creating', 'available', 'deleting', " + "'error', 'restoring' or 'error_restoring')") + ) + parser.add_argument( + "--volume", + metavar="", + help=_("Filters results by the volume which they " + "backup (name or ID)") + ) + parser.add_argument( + '--all-projects', + action='store_true', + default=False, + help=_('Include all projects (admin only)'), + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + + def _format_volume_id(volume_id): + """Return a volume name if available + + :param volume_id: a volume ID + :rtype: either the volume ID or name + """ + + volume = volume_id + if volume_id in volume_cache.keys(): + volume = volume_cache[volume_id].display_name + return volume + + if parsed_args.long: + columns = ['ID', 'Name', 'Description', 'Status', 'Size', + 'Availability Zone', 'Volume ID', 'Container'] + column_headers = copy.deepcopy(columns) + column_headers[6] = 'Volume' + else: + columns = ['ID', 'Name', 'Description', 'Status', 'Size'] + column_headers = columns + + # Cache the volume list + volume_cache = {} + try: + for s in volume_client.volumes.list(): + volume_cache[s.id] = s + except Exception: + # Just forget it if there's any trouble + pass + + filter_volume_id = None + if parsed_args.volume: + filter_volume_id = utils.find_resource(volume_client.volumes, + parsed_args.volume).id + search_opts = { + 'name': parsed_args.name, + 'status': parsed_args.status, + 'volume_id': filter_volume_id, + 'all_tenants': parsed_args.all_projects, + } + data = volume_client.backups.list( + search_opts=search_opts, + ) + + return (column_headers, + (utils.get_item_properties( + s, columns, + formatters={'Volume ID': _format_volume_id}, + ) for s in data)) + + +class RestoreVolumeBackup(command.Command): + _description = _("Restore volume backup") + + def get_parser(self, prog_name): + parser = super(RestoreVolumeBackup, self).get_parser(prog_name) + parser.add_argument( + 'backup', + metavar='', + help=_('Backup to restore (name or ID)') + ) + parser.add_argument( + 'volume', + metavar='', + help=_('Volume to restore to (name or ID)') + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + backup = utils.find_resource(volume_client.backups, + parsed_args.backup) + destination_volume = utils.find_resource(volume_client.volumes, + parsed_args.volume) + return volume_client.restores.restore(backup.id, + destination_volume.id) + + +class ShowVolumeBackup(command.ShowOne): + _description = _("Display volume backup details") + + def get_parser(self, prog_name): + parser = super(ShowVolumeBackup, self).get_parser(prog_name) + parser.add_argument( + 'backup', + metavar='', + help=_('Backup to display (name or ID)') + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + backup = utils.find_resource(volume_client.backups, + parsed_args.backup) + backup._info.pop('links') + return zip(*sorted(six.iteritems(backup._info))) diff --git a/openstackclient/volume/v1/volume_transfer_request.py b/openstackclient/volume/v1/volume_transfer_request.py index f5b567b9..6f79658e 100644 --- a/openstackclient/volume/v1/volume_transfer_request.py +++ b/openstackclient/volume/v1/volume_transfer_request.py @@ -14,7 +14,6 @@ """Volume v1 transfer action implementations""" -import argparse import logging from osc_lib.command import command @@ -38,12 +37,6 @@ class AcceptTransferRequest(command.ShowOne): metavar="", help=_('Volume transfer request to accept (ID only)'), ) - parser.add_argument( - 'old_auth_key', - metavar="", - nargs="?", - help=argparse.SUPPRESS, - ) parser.add_argument( '--auth-key', metavar="", @@ -64,20 +57,9 @@ class AcceptTransferRequest(command.ShowOne): # move on and attempt with the user-supplied information transfer_request_id = parsed_args.transfer_request - # Remain backward-compatible for the previous command layout - # TODO(dtroyer): Remove this back-compat in 4.0 or Oct 2017 if not parsed_args.auth_key: - if parsed_args.old_auth_key: - # Move the old one into the correct place - parsed_args.auth_key = parsed_args.old_auth_key - self.log.warning(_( - 'Specifying the auth-key as a positional argument ' - 'has been deprecated. Please use the --auth-key ' - 'option in the future.' - )) - else: - msg = _("argument --auth-key is required") - raise exceptions.CommandError(msg) + msg = _("argument --auth-key is required") + raise exceptions.CommandError(msg) transfer_accept = volume_client.transfers.accept( transfer_request_id, diff --git a/openstackclient/volume/v2/backup.py b/openstackclient/volume/v2/backup.py deleted file mode 100644 index d4aec8d7..00000000 --- a/openstackclient/volume/v2/backup.py +++ /dev/null @@ -1,440 +0,0 @@ -# -# 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. -# - -"""Volume v2 Backup 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 CreateVolumeBackup(command.ShowOne): - _description = _("Create new volume backup") - - def get_parser(self, prog_name): - parser = super(CreateVolumeBackup, self).get_parser(prog_name) - parser.add_argument( - "volume", - metavar="", - help=_("Volume to backup (name or ID)") - ) - parser.add_argument( - "--name", - metavar="", - help=_("Name of the backup") - ) - parser.add_argument( - "--description", - metavar="", - help=_("Description of the backup") - ) - parser.add_argument( - "--container", - metavar="", - help=_("Optional backup container name") - ) - parser.add_argument( - "--snapshot", - metavar="", - help=_("Snapshot to backup (name or ID)") - ) - parser.add_argument( - '--force', - action='store_true', - default=False, - help=_("Allow to back up an in-use volume") - ) - parser.add_argument( - '--incremental', - action='store_true', - default=False, - help=_("Perform an incremental backup") - ) - return parser - - def take_action(self, parsed_args): - volume_client = self.app.client_manager.volume - volume_id = utils.find_resource( - volume_client.volumes, parsed_args.volume).id - snapshot_id = None - if parsed_args.snapshot: - snapshot_id = utils.find_resource( - volume_client.volume_snapshots, parsed_args.snapshot).id - backup = volume_client.backups.create( - volume_id, - container=parsed_args.container, - name=parsed_args.name, - description=parsed_args.description, - force=parsed_args.force, - incremental=parsed_args.incremental, - snapshot_id=snapshot_id, - ) - backup._info.pop("links", None) - return zip(*sorted(six.iteritems(backup._info))) - - -class CreateBackup(CreateVolumeBackup): - _description = _("Create new backup") - - # TODO(Huanxuan Ao): Remove this class and ``backup create`` command - # two cycles after Newton. - - # This notifies cliff to not display the help for this command - deprecated = True - - log = logging.getLogger('deprecated') - - def take_action(self, parsed_args): - self.log.warning(_('This command has been deprecated. ' - 'Please use "volume backup create" instead.')) - return super(CreateBackup, self).take_action(parsed_args) - - -class DeleteVolumeBackup(command.Command): - _description = _("Delete volume backup(s)") - - def get_parser(self, prog_name): - parser = super(DeleteVolumeBackup, self).get_parser(prog_name) - parser.add_argument( - "backups", - metavar="", - nargs="+", - help=_("Backup(s) to delete (name or ID)") - ) - parser.add_argument( - '--force', - action='store_true', - default=False, - help=_("Allow delete in state other than error or available") - ) - return parser - - def take_action(self, parsed_args): - volume_client = self.app.client_manager.volume - 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 DeleteBackup(DeleteVolumeBackup): - _description = _("Delete backup(s)") - - # TODO(Huanxuan Ao): Remove this class and ``backup delete`` command - # two cycles after Newton. - - # This notifies cliff to not display the help for this command - deprecated = True - - log = logging.getLogger('deprecated') - - def take_action(self, parsed_args): - self.log.warning(_('This command has been deprecated. ' - 'Please use "volume backup delete" instead.')) - return super(DeleteBackup, self).take_action(parsed_args) - - -class ListVolumeBackup(command.Lister): - _description = _("List volume backups") - - def get_parser(self, prog_name): - parser = super(ListVolumeBackup, self).get_parser(prog_name) - parser.add_argument( - "--long", - action="store_true", - default=False, - help=_("List additional fields in output") - ) - parser.add_argument( - "--name", - metavar="", - help=_("Filters results by the backup name") - ) - parser.add_argument( - "--status", - metavar="", - choices=['creating', 'available', 'deleting', - 'error', 'restoring', 'error_restoring'], - help=_("Filters results by the backup status " - "('creating', 'available', 'deleting', " - "'error', 'restoring' or 'error_restoring')") - ) - parser.add_argument( - "--volume", - metavar="", - help=_("Filters results by the volume which they " - "backup (name or ID)") - ) - parser.add_argument( - '--marker', - metavar='', - help=_('The last backup of the previous page (name or ID)'), - ) - parser.add_argument( - '--limit', - type=int, - action=parseractions.NonNegativeAction, - metavar='', - help=_('Maximum number of backups to display'), - ) - parser.add_argument( - '--all-projects', - action='store_true', - default=False, - help=_('Include all projects (admin only)'), - ) - return parser - - def take_action(self, parsed_args): - volume_client = self.app.client_manager.volume - - def _format_volume_id(volume_id): - """Return a volume name if available - - :param volume_id: a volume ID - :rtype: either the volume ID or name - """ - - volume = volume_id - if volume_id in volume_cache.keys(): - volume = volume_cache[volume_id].name - return volume - - if parsed_args.long: - columns = ['ID', 'Name', 'Description', 'Status', 'Size', - 'Availability Zone', 'Volume ID', 'Container'] - column_headers = copy.deepcopy(columns) - column_headers[6] = 'Volume' - else: - columns = ['ID', 'Name', 'Description', 'Status', 'Size'] - column_headers = columns - - # Cache the volume list - volume_cache = {} - try: - for s in volume_client.volumes.list(): - volume_cache[s.id] = s - except Exception: - # Just forget it if there's any trouble - pass - - filter_volume_id = None - if parsed_args.volume: - filter_volume_id = utils.find_resource(volume_client.volumes, - parsed_args.volume).id - marker_backup_id = None - if parsed_args.marker: - marker_backup_id = utils.find_resource(volume_client.backups, - parsed_args.marker).id - search_opts = { - 'name': parsed_args.name, - 'status': parsed_args.status, - 'volume_id': filter_volume_id, - 'all_tenants': parsed_args.all_projects, - } - data = volume_client.backups.list( - search_opts=search_opts, - marker=marker_backup_id, - limit=parsed_args.limit, - ) - - return (column_headers, - (utils.get_item_properties( - s, columns, - formatters={'Volume ID': _format_volume_id}, - ) for s in data)) - - -class ListBackup(ListVolumeBackup): - _description = _("List backups") - - # TODO(Huanxuan Ao): Remove this class and ``backup list`` command - # two cycles after Newton. - - # This notifies cliff to not display the help for this command - deprecated = True - - log = logging.getLogger('deprecated') - - def take_action(self, parsed_args): - self.log.warning(_('This command has been deprecated. ' - 'Please use "volume backup list" instead.')) - return super(ListBackup, self).take_action(parsed_args) - - -class RestoreVolumeBackup(command.ShowOne): - _description = _("Restore volume backup") - - def get_parser(self, prog_name): - parser = super(RestoreVolumeBackup, self).get_parser(prog_name) - parser.add_argument( - "backup", - metavar="", - help=_("Backup to restore (name or ID)") - ) - parser.add_argument( - "volume", - metavar="", - help=_("Volume to restore to (name or ID)") - ) - return parser - - def take_action(self, parsed_args): - volume_client = self.app.client_manager.volume - backup = utils.find_resource(volume_client.backups, parsed_args.backup) - destination_volume = utils.find_resource(volume_client.volumes, - parsed_args.volume) - backup = volume_client.restores.restore(backup.id, - destination_volume.id) - return zip(*sorted(six.iteritems(backup._info))) - - -class RestoreBackup(RestoreVolumeBackup): - _description = _("Restore backup") - - # TODO(Huanxuan Ao): Remove this class and ``backup restore`` command - # two cycles after Newton. - - # This notifies cliff to not display the help for this command - deprecated = True - - log = logging.getLogger('deprecated') - - def take_action(self, parsed_args): - self.log.warning(_('This command has been deprecated. ' - 'Please use "volume backup restore" instead.')) - return super(RestoreBackup, self).take_action(parsed_args) - - -class SetVolumeBackup(command.Command): - _description = _("Set volume backup properties") - - def get_parser(self, prog_name): - parser = super(SetVolumeBackup, self).get_parser(prog_name) - parser.add_argument( - "backup", - metavar="", - help=_("Backup to modify (name or ID)") - ) - parser.add_argument( - '--name', - metavar='', - help=_('New backup name') - ) - parser.add_argument( - '--description', - metavar='', - help=_('New backup description') - ) - parser.add_argument( - '--state', - metavar='', - choices=['available', 'error'], - help=_('New backup state ("available" or "error") (admin only) ' - '(This option simply changes the state of the backup ' - 'in the database with no regard to actual status, ' - 'exercise caution when using)'), - ) - return parser - - def take_action(self, parsed_args): - volume_client = self.app.client_manager.volume - backup = utils.find_resource(volume_client.backups, - parsed_args.backup) - result = 0 - if parsed_args.state: - try: - volume_client.backups.reset_state( - backup.id, parsed_args.state) - except Exception as e: - LOG.error(_("Failed to set backup state: %s"), e) - result += 1 - - kwargs = {} - if parsed_args.name: - kwargs['name'] = parsed_args.name - if parsed_args.description: - kwargs['description'] = parsed_args.description - if kwargs: - try: - volume_client.backups.update(backup.id, **kwargs) - except Exception as e: - LOG.error(_("Failed to update backup name " - "or description: %s"), e) - result += 1 - - if result > 0: - raise exceptions.CommandError(_("One or more of the " - "set operations failed")) - - -class ShowVolumeBackup(command.ShowOne): - _description = _("Display volume backup details") - - def get_parser(self, prog_name): - parser = super(ShowVolumeBackup, self).get_parser(prog_name) - parser.add_argument( - "backup", - metavar="", - help=_("Backup to display (name or ID)") - ) - return parser - - def take_action(self, parsed_args): - volume_client = self.app.client_manager.volume - backup = utils.find_resource(volume_client.backups, - parsed_args.backup) - backup._info.pop("links", None) - return zip(*sorted(six.iteritems(backup._info))) - - -class ShowBackup(ShowVolumeBackup): - _description = _("Display backup details") - - # TODO(Huanxuan Ao): Remove this class and ``backup show`` command - # two cycles after Newton. - - # This notifies cliff to not display the help for this command - deprecated = True - - log = logging.getLogger('deprecated') - - def take_action(self, parsed_args): - self.log.warning(_('This command has been deprecated. ' - 'Please use "volume backup show" instead.')) - return super(ShowBackup, self).take_action(parsed_args) diff --git a/openstackclient/volume/v2/snapshot.py b/openstackclient/volume/v2/snapshot.py deleted file mode 100644 index 82b31033..00000000 --- a/openstackclient/volume/v2/snapshot.py +++ /dev/null @@ -1,351 +0,0 @@ -# -# 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. -# - -# TODO(Huanxuan Ao): Remove this file and "snapshot create", "snapshot delete", -# "snapshot set", "snapshot show" and "snapshot unset" -# commands two cycles after Ocata. - -"""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 _ - - -deprecated = True -LOG_DEP = logging.getLogger('deprecated') -LOG = logging.getLogger(__name__) - - -class CreateSnapshot(command.ShowOne): - _description = _("Create new snapshot") - - def get_parser(self, prog_name): - parser = super(CreateSnapshot, self).get_parser(prog_name) - parser.add_argument( - "volume", - metavar="", - help=_("Volume to snapshot (name or ID)") - ) - parser.add_argument( - "--name", - metavar="", - help=_("Name of the snapshot") - ) - parser.add_argument( - "--description", - metavar="", - help=_("Description of the snapshot") - ) - parser.add_argument( - "--force", - action="store_true", - default=False, - help=_("Create a snapshot attached to an instance. " - "Default is False") - ) - parser.add_argument( - "--property", - metavar="", - action=parseractions.KeyValueAction, - help=_("Set a property to this snapshot " - "(repeat option to set multiple properties)"), - ) - return parser - - def take_action(self, parsed_args): - LOG_DEP.warning(_('This command has been deprecated. ' - 'Please use "volume snapshot create" instead.')) - volume_client = self.app.client_manager.volume - volume_id = utils.find_resource( - volume_client.volumes, parsed_args.volume).id - snapshot = volume_client.volume_snapshots.create( - volume_id, - force=parsed_args.force, - name=parsed_args.name, - description=parsed_args.description, - metadata=parsed_args.property, - ) - snapshot._info.update( - {'properties': utils.format_dict(snapshot._info.pop('metadata'))} - ) - return zip(*sorted(six.iteritems(snapshot._info))) - - -class DeleteSnapshot(command.Command): - _description = _("Delete volume snapshot(s)") - - def get_parser(self, prog_name): - parser = super(DeleteSnapshot, self).get_parser(prog_name) - parser.add_argument( - "snapshots", - metavar="", - nargs="+", - help=_("Snapshot(s) to delete (name or ID)") - ) - return parser - - def take_action(self, parsed_args): - LOG_DEP.warning(_('This command has been deprecated. ' - 'Please use "volume snapshot delete" instead.')) - volume_client = self.app.client_manager.volume - 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): - _description = _("List snapshots") - - def get_parser(self, prog_name): - parser = super(ListSnapshot, self).get_parser(prog_name) - parser.add_argument( - '--all-projects', - action='store_true', - default=False, - help=_('Include all projects (admin only)'), - ) - parser.add_argument( - '--long', - action='store_true', - default=False, - help=_('List additional fields in output'), - ) - parser.add_argument( - '--marker', - metavar='', - help=_('The last snapshot ID of the previous page'), - ) - parser.add_argument( - '--limit', - type=int, - action=parseractions.NonNegativeAction, - metavar='', - help=_('Maximum number of snapshots to display'), - ) - return parser - - def take_action(self, parsed_args): - LOG_DEP.warning(_('This command has been deprecated. ' - 'Please use "volume snapshot list" instead.')) - - def _format_volume_id(volume_id): - """Return a volume name if available - - :param volume_id: a volume ID - :rtype: either the volume ID or name - """ - - volume = volume_id - if volume_id in volume_cache.keys(): - volume = volume_cache[volume_id].name - return volume - - if parsed_args.long: - columns = ['ID', 'Name', 'Description', 'Status', - 'Size', 'Created At', 'Volume ID', 'Metadata'] - column_headers = copy.deepcopy(columns) - column_headers[6] = 'Volume' - column_headers[7] = 'Properties' - else: - columns = ['ID', 'Name', 'Description', 'Status', 'Size'] - column_headers = copy.deepcopy(columns) - - # Cache the volume list - volume_cache = {} - try: - for s in self.app.client_manager.volume.volumes.list(): - volume_cache[s.id] = s - except Exception: - # Just forget it if there's any trouble - pass - - search_opts = { - 'all_tenants': parsed_args.all_projects, - } - - data = self.app.client_manager.volume.volume_snapshots.list( - search_opts=search_opts, - marker=parsed_args.marker, - limit=parsed_args.limit, - ) - return (column_headers, - (utils.get_item_properties( - s, columns, - formatters={'Metadata': utils.format_dict, - 'Volume ID': _format_volume_id}, - ) for s in data)) - - -class SetSnapshot(command.Command): - _description = _("Set snapshot properties") - - def get_parser(self, prog_name): - parser = super(SetSnapshot, self).get_parser(prog_name) - parser.add_argument( - 'snapshot', - metavar='', - help=_('Snapshot to modify (name or ID)') - ) - parser.add_argument( - '--name', - metavar='', - help=_('New snapshot name') - ) - parser.add_argument( - '--description', - metavar='', - help=_('New snapshot description') - ) - parser.add_argument( - '--property', - metavar='', - action=parseractions.KeyValueAction, - help=_('Property to add/change for this snapshot ' - '(repeat option to set multiple properties)'), - ) - parser.add_argument( - '--state', - metavar='', - choices=['available', 'error', 'creating', 'deleting', - 'error-deleting'], - help=_('New snapshot state. ("available", "error", "creating", ' - '"deleting", or "error_deleting") (admin only) ' - '(This option simply changes the state of the snapshot ' - 'in the database with no regard to actual status, ' - 'exercise caution when using)'), - ) - return parser - - def take_action(self, parsed_args): - LOG_DEP.warning(_('This command has been deprecated. ' - 'Please use "volume snapshot set" instead.')) - volume_client = self.app.client_manager.volume - snapshot = utils.find_resource(volume_client.volume_snapshots, - parsed_args.snapshot) - - result = 0 - if parsed_args.property: - try: - volume_client.volume_snapshots.set_metadata( - snapshot.id, parsed_args.property) - except Exception as e: - LOG.error(_("Failed to set snapshot property: %s"), e) - result += 1 - - if parsed_args.state: - try: - volume_client.volume_snapshots.reset_state( - snapshot.id, parsed_args.state) - except Exception as e: - LOG.error(_("Failed to set snapshot state: %s"), e) - result += 1 - - kwargs = {} - if parsed_args.name: - kwargs['name'] = parsed_args.name - if parsed_args.description: - kwargs['description'] = parsed_args.description - if kwargs: - try: - volume_client.volume_snapshots.update( - snapshot.id, **kwargs) - except Exception as e: - LOG.error(_("Failed to update snapshot name " - "or description: %s"), e) - result += 1 - - if result > 0: - raise exceptions.CommandError(_("One or more of the " - "set operations failed")) - - -class ShowSnapshot(command.ShowOne): - _description = _("Display snapshot details") - - def get_parser(self, prog_name): - parser = super(ShowSnapshot, self).get_parser(prog_name) - parser.add_argument( - "snapshot", - metavar="", - help=_("Snapshot to display (name or ID)") - ) - return parser - - def take_action(self, parsed_args): - LOG_DEP.warning(_('This command has been deprecated. ' - 'Please use "volume snapshot show" instead.')) - volume_client = self.app.client_manager.volume - snapshot = utils.find_resource( - volume_client.volume_snapshots, parsed_args.snapshot) - snapshot._info.update( - {'properties': utils.format_dict(snapshot._info.pop('metadata'))} - ) - return zip(*sorted(six.iteritems(snapshot._info))) - - -class UnsetSnapshot(command.Command): - _description = _("Unset snapshot properties") - - def get_parser(self, prog_name): - parser = super(UnsetSnapshot, self).get_parser(prog_name) - parser.add_argument( - 'snapshot', - metavar='', - help=_('Snapshot to modify (name or ID)'), - ) - parser.add_argument( - '--property', - metavar='', - action='append', - default=[], - help=_('Property to remove from snapshot ' - '(repeat option to remove multiple properties)'), - ) - return parser - - def take_action(self, parsed_args): - LOG_DEP.warning(_('This command has been deprecated. ' - 'Please use "volume snapshot unset" instead.')) - volume_client = self.app.client_manager.volume - snapshot = utils.find_resource( - volume_client.volume_snapshots, parsed_args.snapshot) - - if parsed_args.property: - volume_client.volume_snapshots.delete_metadata( - snapshot.id, - parsed_args.property, - ) diff --git a/openstackclient/volume/v2/volume.py b/openstackclient/volume/v2/volume.py index fa587b5f..ef65d097 100644 --- a/openstackclient/volume/v2/volume.py +++ b/openstackclient/volume/v2/volume.py @@ -93,16 +93,6 @@ class CreateVolume(command.ShowOne): metavar="", help=_("Volume description"), ) - parser.add_argument( - '--user', - metavar='', - help=argparse.SUPPRESS, - ) - parser.add_argument( - '--project', - metavar='', - help=argparse.SUPPRESS, - ) parser.add_argument( "--availability-zone", metavar="", @@ -127,12 +117,6 @@ class CreateVolume(command.ShowOne): help=_("Arbitrary scheduler hint key-value pairs to help boot " "an instance (repeat option to set multiple hints)"), ) - parser.add_argument( - "--multi-attach", - action="store_true", - help=_("Allow volume to be attached more than once " - "(default to False)") - ) bootable_group = parser.add_mutually_exclusive_group() bootable_group.add_argument( "--bootable", @@ -196,26 +180,6 @@ class CreateVolume(command.ShowOne): # snapshot size. size = max(size or 0, snapshot_obj.size) - # NOTE(abishop): Cinder's volumes.create() has 'project_id' and - # 'user_id' args, but they're not wired up to anything. The only way - # to specify an alternate project or user for the volume is to use - # the identity overrides (e.g. "--os-project-id"). - # - # Now, if the project or user arg is specified then the command is - # rejected. Otherwise, Cinder would actually create a volume, but - # without the specified property. - if parsed_args.project: - raise exceptions.CommandError( - _("ERROR: --project is deprecated, please use" - " --os-project-name or --os-project-id instead.")) - if parsed_args.user: - raise exceptions.CommandError( - _("ERROR: --user is deprecated, please use" - " --os-username instead.")) - if parsed_args.multi_attach: - LOG.warning(_("'--multi-attach' option is no longer supported by " - "the block storage service.")) - volume = volume_client.volumes.create( size=size, snapshot_id=snapshot, diff --git a/openstackclient/volume/v2/volume_backup.py b/openstackclient/volume/v2/volume_backup.py new file mode 100644 index 00000000..1d2b0cde --- /dev/null +++ b/openstackclient/volume/v2/volume_backup.py @@ -0,0 +1,355 @@ +# +# 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. +# + +"""Volume v2 Backup 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 CreateVolumeBackup(command.ShowOne): + _description = _("Create new volume backup") + + def get_parser(self, prog_name): + parser = super(CreateVolumeBackup, self).get_parser(prog_name) + parser.add_argument( + "volume", + metavar="", + help=_("Volume to backup (name or ID)") + ) + parser.add_argument( + "--name", + metavar="", + help=_("Name of the backup") + ) + parser.add_argument( + "--description", + metavar="", + help=_("Description of the backup") + ) + parser.add_argument( + "--container", + metavar="", + help=_("Optional backup container name") + ) + parser.add_argument( + "--snapshot", + metavar="", + help=_("Snapshot to backup (name or ID)") + ) + parser.add_argument( + '--force', + action='store_true', + default=False, + help=_("Allow to back up an in-use volume") + ) + parser.add_argument( + '--incremental', + action='store_true', + default=False, + help=_("Perform an incremental backup") + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + volume_id = utils.find_resource( + volume_client.volumes, parsed_args.volume).id + snapshot_id = None + if parsed_args.snapshot: + snapshot_id = utils.find_resource( + volume_client.volume_snapshots, parsed_args.snapshot).id + backup = volume_client.backups.create( + volume_id, + container=parsed_args.container, + name=parsed_args.name, + description=parsed_args.description, + force=parsed_args.force, + incremental=parsed_args.incremental, + snapshot_id=snapshot_id, + ) + backup._info.pop("links", None) + return zip(*sorted(six.iteritems(backup._info))) + + +class DeleteVolumeBackup(command.Command): + _description = _("Delete volume backup(s)") + + def get_parser(self, prog_name): + parser = super(DeleteVolumeBackup, self).get_parser(prog_name) + parser.add_argument( + "backups", + metavar="", + nargs="+", + help=_("Backup(s) to delete (name or ID)") + ) + parser.add_argument( + '--force', + action='store_true', + default=False, + help=_("Allow delete in state other than error or available") + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + 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 ListVolumeBackup(command.Lister): + _description = _("List volume backups") + + def get_parser(self, prog_name): + parser = super(ListVolumeBackup, self).get_parser(prog_name) + parser.add_argument( + "--long", + action="store_true", + default=False, + help=_("List additional fields in output") + ) + parser.add_argument( + "--name", + metavar="", + help=_("Filters results by the backup name") + ) + parser.add_argument( + "--status", + metavar="", + choices=['creating', 'available', 'deleting', + 'error', 'restoring', 'error_restoring'], + help=_("Filters results by the backup status " + "('creating', 'available', 'deleting', " + "'error', 'restoring' or 'error_restoring')") + ) + parser.add_argument( + "--volume", + metavar="", + help=_("Filters results by the volume which they " + "backup (name or ID)") + ) + parser.add_argument( + '--marker', + metavar='', + help=_('The last backup of the previous page (name or ID)'), + ) + parser.add_argument( + '--limit', + type=int, + action=parseractions.NonNegativeAction, + metavar='', + help=_('Maximum number of backups to display'), + ) + parser.add_argument( + '--all-projects', + action='store_true', + default=False, + help=_('Include all projects (admin only)'), + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + + def _format_volume_id(volume_id): + """Return a volume name if available + + :param volume_id: a volume ID + :rtype: either the volume ID or name + """ + + volume = volume_id + if volume_id in volume_cache.keys(): + volume = volume_cache[volume_id].name + return volume + + if parsed_args.long: + columns = ['ID', 'Name', 'Description', 'Status', 'Size', + 'Availability Zone', 'Volume ID', 'Container'] + column_headers = copy.deepcopy(columns) + column_headers[6] = 'Volume' + else: + columns = ['ID', 'Name', 'Description', 'Status', 'Size'] + column_headers = columns + + # Cache the volume list + volume_cache = {} + try: + for s in volume_client.volumes.list(): + volume_cache[s.id] = s + except Exception: + # Just forget it if there's any trouble + pass + + filter_volume_id = None + if parsed_args.volume: + filter_volume_id = utils.find_resource(volume_client.volumes, + parsed_args.volume).id + marker_backup_id = None + if parsed_args.marker: + marker_backup_id = utils.find_resource(volume_client.backups, + parsed_args.marker).id + search_opts = { + 'name': parsed_args.name, + 'status': parsed_args.status, + 'volume_id': filter_volume_id, + 'all_tenants': parsed_args.all_projects, + } + data = volume_client.backups.list( + search_opts=search_opts, + marker=marker_backup_id, + limit=parsed_args.limit, + ) + + return (column_headers, + (utils.get_item_properties( + s, columns, + formatters={'Volume ID': _format_volume_id}, + ) for s in data)) + + +class RestoreVolumeBackup(command.ShowOne): + _description = _("Restore volume backup") + + def get_parser(self, prog_name): + parser = super(RestoreVolumeBackup, self).get_parser(prog_name) + parser.add_argument( + "backup", + metavar="", + help=_("Backup to restore (name or ID)") + ) + parser.add_argument( + "volume", + metavar="", + help=_("Volume to restore to (name or ID)") + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + backup = utils.find_resource(volume_client.backups, parsed_args.backup) + destination_volume = utils.find_resource(volume_client.volumes, + parsed_args.volume) + backup = volume_client.restores.restore(backup.id, + destination_volume.id) + return zip(*sorted(six.iteritems(backup._info))) + + +class SetVolumeBackup(command.Command): + _description = _("Set volume backup properties") + + def get_parser(self, prog_name): + parser = super(SetVolumeBackup, self).get_parser(prog_name) + parser.add_argument( + "backup", + metavar="", + help=_("Backup to modify (name or ID)") + ) + parser.add_argument( + '--name', + metavar='', + help=_('New backup name') + ) + parser.add_argument( + '--description', + metavar='', + help=_('New backup description') + ) + parser.add_argument( + '--state', + metavar='', + choices=['available', 'error'], + help=_('New backup state ("available" or "error") (admin only) ' + '(This option simply changes the state of the backup ' + 'in the database with no regard to actual status, ' + 'exercise caution when using)'), + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + backup = utils.find_resource(volume_client.backups, + parsed_args.backup) + result = 0 + if parsed_args.state: + try: + volume_client.backups.reset_state( + backup.id, parsed_args.state) + except Exception as e: + LOG.error(_("Failed to set backup state: %s"), e) + result += 1 + + kwargs = {} + if parsed_args.name: + kwargs['name'] = parsed_args.name + if parsed_args.description: + kwargs['description'] = parsed_args.description + if kwargs: + try: + volume_client.backups.update(backup.id, **kwargs) + except Exception as e: + LOG.error(_("Failed to update backup name " + "or description: %s"), e) + result += 1 + + if result > 0: + raise exceptions.CommandError(_("One or more of the " + "set operations failed")) + + +class ShowVolumeBackup(command.ShowOne): + _description = _("Display volume backup details") + + def get_parser(self, prog_name): + parser = super(ShowVolumeBackup, self).get_parser(prog_name) + parser.add_argument( + "backup", + metavar="", + help=_("Backup to display (name or ID)") + ) + return parser + + def take_action(self, parsed_args): + volume_client = self.app.client_manager.volume + backup = utils.find_resource(volume_client.backups, + parsed_args.backup) + backup._info.pop("links", None) + return zip(*sorted(six.iteritems(backup._info))) diff --git a/openstackclient/volume/v2/volume_transfer_request.py b/openstackclient/volume/v2/volume_transfer_request.py index 2f531dc8..4c4741bc 100644 --- a/openstackclient/volume/v2/volume_transfer_request.py +++ b/openstackclient/volume/v2/volume_transfer_request.py @@ -14,7 +14,6 @@ """Volume v2 transfer action implementations""" -import argparse import logging from osc_lib.command import command @@ -38,15 +37,10 @@ class AcceptTransferRequest(command.ShowOne): metavar="", help=_('Volume transfer request to accept (ID only)'), ) - parser.add_argument( - 'old_auth_key', - metavar="", - nargs="?", - help=argparse.SUPPRESS, - ) parser.add_argument( '--auth-key', metavar="", + required=True, help=_('Volume transfer request authentication key'), ) return parser @@ -64,21 +58,6 @@ class AcceptTransferRequest(command.ShowOne): # move on and attempt with the user-supplied information transfer_request_id = parsed_args.transfer_request - # Remain backward-compatible for the previous command layout - # TODO(dtroyer): Remove this back-compat in 4.0 or Oct 2017 - if not parsed_args.auth_key: - if parsed_args.old_auth_key: - # Move the old one into the correct place - parsed_args.auth_key = parsed_args.old_auth_key - self.log.warning(_( - 'Specifying the auth-key as a positional argument ' - 'has been deprecated. Please use the --auth-key ' - 'option in the future.' - )) - else: - msg = _("argument --auth-key is required") - raise exceptions.CommandError(msg) - transfer_accept = volume_client.transfers.accept( transfer_request_id, parsed_args.auth_key, -- cgit v1.2.1