1# Copyright (c) 2015 Huawei Technologies Co., Ltd. 2# Copyright (c) 2016 Stratoscale, Ltd. 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 16import mock 17from oslo_config import cfg 18import oslo_messaging as messaging 19from oslo_serialization import jsonutils 20from six.moves import http_client 21from six.moves.urllib.parse import urlencode 22import webob 23 24from cinder.common import constants 25from cinder import context 26from cinder import exception 27from cinder import objects 28from cinder import test 29from cinder.tests.unit.api import fakes 30from cinder.tests.unit import fake_constants as fake 31from cinder.tests.unit import fake_service 32 33CONF = cfg.CONF 34 35 36def app(): 37 # no auth, just let environ['cinder.context'] pass through 38 api = fakes.router.APIRouter() 39 mapper = fakes.urlmap.URLMap() 40 mapper['/v2'] = api 41 return mapper 42 43 44def volume_get(self, context, volume_id, viewable_admin_meta=False): 45 if volume_id == fake.VOLUME_ID: 46 return objects.Volume(context, id=fake.VOLUME_ID, 47 _name_id=fake.VOLUME2_ID, 48 host='fake_host', cluster_name=None) 49 raise exception.VolumeNotFound(volume_id=volume_id) 50 51 52def api_get_manageable_snapshots(*args, **kwargs): 53 """Replacement for cinder.volume.api.API.get_manageable_snapshots.""" 54 snap_id = 'ffffffff-0000-ffff-0000-ffffffffffff' 55 snaps = [ 56 {'reference': {'source-name': 'snapshot-%s' % snap_id}, 57 'size': 4, 58 'extra_info': 'qos_setting:high', 59 'safe_to_manage': False, 60 'reason_not_safe': 'snapshot in use', 61 'cinder_id': snap_id, 62 'source_reference': {'source-name': 63 'volume-00000000-ffff-0000-ffff-000000'}}, 64 {'reference': {'source-name': 'mysnap'}, 65 'size': 5, 66 'extra_info': 'qos_setting:low', 67 'safe_to_manage': True, 68 'reason_not_safe': None, 69 'cinder_id': None, 70 'source_reference': {'source-name': 'myvol'}}] 71 return snaps 72 73 74@mock.patch('cinder.volume.api.API.get', volume_get) 75class SnapshotManageTest(test.TestCase): 76 """Test cases for cinder/api/contrib/snapshot_manage.py 77 78 The API extension adds a POST /os-snapshot-manage API that is passed a 79 cinder volume id, and a driver-specific reference parameter. 80 If everything is passed correctly, 81 then the cinder.volume.api.API.manage_existing_snapshot method 82 is invoked to manage an existing storage object on the host. 83 84 In this set of test cases, we are ensuring that the code correctly parses 85 the request structure and raises the correct exceptions when things are not 86 right, and calls down into cinder.volume.api.API.manage_existing_snapshot 87 with the correct arguments. 88 """ 89 90 def setUp(self): 91 super(SnapshotManageTest, self).setUp() 92 self._admin_ctxt = context.RequestContext(fake.USER_ID, 93 fake.PROJECT_ID, 94 is_admin=True) 95 self._non_admin_ctxt = context.RequestContext(fake.USER_ID, 96 fake.PROJECT_ID, 97 is_admin=False) 98 99 def _get_resp_post(self, body): 100 """Helper to execute an os-snapshot-manage API call.""" 101 req = webob.Request.blank('/v2/%s/os-snapshot-manage' % 102 fake.PROJECT_ID) 103 req.method = 'POST' 104 req.headers['Content-Type'] = 'application/json' 105 req.environ['cinder.context'] = self._admin_ctxt 106 req.body = jsonutils.dump_as_bytes(body) 107 res = req.get_response(app()) 108 return res 109 110 @mock.patch('cinder.volume.rpcapi.VolumeAPI.manage_existing_snapshot') 111 @mock.patch('cinder.volume.api.API.create_snapshot_in_db') 112 @mock.patch('cinder.db.sqlalchemy.api.service_get') 113 def test_manage_snapshot_ok(self, mock_db, 114 mock_create_snapshot, mock_rpcapi): 115 """Test successful manage snapshot execution. 116 117 Tests for correct operation when valid arguments are passed in the 118 request body. We ensure that cinder.volume.api.API.manage_existing got 119 called with the correct arguments, and that we return the correct HTTP 120 code to the caller. 121 """ 122 mock_db.return_value = fake_service.fake_service_obj( 123 self._admin_ctxt, 124 binary=constants.VOLUME_BINARY) 125 126 body = {'snapshot': {'volume_id': fake.VOLUME_ID, 127 'ref': {'fake_key': 'fake_ref'}}} 128 129 res = self._get_resp_post(body) 130 self.assertEqual(http_client.ACCEPTED, res.status_int, res) 131 132 # Check the db.service_get was called with correct arguments. 133 mock_db.assert_called_once_with( 134 mock.ANY, None, host='fake_host', binary=constants.VOLUME_BINARY, 135 cluster_name=None) 136 137 # Check the create_snapshot_in_db was called with correct arguments. 138 self.assertEqual(1, mock_create_snapshot.call_count) 139 args = mock_create_snapshot.call_args[0] 140 named_args = mock_create_snapshot.call_args[1] 141 self.assertEqual(fake.VOLUME_ID, args[1].get('id')) 142 self.assertTrue(named_args['commit_quota']) 143 144 # Check the volume_rpcapi.manage_existing_snapshot was called with 145 # correct arguments. 146 self.assertEqual(1, mock_rpcapi.call_count) 147 args = mock_rpcapi.call_args[0] 148 self.assertEqual({u'fake_key': u'fake_ref'}, args[2]) 149 150 @mock.patch('cinder.volume.rpcapi.VolumeAPI.manage_existing_snapshot') 151 @mock.patch('cinder.volume.api.API.create_snapshot_in_db') 152 @mock.patch('cinder.objects.service.Service.get_by_id') 153 def test_manage_snapshot_ok_with_metadata_null( 154 self, mock_db, mock_create_snapshot, mock_rpcapi): 155 mock_db.return_value = fake_service.fake_service_obj( 156 self._admin_ctxt, 157 binary=constants.VOLUME_BINARY) 158 body = {'snapshot': {'volume_id': fake.VOLUME_ID, 159 'ref': {'fake_key': 'fake_ref'}, 160 'name': 'test', 161 'description': 'test', 162 'metadata': None}} 163 164 res = self._get_resp_post(body) 165 self.assertEqual(http_client.ACCEPTED, res.status_int) 166 args = mock_create_snapshot.call_args[0] 167 # 5th argument of args is metadata. 168 self.assertIsNone(args[5]) 169 170 @mock.patch('cinder.volume.rpcapi.VolumeAPI.manage_existing_snapshot') 171 @mock.patch('cinder.volume.api.API.create_snapshot_in_db') 172 @mock.patch('cinder.db.sqlalchemy.api.service_get') 173 def test_manage_snapshot_ok_ref_as_string(self, mock_db, 174 mock_create_snapshot, 175 mock_rpcapi): 176 177 mock_db.return_value = fake_service.fake_service_obj( 178 self._admin_ctxt, 179 binary=constants.VOLUME_BINARY) 180 181 body = {'snapshot': {'volume_id': fake.VOLUME_ID, 182 'ref': "string"}} 183 184 res = self._get_resp_post(body) 185 self.assertEqual(http_client.ACCEPTED, res.status_int, res) 186 187 # Check the volume_rpcapi.manage_existing_snapshot was called with 188 # correct arguments. 189 self.assertEqual(1, mock_rpcapi.call_count) 190 args = mock_rpcapi.call_args[0] 191 self.assertEqual(body['snapshot']['ref'], args[2]) 192 193 @mock.patch('cinder.objects.service.Service.is_up', 194 return_value=True, 195 new_callable=mock.PropertyMock) 196 @mock.patch('cinder.volume.rpcapi.VolumeAPI.manage_existing_snapshot') 197 @mock.patch('cinder.volume.api.API.create_snapshot_in_db') 198 @mock.patch('cinder.db.sqlalchemy.api.service_get') 199 def test_manage_snapshot_disabled(self, mock_db, mock_create_snapshot, 200 mock_rpcapi, mock_is_up): 201 """Test manage snapshot failure due to disabled service.""" 202 mock_db.return_value = fake_service.fake_service_obj(self._admin_ctxt, 203 disabled=True) 204 body = {'snapshot': {'volume_id': fake.VOLUME_ID, 'ref': { 205 'fake_key': 'fake_ref'}}} 206 res = self._get_resp_post(body) 207 self.assertEqual(http_client.BAD_REQUEST, res.status_int, res) 208 self.assertEqual(exception.ServiceUnavailable.message, 209 res.json['badRequest']['message']) 210 mock_create_snapshot.assert_not_called() 211 mock_rpcapi.assert_not_called() 212 mock_is_up.assert_not_called() 213 214 @mock.patch('cinder.objects.service.Service.is_up', return_value=False, 215 new_callable=mock.PropertyMock) 216 @mock.patch('cinder.volume.rpcapi.VolumeAPI.manage_existing_snapshot') 217 @mock.patch('cinder.volume.api.API.create_snapshot_in_db') 218 @mock.patch('cinder.db.sqlalchemy.api.service_get') 219 def test_manage_snapshot_is_down(self, mock_db, mock_create_snapshot, 220 mock_rpcapi, mock_is_up): 221 """Test manage snapshot failure due to down service.""" 222 mock_db.return_value = fake_service.fake_service_obj(self._admin_ctxt) 223 body = {'snapshot': {'volume_id': fake.VOLUME_ID, 224 'ref': {'fake_key': 'fake_ref'}}} 225 res = self._get_resp_post(body) 226 self.assertEqual(http_client.BAD_REQUEST, res.status_int, res) 227 self.assertEqual(exception.ServiceUnavailable.message, 228 res.json['badRequest']['message']) 229 mock_create_snapshot.assert_not_called() 230 mock_rpcapi.assert_not_called() 231 self.assertTrue(mock_is_up.called) 232 233 def test_manage_snapshot_missing_volume_id(self): 234 """Test correct failure when volume_id is not specified.""" 235 body = {'snapshot': {'ref': 'fake_ref'}} 236 res = self._get_resp_post(body) 237 self.assertEqual(http_client.BAD_REQUEST, res.status_int) 238 239 def test_manage_snapshot_missing_ref(self): 240 """Test correct failure when the ref is not specified.""" 241 body = {'snapshot': {'volume_id': fake.VOLUME_ID}} 242 res = self._get_resp_post(body) 243 self.assertEqual(http_client.BAD_REQUEST, res.status_int) 244 245 def test_manage_snapshot_error_body(self): 246 """Test correct failure when body is invaild.""" 247 body = {'error_snapshot': {'volume_id': fake.VOLUME_ID}} 248 res = self._get_resp_post(body) 249 self.assertEqual(http_client.BAD_REQUEST, res.status_int) 250 251 def test_manage_snapshot_error_volume_id(self): 252 """Test correct failure when volume can't be found.""" 253 body = {'snapshot': {'volume_id': 'error_volume_id', 'ref': {}}} 254 res = self._get_resp_post(body) 255 self.assertEqual(http_client.BAD_REQUEST, res.status_int) 256 257 def _get_resp_get(self, host, detailed, paging, admin=True): 258 """Helper to execute a GET os-snapshot-manage API call.""" 259 params = {'host': host} 260 if paging: 261 params.update({'marker': '1234', 'limit': 10, 262 'offset': 4, 'sort': 'reference:asc'}) 263 query_string = "?%s" % urlencode(params) 264 detail = "" 265 if detailed: 266 detail = "/detail" 267 url = "/v2/%s/os-snapshot-manage%s%s" % (fake.PROJECT_ID, detail, 268 query_string) 269 req = webob.Request.blank(url) 270 req.method = 'GET' 271 req.headers['Content-Type'] = 'application/json' 272 req.environ['cinder.context'] = (self._admin_ctxt if admin 273 else self._non_admin_ctxt) 274 res = req.get_response(app()) 275 return res 276 277 @mock.patch('cinder.volume.api.API.get_manageable_snapshots', 278 wraps=api_get_manageable_snapshots) 279 def test_get_manageable_snapshots_non_admin(self, mock_api_manageable): 280 res = self._get_resp_get('fakehost', False, False, admin=False) 281 self.assertEqual(http_client.FORBIDDEN, res.status_int) 282 self.assertEqual(False, mock_api_manageable.called) 283 res = self._get_resp_get('fakehost', True, False, admin=False) 284 self.assertEqual(http_client.FORBIDDEN, res.status_int) 285 self.assertEqual(False, mock_api_manageable.called) 286 287 @mock.patch('cinder.volume.api.API.get_manageable_snapshots', 288 wraps=api_get_manageable_snapshots) 289 def test_get_manageable_snapshots_ok(self, mock_api_manageable): 290 res = self._get_resp_get('fakehost', False, False) 291 snap_name = 'snapshot-ffffffff-0000-ffff-0000-ffffffffffff' 292 exp = {'manageable-snapshots': 293 [{'reference': {'source-name': snap_name}, 'size': 4, 294 'safe_to_manage': False, 295 'source_reference': 296 {'source-name': 'volume-00000000-ffff-0000-ffff-000000'}}, 297 {'reference': {'source-name': 'mysnap'}, 'size': 5, 298 'safe_to_manage': True, 299 'source_reference': {'source-name': 'myvol'}}]} 300 self.assertEqual(http_client.OK, res.status_int) 301 self.assertEqual(jsonutils.loads(res.body), exp) 302 mock_api_manageable.assert_called_once_with( 303 self._admin_ctxt, 'fakehost', None, limit=CONF.osapi_max_limit, 304 marker=None, offset=0, sort_dirs=['desc'], 305 sort_keys=['reference']) 306 307 @mock.patch('cinder.volume.api.API.get_manageable_snapshots', 308 side_effect=messaging.RemoteError( 309 exc_type='InvalidInput', value='marker not found: 1234')) 310 def test_get_manageable_snapshots_non_existent_marker( 311 self, mock_api_manageable): 312 res = self._get_resp_get('fakehost', detailed=False, paging=True) 313 self.assertEqual(http_client.BAD_REQUEST, res.status_int) 314 self.assertTrue(mock_api_manageable.called) 315 316 @mock.patch('cinder.volume.api.API.get_manageable_snapshots', 317 wraps=api_get_manageable_snapshots) 318 def test_get_manageable_snapshots_detailed_ok(self, mock_api_manageable): 319 res = self._get_resp_get('fakehost', True, True) 320 snap_id = 'ffffffff-0000-ffff-0000-ffffffffffff' 321 exp = {'manageable-snapshots': 322 [{'reference': {'source-name': 'snapshot-%s' % snap_id}, 323 'size': 4, 'safe_to_manage': False, 'cinder_id': snap_id, 324 'reason_not_safe': 'snapshot in use', 325 'extra_info': 'qos_setting:high', 326 'source_reference': 327 {'source-name': 'volume-00000000-ffff-0000-ffff-000000'}}, 328 {'reference': {'source-name': 'mysnap'}, 'size': 5, 329 'cinder_id': None, 'safe_to_manage': True, 330 'reason_not_safe': None, 'extra_info': 'qos_setting:low', 331 'source_reference': {'source-name': 'myvol'}}]} 332 self.assertEqual(http_client.OK, res.status_int) 333 self.assertEqual(jsonutils.loads(res.body), exp) 334 mock_api_manageable.assert_called_once_with( 335 self._admin_ctxt, 'fakehost', None, limit=10, marker='1234', 336 offset=4, sort_dirs=['asc'], sort_keys=['reference']) 337 338 @mock.patch('cinder.volume.api.API.get_manageable_snapshots', 339 side_effect=messaging.RemoteError( 340 exc_type='InvalidInput', value='marker not found: 1234')) 341 def test_get_manageable_snapshots_non_existent_marker_detailed( 342 self, mock_api_manageable): 343 res = self._get_resp_get('fakehost', detailed=True, paging=True) 344 self.assertEqual(http_client.BAD_REQUEST, res.status_int) 345 self.assertTrue(mock_api_manageable.called) 346 347 @mock.patch('cinder.objects.service.Service.is_up', return_value=True) 348 @mock.patch('cinder.db.sqlalchemy.api.service_get') 349 def test_get_manageable_snapshots_disabled(self, mock_db, mock_is_up): 350 mock_db.return_value = fake_service.fake_service_obj(self._admin_ctxt, 351 disabled=True) 352 res = self._get_resp_get('host_ok', False, True) 353 self.assertEqual(http_client.BAD_REQUEST, res.status_int, res) 354 self.assertEqual(exception.ServiceUnavailable.message, 355 res.json['badRequest']['message']) 356 mock_is_up.assert_not_called() 357 358 @mock.patch('cinder.objects.service.Service.is_up', return_value=False, 359 new_callable=mock.PropertyMock) 360 @mock.patch('cinder.db.sqlalchemy.api.service_get') 361 def test_get_manageable_snapshots_is_down(self, mock_db, mock_is_up): 362 mock_db.return_value = fake_service.fake_service_obj(self._admin_ctxt) 363 res = self._get_resp_get('host_ok', False, True) 364 self.assertEqual(http_client.BAD_REQUEST, res.status_int, res) 365 self.assertEqual(exception.ServiceUnavailable.message, 366 res.json['badRequest']['message']) 367 self.assertTrue(mock_is_up.called) 368 369 @mock.patch('cinder.volume.rpcapi.VolumeAPI.manage_existing_snapshot') 370 @mock.patch('cinder.volume.api.API.create_snapshot_in_db') 371 @mock.patch('cinder.objects.service.Service.get_by_id') 372 def test_manage_snapshot_with_null_validate( 373 self, mock_db, mock_create_snapshot, mock_rpcapi): 374 mock_db.return_value = fake_service.fake_service_obj( 375 self._admin_ctxt, 376 binary=constants.VOLUME_BINARY) 377 body = {'snapshot': {'volume_id': fake.VOLUME_ID, 378 'ref': {'fake_key': 'fake_ref'}, 379 'name': None, 380 'description': None}} 381 382 res = self._get_resp_post(body) 383 self.assertEqual(http_client.ACCEPTED, res.status_int, res) 384 self.assertIn('snapshot', jsonutils.loads(res.body)) 385