Add test cases for volume-transfer
Volume-transfer is involved in Havana Release in Cinder now.
We need to complete related test cases for it.
Fixes bug: 1243070
Change-Id: Ie00aed32ebed70a3f5fcb5c5b5c16a2ae8e4549c
diff --git a/tempest/api/volume/test_volume_transfers.py b/tempest/api/volume/test_volume_transfers.py
new file mode 100644
index 0000000..dacebf1
--- /dev/null
+++ b/tempest/api/volume/test_volume_transfers.py
@@ -0,0 +1,120 @@
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 OpenStack Foundation
+# All Rights Reserved.
+#
+# 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 tempest.api.volume.base import BaseVolumeTest
+from tempest import clients
+from tempest.common.utils.data_utils import rand_name
+from tempest.test import attr
+
+
+class VolumesTransfersTest(BaseVolumeTest):
+ _interface = "json"
+
+ @classmethod
+ def setUpClass(cls):
+ super(VolumesTransfersTest, cls).setUpClass()
+
+ # Add another tenant to test volume-transfer
+ if cls.config.compute.allow_tenant_isolation:
+ creds = cls.isolated_creds.get_alt_creds()
+ username, tenant_name, password = creds
+ cls.os_alt = clients.Manager(username=username,
+ password=password,
+ tenant_name=tenant_name,
+ interface=cls._interface)
+ cls.alt_tenant_id = cls.isolated_creds.get_alt_tenant()['id']
+
+ # Add admin tenant to cleanup resources
+ adm_creds = cls.isolated_creds.get_admin_creds()
+ admin_username, admin_tenant_name, admin_password = adm_creds
+ cls.os_adm = clients.Manager(username=admin_username,
+ password=admin_password,
+ tenant_name=admin_tenant_name,
+ interface=cls._interface)
+ else:
+ cls.os_alt = clients.AltManager()
+ alt_tenant_name = cls.os_alt.tenant_name
+ identity_client = cls._get_identity_admin_client()
+ _, tenants = identity_client.list_tenants()
+ cls.alt_tenant_id = [tnt['id'] for tnt in tenants
+ if tnt['name'] == alt_tenant_name][0]
+ cls.os_adm = clients.ComputeAdminManager(interface=cls._interface)
+
+ cls.client = cls.volumes_client
+ cls.alt_client = cls.os_alt.volumes_client
+ cls.adm_client = cls.os_adm.volumes_client
+
+ @attr(type='gate')
+ def test_create_get_list_accept_volume_transfer(self):
+ # Create a volume first
+ vol_name = rand_name('-Volume-')
+ _, volume = self.client.create_volume(size=1, display_name=vol_name)
+ self.addCleanup(self.adm_client.delete_volume, volume['id'])
+ self.client.wait_for_volume_status(volume['id'], 'available')
+
+ # Create a volume transfer
+ resp, transfer = self.client.create_volume_transfer(volume['id'])
+ self.assertEqual(202, resp.status)
+ transfer_id = transfer['id']
+ auth_key = transfer['auth_key']
+ self.client.wait_for_volume_status(volume['id'],
+ 'awaiting-transfer')
+
+ # Get a volume transfer
+ resp, body = self.client.get_volume_transfer(transfer_id)
+ self.assertEqual(200, resp.status)
+ self.assertEqual(volume['id'], body['volume_id'])
+
+ # List volume transfers, the result should be greater than
+ # or equal to 1
+ resp, body = self.client.list_volume_transfers()
+ self.assertEqual(200, resp.status)
+ self.assertGreaterEqual(len(body), 1)
+
+ # Accept a volume transfer by alt_tenant
+ resp, body = self.alt_client.accept_volume_transfer(transfer_id,
+ auth_key)
+ self.assertEqual(202, resp.status)
+ self.alt_client.wait_for_volume_status(volume['id'], 'available')
+
+ def test_create_list_delete_volume_transfer(self):
+ # Create a volume first
+ vol_name = rand_name('-Volume-')
+ _, volume = self.client.create_volume(size=1, display_name=vol_name)
+ self.addCleanup(self.adm_client.delete_volume, volume['id'])
+ self.client.wait_for_volume_status(volume['id'], 'available')
+
+ # Create a volume transfer
+ resp, body = self.client.create_volume_transfer(volume['id'])
+ self.assertEqual(202, resp.status)
+ transfer_id = body['id']
+ self.client.wait_for_volume_status(volume['id'],
+ 'awaiting-transfer')
+
+ # List all volume transfers, there's only one in this test
+ resp, body = self.client.list_volume_transfers()
+ self.assertEqual(200, resp.status)
+ self.assertEqual(volume['id'], body[0]['volume_id'])
+
+ # Delete a volume transfer
+ resp, body = self.client.delete_volume_transfer(transfer_id)
+ self.assertEqual(202, resp.status)
+ self.client.wait_for_volume_status(volume['id'], 'available')
+
+
+class VolumesTransfersTestXML(VolumesTransfersTest):
+ _interface = "xml"
diff --git a/tempest/services/volume/json/volumes_client.py b/tempest/services/volume/json/volumes_client.py
index 670492a..93b28a2 100644
--- a/tempest/services/volume/json/volumes_client.py
+++ b/tempest/services/volume/json/volumes_client.py
@@ -201,3 +201,48 @@
resp, body = self.post('volumes/%s/action' % volume_id, post_body,
self.headers)
return resp, body
+
+ def create_volume_transfer(self, vol_id, display_name=None):
+ """Create a volume transfer."""
+ post_body = {
+ 'volume_id': vol_id
+ }
+ if display_name:
+ post_body['name'] = display_name
+ post_body = json.dumps({'transfer': post_body})
+ resp, body = self.post('os-volume-transfer',
+ post_body,
+ self.headers)
+ body = json.loads(body)
+ return resp, body['transfer']
+
+ def get_volume_transfer(self, transfer_id):
+ """Returns the details of a volume transfer."""
+ url = "os-volume-transfer/%s" % str(transfer_id)
+ resp, body = self.get(url, self.headers)
+ body = json.loads(body)
+ return resp, body['transfer']
+
+ def list_volume_transfers(self, params=None):
+ """List all the volume transfers created."""
+ url = 'os-volume-transfer'
+ if params:
+ url += '?%s' % urllib.urlencode(params)
+ resp, body = self.get(url)
+ body = json.loads(body)
+ return resp, body['transfers']
+
+ def delete_volume_transfer(self, transfer_id):
+ """Delete a volume transfer."""
+ return self.delete("os-volume-transfer/%s" % str(transfer_id))
+
+ def accept_volume_transfer(self, transfer_id, transfer_auth_key):
+ """Accept a volume transfer."""
+ post_body = {
+ 'auth_key': transfer_auth_key,
+ }
+ url = 'os-volume-transfer/%s/accept' % transfer_id
+ post_body = json.dumps({'accept': post_body})
+ resp, body = self.post(url, post_body, self.headers)
+ body = json.loads(body)
+ return resp, body['transfer']
diff --git a/tempest/services/volume/xml/volumes_client.py b/tempest/services/volume/xml/volumes_client.py
index 0edf7f3..b1e54ed 100644
--- a/tempest/services/volume/xml/volumes_client.py
+++ b/tempest/services/volume/xml/volumes_client.py
@@ -284,3 +284,56 @@
if body:
body = xml_to_json(etree.fromstring(body))
return resp, body
+
+ def create_volume_transfer(self, vol_id, display_name=None):
+ """Create a volume transfer."""
+ post_body = Element("transfer",
+ volume_id=vol_id)
+ if display_name:
+ post_body.add_attr('name', display_name)
+ resp, body = self.post('os-volume-transfer',
+ str(Document(post_body)),
+ self.headers)
+ volume = xml_to_json(etree.fromstring(body))
+ return resp, volume
+
+ def get_volume_transfer(self, transfer_id):
+ """Returns the details of a volume transfer."""
+ url = "os-volume-transfer/%s" % str(transfer_id)
+ resp, body = self.get(url, self.headers)
+ volume = xml_to_json(etree.fromstring(body))
+ return resp, volume
+
+ def list_volume_transfers(self, params=None):
+ """List all the volume transfers created."""
+ url = 'os-volume-transfer'
+ if params:
+ url += '?%s' % urllib.urlencode(params)
+
+ resp, body = self.get(url, self.headers)
+ body = etree.fromstring(body)
+ volumes = []
+ if body is not None:
+ volumes += [self._parse_volume_transfer(vol) for vol in list(body)]
+ return resp, volumes
+
+ def _parse_volume_transfer(self, body):
+ vol = dict((attr, body.get(attr)) for attr in body.keys())
+ for child in body.getchildren():
+ tag = child.tag
+ if tag.startswith("{"):
+ tag = tag.split("}", 1)
+ vol[tag] = xml_to_json(child)
+ return vol
+
+ def delete_volume_transfer(self, transfer_id):
+ """Delete a volume transfer."""
+ return self.delete("os-volume-transfer/%s" % str(transfer_id))
+
+ def accept_volume_transfer(self, transfer_id, transfer_auth_key):
+ """Accept a volume transfer."""
+ post_body = Element("accept", auth_key=transfer_auth_key)
+ url = 'os-volume-transfer/%s/accept' % transfer_id
+ resp, body = self.post(url, str(Document(post_body)), self.headers)
+ volume = xml_to_json(etree.fromstring(body))
+ return resp, volume