1# Copyright 2011 Denali Systems, Inc. 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 16import datetime 17import ddt 18import mock 19from oslo_config import cfg 20import pytz 21from six.moves import http_client 22from six.moves.urllib import parse as urllib 23import webob 24 25from cinder.api import common 26from cinder.api.v2 import snapshots 27from cinder import context 28from cinder import db 29from cinder import exception 30from cinder import objects 31from cinder.objects import fields 32from cinder.scheduler import rpcapi as scheduler_rpcapi 33from cinder import test 34from cinder.tests.unit.api import fakes 35from cinder.tests.unit.api.v2 import fakes as v2_fakes 36from cinder.tests.unit import fake_constants as fake 37from cinder.tests.unit import fake_snapshot 38from cinder.tests.unit import fake_volume 39from cinder.tests.unit import utils 40from cinder import volume 41 42 43CONF = cfg.CONF 44 45UUID = '00000000-0000-0000-0000-000000000001' 46INVALID_UUID = '00000000-0000-0000-0000-000000000002' 47 48 49def _get_default_snapshot_param(): 50 return { 51 'id': UUID, 52 'volume_id': fake.VOLUME_ID, 53 'status': fields.SnapshotStatus.AVAILABLE, 54 'volume_size': 100, 55 'created_at': None, 56 'updated_at': None, 57 'user_id': 'bcb7746c7a41472d88a1ffac89ba6a9b', 58 'project_id': '7ffe17a15c724e2aa79fc839540aec15', 59 'display_name': 'Default name', 60 'display_description': 'Default description', 61 'deleted': None, 62 'volume': {'availability_zone': 'test_zone'} 63 } 64 65 66def fake_snapshot_delete(self, context, snapshot): 67 if snapshot['id'] != UUID: 68 raise exception.SnapshotNotFound(snapshot['id']) 69 70 71def fake_snapshot_get(self, context, snapshot_id): 72 if snapshot_id != UUID: 73 raise exception.SnapshotNotFound(snapshot_id) 74 75 param = _get_default_snapshot_param() 76 return param 77 78 79def fake_snapshot_get_all(self, context, search_opts=None): 80 param = _get_default_snapshot_param() 81 return [param] 82 83 84@ddt.ddt 85class SnapshotApiTest(test.TestCase): 86 def setUp(self): 87 super(SnapshotApiTest, self).setUp() 88 self.mock_object(scheduler_rpcapi.SchedulerAPI, 'create_snapshot') 89 self.controller = snapshots.SnapshotsController() 90 self.ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True) 91 92 def test_snapshot_create(self): 93 volume = utils.create_volume(self.ctx) 94 snapshot_name = 'Snapshot Test Name' 95 snapshot_description = 'Snapshot Test Desc' 96 snapshot = { 97 "volume_id": volume.id, 98 "force": False, 99 "name": snapshot_name, 100 "description": snapshot_description 101 } 102 103 body = dict(snapshot=snapshot) 104 req = fakes.HTTPRequest.blank('/v2/snapshots') 105 resp_dict = self.controller.create(req, body=body) 106 107 self.assertIn('snapshot', resp_dict) 108 self.assertEqual(snapshot_name, resp_dict['snapshot']['name']) 109 self.assertEqual(snapshot_description, 110 resp_dict['snapshot']['description']) 111 self.assertIn('updated_at', resp_dict['snapshot']) 112 db.volume_destroy(self.ctx, volume.id) 113 114 def test_snapshot_create_with_null_validate(self): 115 116 volume = utils.create_volume(self.ctx) 117 snapshot = { 118 "volume_id": volume.id, 119 "force": False, 120 "name": None, 121 "description": None 122 } 123 124 body = dict(snapshot=snapshot) 125 req = fakes.HTTPRequest.blank('/v2/snapshots') 126 resp_dict = self.controller.create(req, body=body) 127 128 self.assertIn('snapshot', resp_dict) 129 self.assertIsNone(resp_dict['snapshot']['name']) 130 self.assertIsNone(resp_dict['snapshot']['description']) 131 db.volume_destroy(self.ctx, volume.id) 132 133 @ddt.data(True, 'y', 'true', 'yes', '1', 'on') 134 def test_snapshot_create_force(self, force_param): 135 volume = utils.create_volume(self.ctx, status='in-use') 136 snapshot_name = 'Snapshot Test Name' 137 snapshot_description = 'Snapshot Test Desc' 138 snapshot = { 139 "volume_id": volume.id, 140 "force": force_param, 141 "name": snapshot_name, 142 "description": snapshot_description 143 } 144 body = dict(snapshot=snapshot) 145 req = fakes.HTTPRequest.blank('/v2/snapshots') 146 resp_dict = self.controller.create(req, body=body) 147 148 self.assertIn('snapshot', resp_dict) 149 self.assertEqual(snapshot_name, 150 resp_dict['snapshot']['name']) 151 self.assertEqual(snapshot_description, 152 resp_dict['snapshot']['description']) 153 self.assertIn('updated_at', resp_dict['snapshot']) 154 155 db.volume_destroy(self.ctx, volume.id) 156 157 @ddt.data(False, 'n', 'false', 'No', '0', 'off') 158 def test_snapshot_create_force_failure(self, force_param): 159 volume = utils.create_volume(self.ctx, status='in-use') 160 snapshot_name = 'Snapshot Test Name' 161 snapshot_description = 'Snapshot Test Desc' 162 snapshot = { 163 "volume_id": volume.id, 164 "force": force_param, 165 "name": snapshot_name, 166 "description": snapshot_description 167 } 168 body = dict(snapshot=snapshot) 169 req = fakes.HTTPRequest.blank('/v2/snapshots') 170 self.assertRaises(exception.InvalidVolume, 171 self.controller.create, 172 req, 173 body=body) 174 175 db.volume_destroy(self.ctx, volume.id) 176 177 @ddt.data("**&&^^%%$$##@@", '-1', 2, '01', 'falSE', 0, 'trUE', 1, 178 "1 ") 179 def test_snapshot_create_invalid_force_param(self, force_param): 180 volume = utils.create_volume(self.ctx, status='available') 181 snapshot_name = 'Snapshot Test Name' 182 snapshot_description = 'Snapshot Test Desc' 183 184 snapshot = { 185 "volume_id": volume.id, 186 "force": force_param, 187 "name": snapshot_name, 188 "description": snapshot_description 189 } 190 body = dict(snapshot=snapshot) 191 req = fakes.HTTPRequest.blank('/v2/snapshots') 192 self.assertRaises(exception.ValidationError, 193 self.controller.create, 194 req, 195 body=body) 196 197 db.volume_destroy(self.ctx, volume.id) 198 199 def test_snapshot_create_without_volume_id(self): 200 snapshot_name = 'Snapshot Test Name' 201 snapshot_description = 'Snapshot Test Desc' 202 body = { 203 "snapshot": { 204 "force": True, 205 "name": snapshot_name, 206 "description": snapshot_description 207 } 208 } 209 req = fakes.HTTPRequest.blank('/v2/snapshots') 210 self.assertRaises(exception.ValidationError, 211 self.controller.create, req, body=body) 212 213 @ddt.data({"snapshot": {"description": " sample description", 214 "name": " test"}}, 215 {"snapshot": {"description": "sample description ", 216 "name": "test "}}, 217 {"snapshot": {"description": " sample description ", 218 "name": " test name "}}) 219 def test_snapshot_create_with_leading_trailing_spaces(self, body): 220 volume = utils.create_volume(self.ctx) 221 body['snapshot']['volume_id'] = volume.id 222 req = fakes.HTTPRequest.blank('/v2/snapshots') 223 resp_dict = self.controller.create(req, body=body) 224 225 self.assertEqual(body['snapshot']['display_name'].strip(), 226 resp_dict['snapshot']['name']) 227 self.assertEqual(body['snapshot']['description'].strip(), 228 resp_dict['snapshot']['description']) 229 db.volume_destroy(self.ctx, volume.id) 230 231 @mock.patch.object(volume.api.API, "update_snapshot", 232 side_effect=v2_fakes.fake_snapshot_update) 233 @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict()) 234 @mock.patch('cinder.db.volume_get') 235 @mock.patch('cinder.objects.Snapshot.get_by_id') 236 def test_snapshot_update( 237 self, snapshot_get_by_id, volume_get, 238 snapshot_metadata_get, update_snapshot): 239 snapshot = { 240 'id': UUID, 241 'volume_id': fake.VOLUME_ID, 242 'status': fields.SnapshotStatus.AVAILABLE, 243 'created_at': "2014-01-01 00:00:00", 244 'volume_size': 100, 245 'display_name': 'Default name', 246 'display_description': 'Default description', 247 'expected_attrs': ['metadata'], 248 } 249 ctx = context.RequestContext(fake.PROJECT_ID, fake.USER_ID, True) 250 snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot) 251 fake_volume_obj = fake_volume.fake_volume_obj(ctx) 252 snapshot_get_by_id.return_value = snapshot_obj 253 volume_get.return_value = fake_volume_obj 254 255 updates = { 256 "name": "Updated Test Name", 257 } 258 body = {"snapshot": updates} 259 req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % UUID) 260 res_dict = self.controller.update(req, UUID, body=body) 261 expected = { 262 'snapshot': { 263 'id': UUID, 264 'volume_id': fake.VOLUME_ID, 265 'status': fields.SnapshotStatus.AVAILABLE, 266 'size': 100, 267 'created_at': datetime.datetime(2014, 1, 1, 0, 0, 0, 268 tzinfo=pytz.utc), 269 'updated_at': None, 270 'name': u'Updated Test Name', 271 'description': u'Default description', 272 'metadata': {}, 273 } 274 } 275 self.assertEqual(expected, res_dict) 276 self.assertEqual(2, len(self.notifier.notifications)) 277 278 @mock.patch.object(volume.api.API, "update_snapshot", 279 side_effect=v2_fakes.fake_snapshot_update) 280 @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict()) 281 @mock.patch('cinder.db.volume_get') 282 @mock.patch('cinder.objects.Snapshot.get_by_id') 283 def test_snapshot_update_with_null_validate( 284 self, snapshot_get_by_id, volume_get, 285 snapshot_metadata_get, update_snapshot): 286 snapshot = { 287 'id': UUID, 288 'volume_id': fake.VOLUME_ID, 289 'status': fields.SnapshotStatus.AVAILABLE, 290 'created_at': "2014-01-01 00:00:00", 291 'volume_size': 100, 292 'name': 'Default name', 293 'description': 'Default description', 294 'expected_attrs': ['metadata'], 295 } 296 ctx = context.RequestContext(fake.PROJECT_ID, fake.USER_ID, True) 297 snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot) 298 fake_volume_obj = fake_volume.fake_volume_obj(ctx) 299 snapshot_get_by_id.return_value = snapshot_obj 300 volume_get.return_value = fake_volume_obj 301 302 updates = { 303 "name": None, 304 "description": None, 305 } 306 body = {"snapshot": updates} 307 req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % UUID) 308 res_dict = self.controller.update(req, UUID, body=body) 309 310 self.assertEqual(fields.SnapshotStatus.AVAILABLE, 311 res_dict['snapshot']['status']) 312 self.assertIsNone(res_dict['snapshot']['name']) 313 self.assertIsNone(res_dict['snapshot']['description']) 314 315 def test_snapshot_update_missing_body(self): 316 body = {} 317 req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % UUID) 318 self.assertRaises(exception.ValidationError, 319 self.controller.update, req, UUID, body=body) 320 321 def test_snapshot_update_invalid_body(self): 322 body = {'name': 'missing top level snapshot key'} 323 req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % UUID) 324 self.assertRaises(exception.ValidationError, 325 self.controller.update, req, UUID, body=body) 326 327 def test_snapshot_update_not_found(self): 328 self.mock_object(volume.api.API, "get_snapshot", fake_snapshot_get) 329 updates = { 330 "name": "Updated Test Name", 331 } 332 body = {"snapshot": updates} 333 req = fakes.HTTPRequest.blank('/v2/snapshots/not-the-uuid') 334 self.assertRaises(exception.SnapshotNotFound, self.controller.update, 335 req, 'not-the-uuid', body=body) 336 337 @mock.patch.object(volume.api.API, "update_snapshot", 338 side_effect=v2_fakes.fake_snapshot_update) 339 @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict()) 340 @mock.patch('cinder.db.volume_get') 341 @mock.patch('cinder.objects.Snapshot.get_by_id') 342 def test_snapshot_update_with_leading_trailing_spaces( 343 self, snapshot_get_by_id, volume_get, 344 snapshot_metadata_get, update_snapshot): 345 snapshot = { 346 'id': UUID, 347 'volume_id': fake.VOLUME_ID, 348 'status': fields.SnapshotStatus.AVAILABLE, 349 'created_at': "2018-01-14 00:00:00", 350 'volume_size': 100, 351 'display_name': 'Default name', 352 'display_description': 'Default description', 353 'expected_attrs': ['metadata'], 354 } 355 ctx = context.RequestContext(fake.PROJECT_ID, fake.USER_ID, True) 356 snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot) 357 fake_volume_obj = fake_volume.fake_volume_obj(ctx) 358 snapshot_get_by_id.return_value = snapshot_obj 359 volume_get.return_value = fake_volume_obj 360 361 updates = { 362 "name": " test ", 363 "description": " test " 364 } 365 body = {"snapshot": updates} 366 req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % UUID) 367 res_dict = self.controller.update(req, UUID, body=body) 368 expected = { 369 'snapshot': { 370 'id': UUID, 371 'volume_id': fake.VOLUME_ID, 372 'status': fields.SnapshotStatus.AVAILABLE, 373 'size': 100, 374 'created_at': datetime.datetime(2018, 1, 14, 0, 0, 0, 375 tzinfo=pytz.utc), 376 'updated_at': None, 377 'name': u'test', 378 'description': u'test', 379 'metadata': {}, 380 } 381 } 382 self.assertEqual(expected, res_dict) 383 self.assertEqual(2, len(self.notifier.notifications)) 384 385 @mock.patch.object(volume.api.API, "delete_snapshot", 386 side_effect=v2_fakes.fake_snapshot_update) 387 @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict()) 388 @mock.patch('cinder.objects.Volume.get_by_id') 389 @mock.patch('cinder.objects.Snapshot.get_by_id') 390 def test_snapshot_delete(self, snapshot_get_by_id, volume_get_by_id, 391 snapshot_metadata_get, delete_snapshot): 392 snapshot = { 393 'id': UUID, 394 'volume_id': fake.VOLUME_ID, 395 'status': fields.SnapshotStatus.AVAILABLE, 396 'volume_size': 100, 397 'display_name': 'Default name', 398 'display_description': 'Default description', 399 'expected_attrs': ['metadata'], 400 } 401 ctx = context.RequestContext(fake.PROJECT_ID, fake.USER_ID, True) 402 snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot) 403 fake_volume_obj = fake_volume.fake_volume_obj(ctx) 404 snapshot_get_by_id.return_value = snapshot_obj 405 volume_get_by_id.return_value = fake_volume_obj 406 407 snapshot_id = UUID 408 req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % snapshot_id) 409 resp = self.controller.delete(req, snapshot_id) 410 self.assertEqual(http_client.ACCEPTED, resp.status_int) 411 412 def test_snapshot_delete_invalid_id(self): 413 self.mock_object(volume.api.API, "delete_snapshot", 414 fake_snapshot_delete) 415 snapshot_id = INVALID_UUID 416 req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % snapshot_id) 417 self.assertRaises(exception.SnapshotNotFound, self.controller.delete, 418 req, snapshot_id) 419 420 @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict()) 421 @mock.patch('cinder.objects.Volume.get_by_id') 422 @mock.patch('cinder.objects.Snapshot.get_by_id') 423 def test_snapshot_show(self, snapshot_get_by_id, volume_get_by_id, 424 snapshot_metadata_get): 425 snapshot = { 426 'id': UUID, 427 'volume_id': fake.VOLUME_ID, 428 'status': fields.SnapshotStatus.AVAILABLE, 429 'volume_size': 100, 430 'display_name': 'Default name', 431 'display_description': 'Default description', 432 'expected_attrs': ['metadata'], 433 } 434 ctx = context.RequestContext(fake.PROJECT_ID, fake.USER_ID, True) 435 snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot) 436 fake_volume_obj = fake_volume.fake_volume_obj(ctx) 437 snapshot_get_by_id.return_value = snapshot_obj 438 volume_get_by_id.return_value = fake_volume_obj 439 req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % UUID) 440 resp_dict = self.controller.show(req, UUID) 441 442 self.assertIn('snapshot', resp_dict) 443 self.assertEqual(UUID, resp_dict['snapshot']['id']) 444 self.assertIn('updated_at', resp_dict['snapshot']) 445 446 def test_snapshot_show_invalid_id(self): 447 snapshot_id = INVALID_UUID 448 req = fakes.HTTPRequest.blank('/v2/snapshots/%s' % snapshot_id) 449 self.assertRaises(exception.SnapshotNotFound, 450 self.controller.show, req, snapshot_id) 451 452 @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict()) 453 @mock.patch('cinder.objects.Volume.get_by_id') 454 @mock.patch('cinder.objects.Snapshot.get_by_id') 455 @mock.patch('cinder.volume.api.API.get_all_snapshots') 456 def test_snapshot_detail(self, get_all_snapshots, snapshot_get_by_id, 457 volume_get_by_id, snapshot_metadata_get): 458 snapshot = { 459 'id': UUID, 460 'volume_id': fake.VOLUME_ID, 461 'status': fields.SnapshotStatus.AVAILABLE, 462 'volume_size': 100, 463 'display_name': 'Default name', 464 'display_description': 'Default description', 465 'expected_attrs': ['metadata'] 466 } 467 ctx = context.RequestContext(fake.PROJECT_ID, fake.USER_ID, True) 468 snapshot_obj = fake_snapshot.fake_snapshot_obj(ctx, **snapshot) 469 fake_volume_obj = fake_volume.fake_volume_obj(ctx) 470 snapshot_get_by_id.return_value = snapshot_obj 471 volume_get_by_id.return_value = fake_volume_obj 472 snapshots = objects.SnapshotList(objects=[snapshot_obj]) 473 get_all_snapshots.return_value = snapshots 474 475 req = fakes.HTTPRequest.blank('/v2/snapshots/detail') 476 resp_dict = self.controller.detail(req) 477 478 self.assertIn('snapshots', resp_dict) 479 resp_snapshots = resp_dict['snapshots'] 480 self.assertEqual(1, len(resp_snapshots)) 481 self.assertIn('updated_at', resp_snapshots[0]) 482 483 resp_snapshot = resp_snapshots.pop() 484 self.assertEqual(UUID, resp_snapshot['id']) 485 486 @mock.patch.object(db, 'snapshot_get_all_by_project', 487 v2_fakes.fake_snapshot_get_all_by_project) 488 @mock.patch.object(db, 'snapshot_get_all', 489 v2_fakes.fake_snapshot_get_all) 490 @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict()) 491 def test_admin_list_snapshots_limited_to_project(self, 492 snapshot_metadata_get): 493 req = fakes.HTTPRequest.blank('/v2/%s/snapshots' % fake.PROJECT_ID, 494 use_admin_context=True) 495 res = self.controller.index(req) 496 497 self.assertIn('snapshots', res) 498 self.assertEqual(1, len(res['snapshots'])) 499 500 @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict()) 501 def test_list_snapshots_with_limit_and_offset(self, 502 snapshot_metadata_get): 503 def list_snapshots_with_limit_and_offset(snaps, is_admin): 504 req = fakes.HTTPRequest.blank('/v2/%s/snapshots?limit=1' 505 '&offset=1' % fake.PROJECT_ID, 506 use_admin_context=is_admin) 507 res = self.controller.index(req) 508 509 self.assertIn('snapshots', res) 510 self.assertEqual(1, len(res['snapshots'])) 511 self.assertEqual(snaps[1].id, res['snapshots'][0]['id']) 512 self.assertIn('updated_at', res['snapshots'][0]) 513 514 # Test that we get an empty list with an offset greater than the 515 # number of items 516 req = fakes.HTTPRequest.blank('/v2/snapshots?limit=1&offset=3') 517 self.assertEqual({'snapshots': []}, self.controller.index(req)) 518 519 volume, snaps = self._create_db_snapshots(3) 520 # admin case 521 list_snapshots_with_limit_and_offset(snaps, is_admin=True) 522 # non-admin case 523 list_snapshots_with_limit_and_offset(snaps, is_admin=False) 524 525 @mock.patch.object(db, 'snapshot_get_all_by_project') 526 @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict()) 527 def test_list_snpashots_with_wrong_limit_and_offset(self, 528 mock_metadata_get, 529 mock_snapshot_get_all): 530 """Test list with negative and non numeric limit and offset.""" 531 mock_snapshot_get_all.return_value = [] 532 533 # Negative limit 534 req = fakes.HTTPRequest.blank('/v2/snapshots?limit=-1&offset=1') 535 self.assertRaises(webob.exc.HTTPBadRequest, 536 self.controller.index, 537 req) 538 539 # Non numeric limit 540 req = fakes.HTTPRequest.blank('/v2/snapshots?limit=a&offset=1') 541 self.assertRaises(webob.exc.HTTPBadRequest, 542 self.controller.index, 543 req) 544 545 # Negative offset 546 req = fakes.HTTPRequest.blank('/v2/snapshots?limit=1&offset=-1') 547 self.assertRaises(webob.exc.HTTPBadRequest, 548 self.controller.index, 549 req) 550 551 # Non numeric offset 552 req = fakes.HTTPRequest.blank('/v2/snapshots?limit=1&offset=a') 553 self.assertRaises(webob.exc.HTTPBadRequest, 554 self.controller.index, 555 req) 556 557 # Test that we get an exception HTTPBadRequest(400) with an offset 558 # greater than the maximum offset value. 559 url = '/v2/snapshots?limit=1&offset=323245324356534235' 560 req = fakes.HTTPRequest.blank(url) 561 self.assertRaises(webob.exc.HTTPBadRequest, 562 self.controller.index, req) 563 564 def _assert_list_next(self, expected_query=None, project=fake.PROJECT_ID, 565 **kwargs): 566 """Check a page of snapshots list.""" 567 # Since we are accessing v2 api directly we don't need to specify 568 # v2 in the request path, if we did, we'd get /v2/v2 links back 569 request_path = '/v2/%s/snapshots' % project 570 expected_path = request_path 571 572 # Construct the query if there are kwargs 573 if kwargs: 574 request_str = request_path + '?' + urllib.urlencode(kwargs) 575 else: 576 request_str = request_path 577 578 # Make the request 579 req = fakes.HTTPRequest.blank(request_str) 580 res = self.controller.index(req) 581 582 # We only expect to have a next link if there is an actual expected 583 # query. 584 if expected_query: 585 # We must have the links 586 self.assertIn('snapshots_links', res) 587 links = res['snapshots_links'] 588 589 # Must be a list of links, even if we only get 1 back 590 self.assertIsInstance(links, list) 591 next_link = links[0] 592 593 # rel entry must be next 594 self.assertIn('rel', next_link) 595 self.assertIn('next', next_link['rel']) 596 597 # href entry must have the right path 598 self.assertIn('href', next_link) 599 href_parts = urllib.urlparse(next_link['href']) 600 self.assertEqual(expected_path, href_parts.path) 601 602 # And the query from the next link must match what we were 603 # expecting 604 params = urllib.parse_qs(href_parts.query) 605 self.assertDictEqual(expected_query, params) 606 607 # Make sure we don't have links if we were not expecting them 608 else: 609 self.assertNotIn('snapshots_links', res) 610 611 def _create_db_snapshots(self, num_snaps): 612 volume = utils.create_volume(self.ctx) 613 snaps = [utils.create_snapshot(self.ctx, 614 volume.id, 615 display_name='snap' + str(i)) 616 for i in range(num_snaps)] 617 618 self.addCleanup(db.volume_destroy, self.ctx, volume.id) 619 for snap in snaps: 620 self.addCleanup(db.snapshot_destroy, self.ctx, snap.id) 621 622 snaps.reverse() 623 return volume, snaps 624 625 def test_list_snapshots_next_link_default_limit(self): 626 """Test that snapshot list pagination is limited by osapi_max_limit.""" 627 volume, snaps = self._create_db_snapshots(3) 628 629 # NOTE(geguileo): Since cinder.api.common.limited has already been 630 # imported his argument max_limit already has a default value of 1000 631 # so it doesn't matter that we change it to 2. That's why we need to 632 # mock it and send it current value. We still need to set the default 633 # value because other sections of the code use it, for example 634 # _get_collection_links 635 CONF.set_default('osapi_max_limit', 2) 636 637 def get_pagination_params(params, max_limit=CONF.osapi_max_limit, 638 original_call=common.get_pagination_params): 639 return original_call(params, max_limit) 640 641 def _get_limit_param(params, max_limit=CONF.osapi_max_limit, 642 original_call=common._get_limit_param): 643 return original_call(params, max_limit) 644 645 with mock.patch.object(common, 'get_pagination_params', 646 get_pagination_params), \ 647 mock.patch.object(common, '_get_limit_param', 648 _get_limit_param): 649 # The link from the first page should link to the second 650 self._assert_list_next({'marker': [snaps[1].id]}) 651 652 # Second page should have no next link 653 self._assert_list_next(marker=snaps[1].id) 654 655 def test_list_snapshots_next_link_with_limit(self): 656 """Test snapshot list pagination with specific limit.""" 657 volume, snaps = self._create_db_snapshots(2) 658 659 # The link from the first page should link to the second 660 self._assert_list_next({'limit': ['1'], 'marker': [snaps[0].id]}, 661 limit=1) 662 663 # Even though there are no more elements, we should get a next element 664 # per specification. 665 expected = {'limit': ['1'], 'marker': [snaps[1].id]} 666 self._assert_list_next(expected, limit=1, marker=snaps[0].id) 667 668 # When we go beyond the number of elements there should be no more 669 # next links 670 self._assert_list_next(limit=1, marker=snaps[1].id) 671 672 @mock.patch.object(db, 'snapshot_get_all_by_project', 673 v2_fakes.fake_snapshot_get_all_by_project) 674 @mock.patch.object(db, 'snapshot_get_all', 675 v2_fakes.fake_snapshot_get_all) 676 @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict()) 677 def test_admin_list_snapshots_all_tenants(self, snapshot_metadata_get): 678 req = fakes.HTTPRequest.blank('/v2/%s/snapshots?all_tenants=1' % 679 fake.PROJECT_ID, 680 use_admin_context=True) 681 res = self.controller.index(req) 682 self.assertIn('snapshots', res) 683 self.assertEqual(3, len(res['snapshots'])) 684 685 @mock.patch.object(db, 'snapshot_get_all') 686 @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict()) 687 def test_admin_list_snapshots_by_tenant_id(self, snapshot_metadata_get, 688 snapshot_get_all): 689 def get_all(context, filters=None, marker=None, limit=None, 690 sort_keys=None, sort_dirs=None, offset=None): 691 if 'project_id' in filters and 'tenant1' in filters['project_id']: 692 return [v2_fakes.fake_snapshot(fake.VOLUME_ID, 693 tenant_id='tenant1')] 694 else: 695 return [] 696 697 snapshot_get_all.side_effect = get_all 698 699 req = fakes.HTTPRequest.blank('/v2/%s/snapshots?all_tenants=1' 700 '&project_id=tenant1' % fake.PROJECT_ID, 701 use_admin_context=True) 702 res = self.controller.index(req) 703 self.assertIn('snapshots', res) 704 self.assertEqual(1, len(res['snapshots'])) 705 706 @mock.patch.object(db, 'snapshot_get_all_by_project', 707 v2_fakes.fake_snapshot_get_all_by_project) 708 @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict()) 709 def test_all_tenants_non_admin_gets_all_tenants(self, 710 snapshot_metadata_get): 711 req = fakes.HTTPRequest.blank('/v2/%s/snapshots?all_tenants=1' % 712 fake.PROJECT_ID) 713 res = self.controller.index(req) 714 self.assertIn('snapshots', res) 715 self.assertEqual(1, len(res['snapshots'])) 716 717 @mock.patch.object(db, 'snapshot_get_all_by_project', 718 v2_fakes.fake_snapshot_get_all_by_project) 719 @mock.patch.object(db, 'snapshot_get_all', 720 v2_fakes.fake_snapshot_get_all) 721 @mock.patch('cinder.db.snapshot_metadata_get', return_value=dict()) 722 def test_non_admin_get_by_project(self, snapshot_metadata_get): 723 req = fakes.HTTPRequest.blank('/v2/%s/snapshots' % fake.PROJECT_ID) 724 res = self.controller.index(req) 725 self.assertIn('snapshots', res) 726 self.assertEqual(1, len(res['snapshots'])) 727 728 def _create_snapshot_bad_body(self, body): 729 req = fakes.HTTPRequest.blank('/v2/%s/snapshots' % fake.PROJECT_ID) 730 req.method = 'POST' 731 732 self.assertRaises(exception.ValidationError, 733 self.controller.create, req, body=body) 734 735 def test_create_no_body(self): 736 self._create_snapshot_bad_body(body=None) 737 738 def test_create_missing_snapshot(self): 739 body = {'foo': {'a': 'b'}} 740 self._create_snapshot_bad_body(body=body) 741 742 def test_create_malformed_entity(self): 743 body = {'snapshot': 'string'} 744 self._create_snapshot_bad_body(body=body) 745