blob: cb9ff11c328937e44f9e30a342a13dc3b8d88af6 [file] [log] [blame]
wanghaoaa1f2f92013-10-10 11:30:37 +08001# vim: tabstop=4 shiftwidth=4 softtabstop=4
2
3# Copyright 2013 Huawei Technologies Co.,LTD
4# All Rights Reserved.
5#
6# Licensed under the Apache License, Version 2.0 (the "License"); you may
7# not use this file except in compliance with the License. You may obtain
8# a copy of the License at
9#
10# http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
14# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
15# License for the specific language governing permissions and limitations
16# under the License.
17
18from tempest.api.volume.base import BaseVolumeAdminTest
19from tempest.common.utils import data_utils as utils
20from tempest.test import attr
21
22
23class VolumesActionsTest(BaseVolumeAdminTest):
24 _interface = "json"
25
26 @classmethod
27 def setUpClass(cls):
28 super(VolumesActionsTest, cls).setUpClass()
29 cls.client = cls.volumes_client
30
31 # Create admin volume client
32 cls.admin_volume_client = cls.os_adm.volumes_client
33
34 # Create a test shared volume for tests
35 vol_name = utils.rand_name(cls.__name__ + '-Volume-')
36
37 resp, cls.volume = cls.client.create_volume(size=1,
38 display_name=vol_name)
39 cls.client.wait_for_volume_status(cls.volume['id'], 'available')
40
41 @classmethod
42 def tearDownClass(cls):
43 # Delete the test volume
44 cls.client.delete_volume(cls.volume['id'])
45 cls.client.wait_for_resource_deletion(cls.volume['id'])
46
47 super(VolumesActionsTest, cls).tearDownClass()
48
49 def _reset_volume_status(self, volume_id, status):
wanghao9d3d6cb2013-11-12 15:10:10 +080050 # Reset the volume status
wanghaoaa1f2f92013-10-10 11:30:37 +080051 resp, body = self.admin_volume_client.reset_volume_status(volume_id,
52 status)
53 return resp, body
54
55 def tearDown(self):
56 # Set volume's status to available after test
57 self._reset_volume_status(self.volume['id'], 'available')
58 super(VolumesActionsTest, self).tearDown()
59
wanghao9d3d6cb2013-11-12 15:10:10 +080060 def _create_temp_volume(self):
61 # Create a temp volume for force delete tests
62 vol_name = utils.rand_name('Volume')
63 resp, temp_volume = self.client.create_volume(size=1,
64 display_name=vol_name)
65 self.client.wait_for_volume_status(temp_volume['id'], 'available')
66
67 return temp_volume
68
69 def _create_reset_and_force_delete_temp_volume(self, status=None):
70 # Create volume, reset volume status, and force delete temp volume
71 temp_volume = self._create_temp_volume()
72 if status:
73 resp, body = self._reset_volume_status(temp_volume['id'], status)
74 self.assertEqual(202, resp.status)
75 resp_delete, volume_delete = self.admin_volume_client.\
76 force_delete_volume(temp_volume['id'])
77 self.assertEqual(202, resp_delete.status)
78 self.client.wait_for_resource_deletion(temp_volume['id'])
79
wanghaoaa1f2f92013-10-10 11:30:37 +080080 @attr(type='gate')
81 def test_volume_reset_status(self):
82 # test volume reset status : available->error->available
83 resp, body = self._reset_volume_status(self.volume['id'], 'error')
84 self.assertEqual(202, resp.status)
85 resp_get, volume_get = self.admin_volume_client.get_volume(
86 self.volume['id'])
87 self.assertEqual('error', volume_get['status'])
88
89 @attr(type='gate')
90 def test_volume_begin_detaching(self):
91 # test volume begin detaching : available -> detaching -> available
92 resp, body = self.client.volume_begin_detaching(self.volume['id'])
93 self.assertEqual(202, resp.status)
94 resp_get, volume_get = self.client.get_volume(self.volume['id'])
95 self.assertEqual('detaching', volume_get['status'])
96
97 @attr(type='gate')
98 def test_volume_roll_detaching(self):
99 # test volume roll detaching : detaching -> in-use -> available
100 resp, body = self.client.volume_begin_detaching(self.volume['id'])
101 self.assertEqual(202, resp.status)
102 resp, body = self.client.volume_roll_detaching(self.volume['id'])
103 self.assertEqual(202, resp.status)
104 resp_get, volume_get = self.client.get_volume(self.volume['id'])
105 self.assertEqual('in-use', volume_get['status'])
106
wanghao9d3d6cb2013-11-12 15:10:10 +0800107 def test_volume_force_delete_when_volume_is_creating(self):
108 # test force delete when status of volume is creating
109 self._create_reset_and_force_delete_temp_volume('creating')
110
111 def test_volume_force_delete_when_volume_is_attaching(self):
112 # test force delete when status of volume is attaching
113 self._create_reset_and_force_delete_temp_volume('attaching')
114
115 @attr(type='gate')
116 def test_volume_force_delete_when_volume_is_error(self):
117 # test force delete when status of volume is error
118 self._create_reset_and_force_delete_temp_volume('error')
119
wanghaoaa1f2f92013-10-10 11:30:37 +0800120
121class VolumesActionsTestXML(VolumesActionsTest):
122 _interface = "xml"