Adds testcase for volume backup-export and import
This submission adds 'test_volume_backup_export_import' testcase
to verify cinder backup-export and import functionality.
Also added required support functions under
"backups_client.py" file.
Change-Id: I7cd268598487c61352f7f943940f3c77596be4b0
diff --git a/tempest/api/volume/admin/test_volumes_backup.py b/tempest/api/volume/admin/test_volumes_backup.py
index 2c545a7..8015c35 100644
--- a/tempest/api/volume/admin/test_volumes_backup.py
+++ b/tempest/api/volume/admin/test_volumes_backup.py
@@ -38,6 +38,10 @@
cls.volume = cls.create_volume()
+ def _delete_backup(self, backup_id):
+ self.backups_adm_client.delete_backup(backup_id)
+ self.backups_adm_client.wait_for_backup_deletion(backup_id)
+
@test.idempotent_id('a66eb488-8ee1-47d4-8e9f-575a095728c6')
def test_volume_backup_create_get_detailed_list_restore_delete(self):
# Create backup
@@ -74,6 +78,52 @@
self.admin_volume_client.wait_for_volume_status(
restore['volume_id'], 'available')
+ @test.idempotent_id('a99c54a1-dd80-4724-8a13-13bf58d4068d')
+ def test_volume_backup_export_import(self):
+ # Create backup
+ backup_name = data_utils.rand_name('Backup')
+ backup = self.backups_adm_client.create_backup(self.volume['id'],
+ name=backup_name)
+ self.addCleanup(self._delete_backup, backup['id'])
+ self.assertEqual(backup_name, backup['name'])
+ self.backups_adm_client.wait_for_backup_status(backup['id'],
+ 'available')
+
+ # Export Backup
+ export_backup = self.backups_adm_client.export_backup(backup['id'])
+ self.assertIn('backup_service', export_backup)
+ self.assertIn('backup_url', export_backup)
+ self.assertTrue(export_backup['backup_service'].startswith(
+ 'cinder.backup.drivers'))
+ self.assertIsNotNone(export_backup['backup_url'])
+
+ # Import Backup
+ import_backup = self.backups_adm_client.import_backup(
+ backup_service=export_backup['backup_service'],
+ backup_url=export_backup['backup_url'])
+ self.addCleanup(self._delete_backup, import_backup['id'])
+ self.assertIn("id", import_backup)
+ self.backups_adm_client.wait_for_backup_status(import_backup['id'],
+ 'available')
+
+ # Verify Import Backup
+ backups = self.backups_adm_client.list_backups(detail=True)
+ self.assertIn(import_backup['id'], [b['id'] for b in backups])
+
+ # Restore backup
+ restore = self.backups_adm_client.restore_backup(import_backup['id'])
+ self.addCleanup(self.admin_volume_client.delete_volume,
+ restore['volume_id'])
+ self.assertEqual(import_backup['id'], restore['backup_id'])
+ self.admin_volume_client.wait_for_volume_status(restore['volume_id'],
+ 'available')
+
+ # Verify if restored volume is there in volume list
+ volumes = self.admin_volume_client.list_volumes()
+ self.assertIn(restore['volume_id'], [v['id'] for v in volumes])
+ self.backups_adm_client.wait_for_backup_status(import_backup['id'],
+ 'available')
+
class VolumesBackupsV1Test(VolumesBackupsV2Test):
_api_version = 1
diff --git a/tempest/services/volume/json/backups_client.py b/tempest/services/volume/json/backups_client.py
index 0f83b8d..8d34230 100644
--- a/tempest/services/volume/json/backups_client.py
+++ b/tempest/services/volume/json/backups_client.py
@@ -17,6 +17,8 @@
from oslo_serialization import jsonutils as json
+from tempest_lib import exceptions as lib_exc
+
from tempest.common import service_client
from tempest import exceptions
@@ -75,6 +77,24 @@
self.expected_success(200, resp.status)
return service_client.ResponseBodyList(resp, body['backups'])
+ def export_backup(self, backup_id):
+ """Export backup metadata record."""
+ url = "backups/%s/export_record" % backup_id
+ resp, body = self.get(url)
+ body = json.loads(body)
+ self.expected_success(200, resp.status)
+ return service_client.ResponseBody(resp, body['backup-record'])
+
+ def import_backup(self, backup_service, backup_url):
+ """Import backup metadata record."""
+ post_body = {'backup_service': backup_service,
+ 'backup_url': backup_url}
+ post_body = json.dumps({'backup-record': post_body})
+ resp, body = self.post("backups/import_record", post_body)
+ body = json.loads(body)
+ self.expected_success(201, resp.status)
+ return service_client.ResponseBody(resp, body['backup'])
+
def wait_for_backup_status(self, backup_id, status):
"""Waits for a Backup to reach a given status."""
body = self.show_backup(backup_id)
@@ -95,6 +115,18 @@
self.build_timeout))
raise exceptions.TimeoutException(message)
+ def wait_for_backup_deletion(self, backup_id):
+ """Waits for backup deletion"""
+ start_time = int(time.time())
+ while True:
+ try:
+ self.show_backup(backup_id)
+ except lib_exc.NotFound:
+ return
+ if int(time.time()) - start_time >= self.build_timeout:
+ raise exceptions.TimeoutException
+ time.sleep(self.build_interval)
+
class BackupsClient(BaseBackupsClient):
"""Volume V1 Backups client"""