blob: 1d0d0d0266a4c8c1a84d4f5ce306d9298c0c025e [file] [log] [blame]
Gorka Eguileorcbaf22e2023-04-20 17:09:43 +02001# Copyright 2023 Red Hat
2# All Rights Reserved.
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may
5# not use this file except in compliance with the License. You may obtain
6# a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations
14# under the License.
15
16from unittest import mock
17
18from tempest.common import utils
19from tempest.common import waiters
20from tempest import config
21from tempest.lib import decorators
22from tempest.lib import exceptions
23from tempest.scenario import manager
24
25CONF = config.CONF
26
27
28class BaseAttachmentTest(manager.ScenarioTest):
29 @classmethod
30 def setup_clients(cls):
31 super().setup_clients()
32 cls.attachments_client = cls.os_primary.attachments_client_latest
33 cls.admin_volume_client = cls.os_admin.volumes_client_latest
34
35 def _call_with_fake_service_token(self, valid_token,
36 client, method_name, *args, **kwargs):
37 """Call client method with non-service service token
38
39 Add a service token header that can be a valid normal user token (which
40 won't have the service role) or an invalid token altogether.
41 """
42 original_raw_request = client.raw_request
43
44 def raw_request(url, method, headers=None, body=None, chunked=False,
45 log_req_body=None):
46 token = headers['X-Auth-Token']
47 if not valid_token:
48 token = token[:-1] + ('a' if token[-1] != 'a' else 'b')
49 headers['X-Service-Token'] = token
50 return original_raw_request(url, method, headers=headers,
51 body=body, chunked=chunked,
52 log_req_body=log_req_body)
53
54 client_method = getattr(client, method_name)
55 with mock.patch.object(client, 'raw_request', raw_request):
56 return client_method(*args, **kwargs)
57
58
59class TestServerVolumeAttachmentScenario(BaseAttachmentTest):
60
61 """Test server attachment behaviors
62
63 This tests that volume attachments to servers may not be removed directly
64 and are only allowed through the compute service (bug #2004555).
65 """
66
67 @decorators.attr(type='slow')
68 @decorators.idempotent_id('be615530-f105-437a-8afe-ce998c9535d9')
69 @utils.services('compute', 'volume', 'image', 'network')
70 def test_server_detach_rules(self):
71 """Test that various methods of detaching a volume honors the rules"""
Ghanshyam Mann51c0f9a2023-07-21 14:09:40 -050072 volume = self.create_volume(wait_until=None)
73 volume2 = self.create_volume(wait_until=None)
74
Gorka Eguileorcbaf22e2023-04-20 17:09:43 +020075 server = self.create_server(wait_until='SSHABLE')
76 servers = self.servers_client.list_servers()['servers']
77 self.assertIn(server['id'], [x['id'] for x in servers])
78
Ghanshyam Mann51c0f9a2023-07-21 14:09:40 -050079 waiters.wait_for_volume_resource_status(self.volumes_client,
80 volume['id'], 'available')
81 # The volume retrieved on creation has a non-up-to-date status.
82 # Retrieval after it becomes active ensures correct details.
83 volume = self.volumes_client.show_volume(volume['id'])['volume']
Gorka Eguileorcbaf22e2023-04-20 17:09:43 +020084
85 volume = self.nova_volume_attach(server, volume)
86 self.addCleanup(self.nova_volume_detach, server, volume)
87 att_id = volume['attachments'][0]['attachment_id']
88
89 # Test user call to detach volume is rejected
90 self.assertRaises((exceptions.Forbidden, exceptions.Conflict),
91 self.volumes_client.detach_volume, volume['id'])
92
93 # Test user call to terminate connection is rejected
94 self.assertRaises((exceptions.Forbidden, exceptions.Conflict),
95 self.volumes_client.terminate_connection,
96 volume['id'], connector={})
97
98 # Test faking of service token on call to detach, force detach,
99 # terminate_connection
100 for valid_token in (True, False):
101 valid_exceptions = [exceptions.Forbidden, exceptions.Conflict]
102 if not valid_token:
103 valid_exceptions.append(exceptions.Unauthorized)
104 self.assertRaises(
105 tuple(valid_exceptions),
106 self._call_with_fake_service_token,
107 valid_token,
108 self.volumes_client,
109 'detach_volume',
110 volume['id'])
111 self.assertRaises(
112 tuple(valid_exceptions),
113 self._call_with_fake_service_token,
114 valid_token,
115 self.volumes_client,
116 'terminate_connection',
117 volume['id'], connector={})
118
119 # Reset volume's status to error
120 self.admin_volume_client.reset_volume_status(volume['id'],
121 status='error')
122 waiters.wait_for_volume_resource_status(self.volumes_client,
123 volume['id'], 'error')
124
125 # For the cleanup, we need to reset the volume status to in-use before
126 # the other cleanup steps try to detach it.
127 self.addCleanup(waiters.wait_for_volume_resource_status,
128 self.volumes_client, volume['id'], 'in-use')
129 self.addCleanup(self.admin_volume_client.reset_volume_status,
130 volume['id'], status='in-use')
131
132 # Test user call to force detach volume is rejected
133 self.assertRaises(
134 (exceptions.Forbidden, exceptions.Conflict),
135 self.admin_volume_client.force_detach_volume,
136 volume['id'], connector=None,
137 attachment_id=att_id)
138
139 # Test trying to override detach with force and service token
140 for valid_token in (True, False):
141 valid_exceptions = [exceptions.Forbidden, exceptions.Conflict]
142 if not valid_token:
143 valid_exceptions.append(exceptions.Unauthorized)
144 self.assertRaises(
145 tuple(valid_exceptions),
146 self._call_with_fake_service_token,
147 valid_token,
148 self.admin_volume_client,
149 'force_detach_volume',
150 volume['id'], connector=None, attachment_id=att_id)
151
152 # Test user call to detach with mismatch is rejected
Ghanshyam Mann51c0f9a2023-07-21 14:09:40 -0500153 waiters.wait_for_volume_resource_status(self.volumes_client,
154 volume2['id'], 'available')
155 # The volume retrieved on creation has a non-up-to-date status.
156 # Retrieval after it becomes active ensures correct details.
157 volume2 = self.volumes_client.show_volume(volume2['id'])['volume']
158
Gorka Eguileorcbaf22e2023-04-20 17:09:43 +0200159 volume2 = self.nova_volume_attach(server, volume2)
160 att_id2 = volume2['attachments'][0]['attachment_id']
161 self.assertRaises(
162 (exceptions.Forbidden, exceptions.BadRequest),
163 self.volumes_client.detach_volume,
164 volume['id'], attachment_id=att_id2)
165
166
167class TestServerVolumeAttachScenarioOldVersion(BaseAttachmentTest):
168 volume_min_microversion = '3.27'
169 volume_max_microversion = 'latest'
170
171 @decorators.attr(type='slow')
172 @decorators.idempotent_id('6f4d2144-99f4-495c-8b0b-c6a537971418')
173 @utils.services('compute', 'volume', 'image', 'network')
174 def test_old_versions_reject(self):
175 server = self.create_server(wait_until='SSHABLE')
176 servers = self.servers_client.list_servers()['servers']
177 self.assertIn(server['id'], [x['id'] for x in servers])
178
179 volume = self.create_volume()
180
181 volume = self.nova_volume_attach(server, volume)
182 self.addCleanup(self.nova_volume_detach, server, volume)
183 att_id = volume['attachments'][0]['attachment_id']
184
185 for valid_token in (True, False):
186 valid_exceptions = [exceptions.Forbidden,
187 exceptions.Conflict]
188 if not valid_token:
189 valid_exceptions.append(exceptions.Unauthorized)
190 self.assertRaises(
191 tuple(valid_exceptions),
192 self._call_with_fake_service_token,
193 valid_token,
194 self.attachments_client,
195 'delete_attachment',
196 att_id)
197
198 self.assertRaises(
199 (exceptions.Forbidden, exceptions.Conflict),
200 self.attachments_client.delete_attachment,
201 att_id)