# 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 openstack.cloud import meta from openstack.tests.unit import base class TestVolumeBackups(base.TestCase): def setUp(self): super(TestVolumeBackups, self).setUp() self.use_cinder() def test_search_volume_backups(self): name = 'Volume1' vol1 = {'name': name, 'availability_zone': 'az1'} vol2 = {'name': name, 'availability_zone': 'az1'} vol3 = {'name': 'Volume2', 'availability_zone': 'az2'} self.register_uris([ dict(method='GET', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', 'detail']), json={"backups": [vol1, vol2, vol3]})]) result = self.cloud.search_volume_backups( name, {'availability_zone': 'az1'}) self.assertEqual(len(result), 2) self.assertEqual( meta.obj_list_to_munch([vol1, vol2]), result) self.assert_calls() def test_get_volume_backup(self): name = 'Volume1' vol1 = {'name': name, 'availability_zone': 'az1'} vol2 = {'name': name, 'availability_zone': 'az2'} vol3 = {'name': 'Volume2', 'availability_zone': 'az1'} self.register_uris([ dict(method='GET', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', 'detail']), json={"backups": [vol1, vol2, vol3]})]) result = self.cloud.get_volume_backup( name, {'availability_zone': 'az1'}) result = meta.obj_to_munch(result) self.assertEqual( meta.obj_to_munch(vol1), result) self.assert_calls() def test_list_volume_backups(self): backup = {'id': '6ff16bdf-44d5-4bf9-b0f3-687549c76414', 'status': 'available'} search_opts = {'status': 'available'} self.register_uris([ dict(method='GET', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', 'detail'], qs_elements=['='.join(i) for i in search_opts.items()]), json={"backups": [backup]})]) result = self.cloud.list_volume_backups(True, search_opts) self.assertEqual(len(result), 1) self.assertEqual( meta.obj_list_to_munch([backup]), result) self.assert_calls() def test_delete_volume_backup_wait(self): backup_id = '6ff16bdf-44d5-4bf9-b0f3-687549c76414' backup = {'id': backup_id} self.register_uris([ dict(method='GET', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', 'detail']), json={"backups": [backup]}), dict(method='DELETE', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', backup_id])), dict(method='GET', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', 'detail']), json={"backups": [backup]}), dict(method='GET', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', 'detail']), json={"backups": []})]) self.cloud.delete_volume_backup(backup_id, False, True, 1) self.assert_calls() def test_delete_volume_backup_force(self): backup_id = '6ff16bdf-44d5-4bf9-b0f3-687549c76414' backup = {'id': backup_id} self.register_uris([ dict(method='GET', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', 'detail']), json={"backups": [backup]}), dict(method='POST', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', backup_id, 'action']), json={'os-force_delete': {}}, validate=dict(json={u'os-force_delete': None})), dict(method='GET', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', 'detail']), json={"backups": [backup]}), dict(method='GET', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', 'detail']), json={"backups": []}) ]) self.cloud.delete_volume_backup(backup_id, True, True, 1) self.assert_calls() def test_create_volume_backup(self): volume_id = '1234' backup_name = 'bak1' bak1 = { 'id': '5678', 'volume_id': volume_id, 'status': 'available', 'name': backup_name } self.register_uris([ dict(method='POST', uri=self.get_mock_url( 'volumev2', 'public', append=['backups']), json={'backup': bak1}, validate=dict(json={ 'backup': { 'name': backup_name, 'volume_id': volume_id, 'description': None, 'force': False, 'snapshot_id': None, 'incremental': False } })), dict(method='GET', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', 'detail']), json={"backups": [bak1]}), ]) self.cloud.create_volume_backup(volume_id, name=backup_name) self.assert_calls() def test_create_incremental_volume_backup(self): volume_id = '1234' backup_name = 'bak1' bak1 = { 'id': '5678', 'volume_id': volume_id, 'status': 'available', 'name': backup_name } self.register_uris([ dict(method='POST', uri=self.get_mock_url( 'volumev2', 'public', append=['backups']), json={'backup': bak1}, validate=dict(json={ 'backup': { 'name': backup_name, 'volume_id': volume_id, 'description': None, 'force': False, 'snapshot_id': None, 'incremental': True } })), dict(method='GET', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', 'detail']), json={"backups": [bak1]}), ]) self.cloud.create_volume_backup(volume_id, name=backup_name, incremental=True) self.assert_calls() def test_create_volume_backup_from_snapshot(self): volume_id = '1234' backup_name = 'bak1' snapshot_id = '5678' bak1 = { 'id': '5678', 'volume_id': volume_id, 'status': 'available', 'name': 'bak1' } self.register_uris([ dict(method='POST', uri=self.get_mock_url( 'volumev2', 'public', append=['backups']), json={'backup': bak1}, validate=dict(json={ 'backup': { 'name': backup_name, 'volume_id': volume_id, 'description': None, 'force': False, 'snapshot_id': snapshot_id, 'incremental': False } })), dict(method='GET', uri=self.get_mock_url( 'volumev2', 'public', append=['backups', 'detail']), json={"backups": [bak1]}), ]) self.cloud.create_volume_backup(volume_id, name=backup_name, snapshot_id=snapshot_id) self.assert_calls()