1# Copyright 2013 Nebula Inc. 2# 3# Licensed under the Apache License, Version 2.0 (the "License"); you may 4# not use this file except in compliance with the License. You may obtain 5# a copy of the License at 6# 7# http://www.apache.org/licenses/LICENSE-2.0 8# 9# Unless required by applicable law or agreed to in writing, software 10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 12# License for the specific language governing permissions and limitations 13# under the License. 14# 15 16import argparse 17from unittest import mock 18from unittest.mock import call 19 20from osc_lib.cli import format_columns 21from osc_lib import exceptions 22from osc_lib import utils 23 24from openstackclient.tests.unit.identity.v2_0 import fakes as identity_fakes 25from openstackclient.tests.unit.image.v1 import fakes as image_fakes 26from openstackclient.tests.unit import utils as tests_utils 27from openstackclient.tests.unit.volume.v1 import fakes as volume_fakes 28from openstackclient.volume.v1 import volume 29 30 31class TestVolume(volume_fakes.TestVolumev1): 32 33 def setUp(self): 34 super(TestVolume, self).setUp() 35 36 # Get a shortcut to the VolumeManager Mock 37 self.volumes_mock = self.app.client_manager.volume.volumes 38 self.volumes_mock.reset_mock() 39 40 # Get a shortcut to the TenantManager Mock 41 self.projects_mock = self.app.client_manager.identity.tenants 42 self.projects_mock.reset_mock() 43 44 # Get a shortcut to the UserManager Mock 45 self.users_mock = self.app.client_manager.identity.users 46 self.users_mock.reset_mock() 47 48 # Get a shortcut to the ImageManager Mock 49 self.images_mock = self.app.client_manager.image.images 50 self.images_mock.reset_mock() 51 52 def setup_volumes_mock(self, count): 53 volumes = volume_fakes.FakeVolume.create_volumes(count=count) 54 55 self.volumes_mock.get = volume_fakes.FakeVolume.get_volumes( 56 volumes, 57 0) 58 return volumes 59 60 61class TestVolumeCreate(TestVolume): 62 63 project = identity_fakes.FakeProject.create_one_project() 64 user = identity_fakes.FakeUser.create_one_user() 65 66 columns = ( 67 'attachments', 68 'availability_zone', 69 'bootable', 70 'created_at', 71 'display_description', 72 'id', 73 'name', 74 'properties', 75 'size', 76 'snapshot_id', 77 'status', 78 'type', 79 ) 80 81 def setUp(self): 82 super(TestVolumeCreate, self).setUp() 83 self.new_volume = volume_fakes.FakeVolume.create_one_volume() 84 self.datalist = ( 85 self.new_volume.attachments, 86 self.new_volume.availability_zone, 87 self.new_volume.bootable, 88 self.new_volume.created_at, 89 self.new_volume.display_description, 90 self.new_volume.id, 91 self.new_volume.display_name, 92 format_columns.DictColumn(self.new_volume.metadata), 93 self.new_volume.size, 94 self.new_volume.snapshot_id, 95 self.new_volume.status, 96 self.new_volume.volume_type, 97 ) 98 self.volumes_mock.create.return_value = self.new_volume 99 100 # Get the command object to test 101 self.cmd = volume.CreateVolume(self.app, None) 102 103 def test_volume_create_min_options(self): 104 arglist = [ 105 '--size', str(self.new_volume.size), 106 self.new_volume.display_name, 107 ] 108 verifylist = [ 109 ('size', self.new_volume.size), 110 ('name', self.new_volume.display_name), 111 ] 112 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 113 114 # In base command class ShowOne in cliff, abstract method take_action() 115 # returns a two-part tuple with a tuple of column names and a tuple of 116 # data to be shown. 117 columns, data = self.cmd.take_action(parsed_args) 118 119 # VolumeManager.create(size, snapshot_id=, source_volid=, 120 # display_name=, display_description=, 121 # volume_type=, user_id=, 122 # project_id=, availability_zone=, 123 # metadata=, imageRef=) 124 self.volumes_mock.create.assert_called_with( 125 self.new_volume.size, 126 None, 127 None, 128 self.new_volume.display_name, 129 None, 130 None, 131 None, 132 None, 133 None, 134 None, 135 None, 136 ) 137 self.assertEqual(self.columns, columns) 138 self.assertItemEqual(self.datalist, data) 139 140 def test_volume_create_options(self): 141 arglist = [ 142 '--size', str(self.new_volume.size), 143 '--description', self.new_volume.display_description, 144 '--type', self.new_volume.volume_type, 145 '--availability-zone', self.new_volume.availability_zone, 146 self.new_volume.display_name, 147 ] 148 verifylist = [ 149 ('size', self.new_volume.size), 150 ('description', self.new_volume.display_description), 151 ('type', self.new_volume.volume_type), 152 ('availability_zone', self.new_volume.availability_zone), 153 ('name', self.new_volume.display_name), 154 ] 155 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 156 157 # In base command class ShowOne in cliff, abstract method take_action() 158 # returns a two-part tuple with a tuple of column names and a tuple of 159 # data to be shown. 160 columns, data = self.cmd.take_action(parsed_args) 161 162 # VolumeManager.create(size, snapshot_id=, source_volid=, 163 # display_name=, display_description=, 164 # volume_type=, user_id=, 165 # project_id=, availability_zone=, 166 # metadata=, imageRef=) 167 self.volumes_mock.create.assert_called_with( 168 self.new_volume.size, 169 None, 170 None, 171 self.new_volume.display_name, 172 self.new_volume.display_description, 173 self.new_volume.volume_type, 174 None, 175 None, 176 self.new_volume.availability_zone, 177 None, 178 None, 179 ) 180 181 self.assertEqual(self.columns, columns) 182 self.assertItemEqual(self.datalist, data) 183 184 def test_volume_create_user_project_id(self): 185 # Return a project 186 self.projects_mock.get.return_value = self.project 187 # Return a user 188 self.users_mock.get.return_value = self.user 189 190 arglist = [ 191 '--size', str(self.new_volume.size), 192 '--project', self.project.id, 193 '--user', self.user.id, 194 self.new_volume.display_name, 195 ] 196 verifylist = [ 197 ('size', self.new_volume.size), 198 ('project', self.project.id), 199 ('user', self.user.id), 200 ('name', self.new_volume.display_name), 201 ] 202 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 203 204 # In base command class ShowOne in cliff, abstract method take_action() 205 # returns a two-part tuple with a tuple of column names and a tuple of 206 # data to be shown. 207 columns, data = self.cmd.take_action(parsed_args) 208 209 # VolumeManager.create(size, snapshot_id=, source_volid=, 210 # display_name=, display_description=, 211 # volume_type=, user_id=, 212 # project_id=, availability_zone=, 213 # metadata=, imageRef=) 214 self.volumes_mock.create.assert_called_with( 215 self.new_volume.size, 216 None, 217 None, 218 self.new_volume.display_name, 219 None, 220 None, 221 self.user.id, 222 self.project.id, 223 None, 224 None, 225 None, 226 ) 227 228 self.assertEqual(self.columns, columns) 229 self.assertItemEqual(self.datalist, data) 230 231 def test_volume_create_user_project_name(self): 232 # Return a project 233 self.projects_mock.get.return_value = self.project 234 # Return a user 235 self.users_mock.get.return_value = self.user 236 237 arglist = [ 238 '--size', str(self.new_volume.size), 239 '--project', self.project.name, 240 '--user', self.user.name, 241 self.new_volume.display_name, 242 ] 243 verifylist = [ 244 ('size', self.new_volume.size), 245 ('project', self.project.name), 246 ('user', self.user.name), 247 ('name', self.new_volume.display_name), 248 ] 249 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 250 251 # In base command class ShowOne in cliff, abstract method take_action() 252 # returns a two-part tuple with a tuple of column names and a tuple of 253 # data to be shown. 254 columns, data = self.cmd.take_action(parsed_args) 255 256 # VolumeManager.create(size, snapshot_id=, source_volid=, 257 # display_name=, display_description=, 258 # volume_type=, user_id=, 259 # project_id=, availability_zone=, 260 # metadata=, imageRef=) 261 self.volumes_mock.create.assert_called_with( 262 self.new_volume.size, 263 None, 264 None, 265 self.new_volume.display_name, 266 None, 267 None, 268 self.user.id, 269 self.project.id, 270 None, 271 None, 272 None, 273 ) 274 275 self.assertEqual(self.columns, columns) 276 self.assertItemEqual(self.datalist, data) 277 278 def test_volume_create_properties(self): 279 arglist = [ 280 '--property', 'Alpha=a', 281 '--property', 'Beta=b', 282 '--size', str(self.new_volume.size), 283 self.new_volume.display_name, 284 ] 285 verifylist = [ 286 ('property', {'Alpha': 'a', 'Beta': 'b'}), 287 ('size', self.new_volume.size), 288 ('name', self.new_volume.display_name), 289 ] 290 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 291 292 # In base command class ShowOne in cliff, abstract method take_action() 293 # returns a two-part tuple with a tuple of column names and a tuple of 294 # data to be shown. 295 columns, data = self.cmd.take_action(parsed_args) 296 297 # VolumeManager.create(size, snapshot_id=, source_volid=, 298 # display_name=, display_description=, 299 # volume_type=, user_id=, 300 # project_id=, availability_zone=, 301 # metadata=, imageRef=) 302 self.volumes_mock.create.assert_called_with( 303 self.new_volume.size, 304 None, 305 None, 306 self.new_volume.display_name, 307 None, 308 None, 309 None, 310 None, 311 None, 312 {'Alpha': 'a', 'Beta': 'b'}, 313 None, 314 ) 315 316 self.assertEqual(self.columns, columns) 317 self.assertItemEqual(self.datalist, data) 318 319 def test_volume_create_image_id(self): 320 image = image_fakes.FakeImage.create_one_image() 321 self.images_mock.get.return_value = image 322 323 arglist = [ 324 '--image', image.id, 325 '--size', str(self.new_volume.size), 326 self.new_volume.display_name, 327 ] 328 verifylist = [ 329 ('image', image.id), 330 ('size', self.new_volume.size), 331 ('name', self.new_volume.display_name), 332 ] 333 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 334 335 # In base command class ShowOne in cliff, abstract method take_action() 336 # returns a two-part tuple with a tuple of column names and a tuple of 337 # data to be shown. 338 columns, data = self.cmd.take_action(parsed_args) 339 340 # VolumeManager.create(size, snapshot_id=, source_volid=, 341 # display_name=, display_description=, 342 # volume_type=, user_id=, 343 # project_id=, availability_zone=, 344 # metadata=, imageRef=) 345 self.volumes_mock.create.assert_called_with( 346 self.new_volume.size, 347 None, 348 None, 349 self.new_volume.display_name, 350 None, 351 None, 352 None, 353 None, 354 None, 355 None, 356 image.id, 357 ) 358 359 self.assertEqual(self.columns, columns) 360 self.assertItemEqual(self.datalist, data) 361 362 def test_volume_create_image_name(self): 363 image = image_fakes.FakeImage.create_one_image() 364 self.images_mock.get.return_value = image 365 366 arglist = [ 367 '--image', image.name, 368 '--size', str(self.new_volume.size), 369 self.new_volume.display_name, 370 ] 371 verifylist = [ 372 ('image', image.name), 373 ('size', self.new_volume.size), 374 ('name', self.new_volume.display_name), 375 ] 376 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 377 378 # In base command class ShowOne in cliff, abstract method take_action() 379 # returns a two-part tuple with a tuple of column names and a tuple of 380 # data to be shown. 381 columns, data = self.cmd.take_action(parsed_args) 382 383 # VolumeManager.create(size, snapshot_id=, source_volid=, 384 # display_name=, display_description=, 385 # volume_type=, user_id=, 386 # project_id=, availability_zone=, 387 # metadata=, imageRef=) 388 self.volumes_mock.create.assert_called_with( 389 self.new_volume.size, 390 None, 391 None, 392 self.new_volume.display_name, 393 None, 394 None, 395 None, 396 None, 397 None, 398 None, 399 image.id, 400 ) 401 402 self.assertEqual(self.columns, columns) 403 self.assertItemEqual(self.datalist, data) 404 405 def test_volume_create_with_source(self): 406 self.volumes_mock.get.return_value = self.new_volume 407 arglist = [ 408 '--source', self.new_volume.id, 409 self.new_volume.display_name, 410 ] 411 verifylist = [ 412 ('source', self.new_volume.id), 413 ('name', self.new_volume.display_name), 414 ] 415 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 416 417 columns, data = self.cmd.take_action(parsed_args) 418 419 self.volumes_mock.create.assert_called_with( 420 None, 421 None, 422 self.new_volume.id, 423 self.new_volume.display_name, 424 None, 425 None, 426 None, 427 None, 428 None, 429 None, 430 None, 431 ) 432 self.assertEqual(self.columns, columns) 433 self.assertItemEqual(self.datalist, data) 434 435 def test_volume_create_with_bootable_and_readonly(self): 436 arglist = [ 437 '--bootable', 438 '--read-only', 439 '--size', str(self.new_volume.size), 440 self.new_volume.display_name, 441 ] 442 verifylist = [ 443 ('bootable', True), 444 ('non_bootable', False), 445 ('read_only', True), 446 ('read_write', False), 447 ('size', self.new_volume.size), 448 ('name', self.new_volume.display_name), 449 ] 450 451 parsed_args = self.check_parser( 452 self.cmd, arglist, verifylist) 453 454 columns, data = self.cmd.take_action(parsed_args) 455 456 self.volumes_mock.create.assert_called_with( 457 self.new_volume.size, 458 None, 459 None, 460 self.new_volume.display_name, 461 None, 462 None, 463 None, 464 None, 465 None, 466 None, 467 None, 468 ) 469 470 self.assertEqual(self.columns, columns) 471 self.assertItemEqual(self.datalist, data) 472 self.volumes_mock.set_bootable.assert_called_with( 473 self.new_volume.id, True) 474 self.volumes_mock.update_readonly_flag.assert_called_with( 475 self.new_volume.id, True) 476 477 def test_volume_create_with_nonbootable_and_readwrite(self): 478 arglist = [ 479 '--non-bootable', 480 '--read-write', 481 '--size', str(self.new_volume.size), 482 self.new_volume.display_name, 483 ] 484 verifylist = [ 485 ('bootable', False), 486 ('non_bootable', True), 487 ('read_only', False), 488 ('read_write', True), 489 ('size', self.new_volume.size), 490 ('name', self.new_volume.display_name), 491 ] 492 493 parsed_args = self.check_parser( 494 self.cmd, arglist, verifylist) 495 496 columns, data = self.cmd.take_action(parsed_args) 497 498 self.volumes_mock.create.assert_called_with( 499 self.new_volume.size, 500 None, 501 None, 502 self.new_volume.display_name, 503 None, 504 None, 505 None, 506 None, 507 None, 508 None, 509 None, 510 ) 511 512 self.assertEqual(self.columns, columns) 513 self.assertItemEqual(self.datalist, data) 514 self.volumes_mock.set_bootable.assert_called_with( 515 self.new_volume.id, False) 516 self.volumes_mock.update_readonly_flag.assert_called_with( 517 self.new_volume.id, False) 518 519 @mock.patch.object(volume.LOG, 'error') 520 def test_volume_create_with_bootable_and_readonly_fail( 521 self, mock_error): 522 523 self.volumes_mock.set_bootable.side_effect = ( 524 exceptions.CommandError()) 525 526 self.volumes_mock.update_readonly_flag.side_effect = ( 527 exceptions.CommandError()) 528 529 arglist = [ 530 '--bootable', 531 '--read-only', 532 '--size', str(self.new_volume.size), 533 self.new_volume.display_name, 534 ] 535 verifylist = [ 536 ('bootable', True), 537 ('non_bootable', False), 538 ('read_only', True), 539 ('read_write', False), 540 ('size', self.new_volume.size), 541 ('name', self.new_volume.display_name), 542 ] 543 544 parsed_args = self.check_parser( 545 self.cmd, arglist, verifylist) 546 547 columns, data = self.cmd.take_action(parsed_args) 548 549 self.volumes_mock.create.assert_called_with( 550 self.new_volume.size, 551 None, 552 None, 553 self.new_volume.display_name, 554 None, 555 None, 556 None, 557 None, 558 None, 559 None, 560 None, 561 ) 562 563 self.assertEqual(2, mock_error.call_count) 564 self.assertEqual(self.columns, columns) 565 self.assertItemEqual(self.datalist, data) 566 self.volumes_mock.set_bootable.assert_called_with( 567 self.new_volume.id, True) 568 self.volumes_mock.update_readonly_flag.assert_called_with( 569 self.new_volume.id, True) 570 571 def test_volume_create_without_size(self): 572 arglist = [ 573 self.new_volume.display_name, 574 ] 575 verifylist = [ 576 ('name', self.new_volume.display_name), 577 ] 578 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 579 580 self.assertRaises(exceptions.CommandError, self.cmd.take_action, 581 parsed_args) 582 583 def test_volume_create_with_multi_source(self): 584 arglist = [ 585 '--image', 'source_image', 586 '--source', 'source_volume', 587 '--snapshot', 'source_snapshot', 588 '--size', str(self.new_volume.size), 589 self.new_volume.display_name, 590 ] 591 verifylist = [ 592 ('image', 'source_image'), 593 ('source', 'source_volume'), 594 ('snapshot', 'source_snapshot'), 595 ('size', self.new_volume.size), 596 ('name', self.new_volume.display_name), 597 ] 598 599 self.assertRaises(tests_utils.ParserException, self.check_parser, 600 self.cmd, arglist, verifylist) 601 602 def test_volume_create_backward_compatibility(self): 603 arglist = [ 604 '-c', 'display_name', 605 '--size', str(self.new_volume.size), 606 self.new_volume.display_name, 607 ] 608 verifylist = [ 609 ('columns', ['display_name']), 610 ('size', self.new_volume.size), 611 ('name', self.new_volume.display_name), 612 ] 613 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 614 615 columns, data = self.cmd.take_action(parsed_args) 616 617 self.volumes_mock.create.assert_called_with( 618 self.new_volume.size, 619 None, 620 None, 621 self.new_volume.display_name, 622 None, 623 None, 624 None, 625 None, 626 None, 627 None, 628 None, 629 ) 630 self.assertIn('display_name', columns) 631 self.assertNotIn('name', columns) 632 self.assertIn(self.new_volume.display_name, data) 633 634 635class TestVolumeDelete(TestVolume): 636 637 def setUp(self): 638 super(TestVolumeDelete, self).setUp() 639 640 self.volumes_mock.delete.return_value = None 641 642 # Get the command object to mock 643 self.cmd = volume.DeleteVolume(self.app, None) 644 645 def test_volume_delete_one_volume(self): 646 volumes = self.setup_volumes_mock(count=1) 647 648 arglist = [ 649 volumes[0].id 650 ] 651 verifylist = [ 652 ("force", False), 653 ("volumes", [volumes[0].id]), 654 ] 655 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 656 657 result = self.cmd.take_action(parsed_args) 658 659 self.volumes_mock.delete.assert_called_once_with(volumes[0].id) 660 self.assertIsNone(result) 661 662 def test_volume_delete_multi_volumes(self): 663 volumes = self.setup_volumes_mock(count=3) 664 665 arglist = [v.id for v in volumes] 666 verifylist = [ 667 ('force', False), 668 ('volumes', arglist), 669 ] 670 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 671 672 result = self.cmd.take_action(parsed_args) 673 674 calls = [call(v.id) for v in volumes] 675 self.volumes_mock.delete.assert_has_calls(calls) 676 self.assertIsNone(result) 677 678 def test_volume_delete_multi_volumes_with_exception(self): 679 volumes = self.setup_volumes_mock(count=2) 680 681 arglist = [ 682 volumes[0].id, 683 'unexist_volume', 684 ] 685 verifylist = [ 686 ('force', False), 687 ('volumes', arglist), 688 ] 689 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 690 691 find_mock_result = [volumes[0], exceptions.CommandError] 692 with mock.patch.object(utils, 'find_resource', 693 side_effect=find_mock_result) as find_mock: 694 try: 695 self.cmd.take_action(parsed_args) 696 self.fail('CommandError should be raised.') 697 except exceptions.CommandError as e: 698 self.assertEqual('1 of 2 volumes failed to delete.', 699 str(e)) 700 701 find_mock.assert_any_call(self.volumes_mock, volumes[0].id) 702 find_mock.assert_any_call(self.volumes_mock, 'unexist_volume') 703 704 self.assertEqual(2, find_mock.call_count) 705 self.volumes_mock.delete.assert_called_once_with(volumes[0].id) 706 707 def test_volume_delete_with_force(self): 708 volumes = self.setup_volumes_mock(count=1) 709 710 arglist = [ 711 '--force', 712 volumes[0].id, 713 ] 714 verifylist = [ 715 ('force', True), 716 ('volumes', [volumes[0].id]), 717 ] 718 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 719 720 result = self.cmd.take_action(parsed_args) 721 722 self.volumes_mock.force_delete.assert_called_once_with(volumes[0].id) 723 self.assertIsNone(result) 724 725 726class TestVolumeList(TestVolume): 727 728 _volume = volume_fakes.FakeVolume.create_one_volume() 729 columns = ( 730 'ID', 731 'Name', 732 'Status', 733 'Size', 734 'Attached to', 735 ) 736 datalist = ( 737 ( 738 _volume.id, 739 _volume.display_name, 740 _volume.status, 741 _volume.size, 742 volume.AttachmentsColumn(_volume.attachments), 743 ), 744 ) 745 746 def setUp(self): 747 super(TestVolumeList, self).setUp() 748 749 self.volumes_mock.list.return_value = [self._volume] 750 751 # Get the command object to test 752 self.cmd = volume.ListVolume(self.app, None) 753 754 def test_volume_list_no_options(self): 755 arglist = [] 756 verifylist = [ 757 ('long', False), 758 ('all_projects', False), 759 ('name', None), 760 ('status', None), 761 ('limit', None), 762 ] 763 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 764 765 columns, data = self.cmd.take_action(parsed_args) 766 767 self.assertEqual(self.columns, columns) 768 self.assertListItemEqual(self.datalist, tuple(data)) 769 770 def test_volume_list_name(self): 771 arglist = [ 772 '--name', self._volume.display_name, 773 ] 774 verifylist = [ 775 ('long', False), 776 ('all_projects', False), 777 ('name', self._volume.display_name), 778 ('status', None), 779 ('limit', None), 780 ] 781 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 782 783 columns, data = self.cmd.take_action(parsed_args) 784 self.assertEqual(self.columns, tuple(columns)) 785 self.assertListItemEqual(self.datalist, tuple(data)) 786 787 def test_volume_list_status(self): 788 arglist = [ 789 '--status', self._volume.status, 790 ] 791 verifylist = [ 792 ('long', False), 793 ('all_projects', False), 794 ('name', None), 795 ('status', self._volume.status), 796 ('limit', None), 797 ] 798 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 799 800 columns, data = self.cmd.take_action(parsed_args) 801 self.assertEqual(self.columns, tuple(columns)) 802 self.assertListItemEqual(self.datalist, tuple(data)) 803 804 def test_volume_list_all_projects(self): 805 arglist = [ 806 '--all-projects', 807 ] 808 verifylist = [ 809 ('long', False), 810 ('all_projects', True), 811 ('name', None), 812 ('status', None), 813 ('limit', None), 814 ] 815 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 816 817 columns, data = self.cmd.take_action(parsed_args) 818 self.assertEqual(self.columns, tuple(columns)) 819 self.assertListItemEqual(self.datalist, tuple(data)) 820 821 def test_volume_list_long(self): 822 arglist = [ 823 '--long', 824 ] 825 verifylist = [ 826 ('long', True), 827 ('all_projects', False), 828 ('name', None), 829 ('status', None), 830 ('limit', None), 831 ] 832 833 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 834 835 columns, data = self.cmd.take_action(parsed_args) 836 837 collist = ( 838 'ID', 839 'Name', 840 'Status', 841 'Size', 842 'Type', 843 'Bootable', 844 'Attached to', 845 'Properties', 846 ) 847 self.assertEqual(collist, columns) 848 849 datalist = (( 850 self._volume.id, 851 self._volume.display_name, 852 self._volume.status, 853 self._volume.size, 854 self._volume.volume_type, 855 self._volume.bootable, 856 volume.AttachmentsColumn(self._volume.attachments), 857 format_columns.DictColumn(self._volume.metadata), 858 ), ) 859 self.assertListItemEqual(datalist, tuple(data)) 860 861 def test_volume_list_with_limit(self): 862 arglist = [ 863 '--limit', '2', 864 ] 865 verifylist = [ 866 ('long', False), 867 ('all_projects', False), 868 ('name', None), 869 ('status', None), 870 ('limit', 2), 871 ] 872 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 873 874 columns, data = self.cmd.take_action(parsed_args) 875 876 self.volumes_mock.list.assert_called_once_with( 877 limit=2, 878 search_opts={ 879 'status': None, 880 'display_name': None, 881 'all_tenants': False, } 882 ) 883 self.assertEqual(self.columns, columns) 884 self.assertListItemEqual(self.datalist, tuple(data)) 885 886 def test_volume_list_negative_limit(self): 887 arglist = [ 888 "--limit", "-2", 889 ] 890 verifylist = [ 891 ("limit", -2), 892 ] 893 self.assertRaises(argparse.ArgumentTypeError, self.check_parser, 894 self.cmd, arglist, verifylist) 895 896 def test_volume_list_backward_compatibility(self): 897 arglist = [ 898 '-c', 'Display Name', 899 ] 900 verifylist = [ 901 ('columns', ['Display Name']), 902 ('long', False), 903 ('all_projects', False), 904 ('name', None), 905 ('status', None), 906 ('limit', None), 907 ] 908 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 909 910 columns, data = self.cmd.take_action(parsed_args) 911 912 self.assertIn('Display Name', columns) 913 self.assertNotIn('Name', columns) 914 for each_volume in data: 915 self.assertIn(self._volume.display_name, each_volume) 916 917 918class TestVolumeMigrate(TestVolume): 919 920 _volume = volume_fakes.FakeVolume.create_one_volume() 921 922 def setUp(self): 923 super(TestVolumeMigrate, self).setUp() 924 925 self.volumes_mock.get.return_value = self._volume 926 self.volumes_mock.migrate_volume.return_value = None 927 # Get the command object to test 928 self.cmd = volume.MigrateVolume(self.app, None) 929 930 def test_volume_migrate(self): 931 arglist = [ 932 "--host", "host@backend-name#pool", 933 self._volume.id, 934 ] 935 verifylist = [ 936 ("force_host_copy", False), 937 ("host", "host@backend-name#pool"), 938 ("volume", self._volume.id), 939 ] 940 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 941 942 result = self.cmd.take_action(parsed_args) 943 self.volumes_mock.get.assert_called_once_with(self._volume.id) 944 self.volumes_mock.migrate_volume.assert_called_once_with( 945 self._volume.id, "host@backend-name#pool", False) 946 self.assertIsNone(result) 947 948 def test_volume_migrate_with_option(self): 949 arglist = [ 950 "--force-host-copy", 951 "--host", "host@backend-name#pool", 952 self._volume.id, 953 ] 954 verifylist = [ 955 ("force_host_copy", True), 956 ("host", "host@backend-name#pool"), 957 ("volume", self._volume.id), 958 ] 959 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 960 961 result = self.cmd.take_action(parsed_args) 962 self.volumes_mock.get.assert_called_once_with(self._volume.id) 963 self.volumes_mock.migrate_volume.assert_called_once_with( 964 self._volume.id, "host@backend-name#pool", True) 965 self.assertIsNone(result) 966 967 def test_volume_migrate_without_host(self): 968 arglist = [ 969 self._volume.id, 970 ] 971 verifylist = [ 972 ("force_host_copy", False), 973 ("volume", self._volume.id), 974 ] 975 976 self.assertRaises(tests_utils.ParserException, self.check_parser, 977 self.cmd, arglist, verifylist) 978 979 980class TestVolumeSet(TestVolume): 981 982 _volume = volume_fakes.FakeVolume.create_one_volume() 983 984 def setUp(self): 985 super(TestVolumeSet, self).setUp() 986 987 self.volumes_mock.get.return_value = self._volume 988 989 self.volumes_mock.update.return_value = self._volume 990 # Get the command object to test 991 self.cmd = volume.SetVolume(self.app, None) 992 993 def test_volume_set_no_options(self): 994 arglist = [ 995 self._volume.display_name, 996 ] 997 verifylist = [ 998 ('name', None), 999 ('description', None), 1000 ('size', None), 1001 ('property', None), 1002 ('volume', self._volume.display_name), 1003 ] 1004 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 1005 1006 result = self.cmd.take_action(parsed_args) 1007 self.assertIsNone(result) 1008 1009 def test_volume_set_name(self): 1010 arglist = [ 1011 '--name', 'qwerty', 1012 self._volume.display_name, 1013 ] 1014 verifylist = [ 1015 ('name', 'qwerty'), 1016 ('description', None), 1017 ('size', None), 1018 ('property', None), 1019 ('volume', self._volume.display_name), 1020 ] 1021 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 1022 1023 result = self.cmd.take_action(parsed_args) 1024 1025 # Set expected values 1026 kwargs = { 1027 'display_name': 'qwerty', 1028 } 1029 self.volumes_mock.update.assert_called_with( 1030 self._volume.id, 1031 **kwargs 1032 ) 1033 self.assertIsNone(result) 1034 1035 def test_volume_set_description(self): 1036 arglist = [ 1037 '--description', 'new desc', 1038 self._volume.display_name, 1039 ] 1040 verifylist = [ 1041 ('name', None), 1042 ('description', 'new desc'), 1043 ('size', None), 1044 ('property', None), 1045 ('volume', self._volume.display_name), 1046 ] 1047 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 1048 1049 result = self.cmd.take_action(parsed_args) 1050 1051 # Set expected values 1052 kwargs = { 1053 'display_description': 'new desc', 1054 } 1055 self.volumes_mock.update.assert_called_with( 1056 self._volume.id, 1057 **kwargs 1058 ) 1059 self.assertIsNone(result) 1060 1061 def test_volume_set_size(self): 1062 arglist = [ 1063 '--size', '130', 1064 self._volume.display_name, 1065 ] 1066 verifylist = [ 1067 ('name', None), 1068 ('description', None), 1069 ('size', 130), 1070 ('property', None), 1071 ('volume', self._volume.display_name), 1072 ] 1073 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 1074 1075 result = self.cmd.take_action(parsed_args) 1076 1077 # Set expected values 1078 size = 130 1079 self.volumes_mock.extend.assert_called_with( 1080 self._volume.id, 1081 size 1082 ) 1083 self.assertIsNone(result) 1084 1085 def test_volume_set_size_smaller(self): 1086 self._volume.status = 'available' 1087 arglist = [ 1088 '--size', '1', 1089 self._volume.display_name, 1090 ] 1091 verifylist = [ 1092 ('name', None), 1093 ('description', None), 1094 ('size', 1), 1095 ('property', None), 1096 ('volume', self._volume.display_name), 1097 ] 1098 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 1099 1100 self.assertRaises(exceptions.CommandError, 1101 self.cmd.take_action, 1102 parsed_args) 1103 1104 def test_volume_set_size_not_available(self): 1105 self._volume.status = 'error' 1106 arglist = [ 1107 '--size', '130', 1108 self._volume.display_name, 1109 ] 1110 verifylist = [ 1111 ('name', None), 1112 ('description', None), 1113 ('size', 130), 1114 ('property', None), 1115 ('volume', self._volume.display_name), 1116 ] 1117 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 1118 1119 self.assertRaises(exceptions.CommandError, 1120 self.cmd.take_action, 1121 parsed_args) 1122 1123 def test_volume_set_property(self): 1124 arglist = [ 1125 '--no-property', 1126 '--property', 'myprop=myvalue', 1127 self._volume.display_name, 1128 ] 1129 verifylist = [ 1130 ('read_only', False), 1131 ('read_write', False), 1132 ('name', None), 1133 ('description', None), 1134 ('size', None), 1135 ('no_property', True), 1136 ('property', {'myprop': 'myvalue'}), 1137 ('volume', self._volume.display_name), 1138 ('bootable', False), 1139 ('non_bootable', False) 1140 ] 1141 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 1142 1143 result = self.cmd.take_action(parsed_args) 1144 1145 # Set expected values 1146 metadata = { 1147 'myprop': 'myvalue' 1148 } 1149 self.volumes_mock.set_metadata.assert_called_with( 1150 self._volume.id, 1151 metadata 1152 ) 1153 self.volumes_mock.delete_metadata.assert_called_with( 1154 self._volume.id, 1155 self._volume.metadata.keys() 1156 ) 1157 self.volumes_mock.update_readonly_flag.assert_not_called() 1158 self.assertIsNone(result) 1159 1160 def test_volume_set_bootable(self): 1161 arglist = [ 1162 ['--bootable', self._volume.id], 1163 ['--non-bootable', self._volume.id] 1164 ] 1165 verifylist = [ 1166 [ 1167 ('bootable', True), 1168 ('non_bootable', False), 1169 ('volume', self._volume.id) 1170 ], 1171 [ 1172 ('bootable', False), 1173 ('non_bootable', True), 1174 ('volume', self._volume.id) 1175 ] 1176 ] 1177 for index in range(len(arglist)): 1178 parsed_args = self.check_parser( 1179 self.cmd, arglist[index], verifylist[index]) 1180 1181 self.cmd.take_action(parsed_args) 1182 self.volumes_mock.set_bootable.assert_called_with( 1183 self._volume.id, verifylist[index][0][1]) 1184 1185 def test_volume_set_readonly(self): 1186 arglist = [ 1187 '--read-only', 1188 self._volume.id 1189 ] 1190 verifylist = [ 1191 ('read_only', True), 1192 ('read_write', False), 1193 ('volume', self._volume.id) 1194 ] 1195 1196 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 1197 1198 result = self.cmd.take_action(parsed_args) 1199 self.volumes_mock.update_readonly_flag.assert_called_once_with( 1200 self._volume.id, 1201 True) 1202 self.assertIsNone(result) 1203 1204 def test_volume_set_read_write(self): 1205 arglist = [ 1206 '--read-write', 1207 self._volume.id 1208 ] 1209 verifylist = [ 1210 ('read_only', False), 1211 ('read_write', True), 1212 ('volume', self._volume.id) 1213 ] 1214 1215 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 1216 1217 result = self.cmd.take_action(parsed_args) 1218 self.volumes_mock.update_readonly_flag.assert_called_once_with( 1219 self._volume.id, 1220 False) 1221 self.assertIsNone(result) 1222 1223 1224class TestVolumeShow(TestVolume): 1225 1226 columns = ( 1227 'attachments', 1228 'availability_zone', 1229 'bootable', 1230 'created_at', 1231 'display_description', 1232 'id', 1233 'name', 1234 'properties', 1235 'size', 1236 'snapshot_id', 1237 'status', 1238 'type', 1239 ) 1240 1241 def setUp(self): 1242 super(TestVolumeShow, self).setUp() 1243 self._volume = volume_fakes.FakeVolume.create_one_volume() 1244 self.datalist = ( 1245 self._volume.attachments, 1246 self._volume.availability_zone, 1247 self._volume.bootable, 1248 self._volume.created_at, 1249 self._volume.display_description, 1250 self._volume.id, 1251 self._volume.display_name, 1252 format_columns.DictColumn(self._volume.metadata), 1253 self._volume.size, 1254 self._volume.snapshot_id, 1255 self._volume.status, 1256 self._volume.volume_type, 1257 ) 1258 self.volumes_mock.get.return_value = self._volume 1259 # Get the command object to test 1260 self.cmd = volume.ShowVolume(self.app, None) 1261 1262 def test_volume_show(self): 1263 arglist = [ 1264 self._volume.id 1265 ] 1266 verifylist = [ 1267 ("volume", self._volume.id) 1268 ] 1269 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 1270 1271 columns, data = self.cmd.take_action(parsed_args) 1272 self.volumes_mock.get.assert_called_with(self._volume.id) 1273 1274 self.assertEqual(self.columns, columns) 1275 self.assertItemEqual(self.datalist, data) 1276 1277 def test_volume_show_backward_compatibility(self): 1278 arglist = [ 1279 '-c', 'display_name', 1280 self._volume.id, 1281 ] 1282 verifylist = [ 1283 ('columns', ['display_name']), 1284 ('volume', self._volume.id), 1285 ] 1286 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 1287 1288 columns, data = self.cmd.take_action(parsed_args) 1289 1290 self.volumes_mock.get.assert_called_with(self._volume.id) 1291 1292 self.assertIn('display_name', columns) 1293 self.assertNotIn('name', columns) 1294 self.assertIn(self._volume.display_name, data) 1295 1296 1297class TestVolumeUnset(TestVolume): 1298 1299 _volume = volume_fakes.FakeVolume.create_one_volume() 1300 1301 def setUp(self): 1302 super(TestVolumeUnset, self).setUp() 1303 1304 self.volumes_mock.get.return_value = self._volume 1305 1306 self.volumes_mock.delete_metadata.return_value = None 1307 # Get the command object to test 1308 self.cmd = volume.UnsetVolume(self.app, None) 1309 1310 def test_volume_unset_no_options(self): 1311 arglist = [ 1312 self._volume.display_name, 1313 ] 1314 verifylist = [ 1315 ('property', None), 1316 ('volume', self._volume.display_name), 1317 ] 1318 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 1319 1320 result = self.cmd.take_action(parsed_args) 1321 self.assertIsNone(result) 1322 1323 def test_volume_unset_property(self): 1324 arglist = [ 1325 '--property', 'myprop', 1326 self._volume.display_name, 1327 ] 1328 verifylist = [ 1329 ('property', ['myprop']), 1330 ('volume', self._volume.display_name), 1331 ] 1332 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 1333 1334 result = self.cmd.take_action(parsed_args) 1335 1336 self.volumes_mock.delete_metadata.assert_called_with( 1337 self._volume.id, ['myprop'] 1338 ) 1339 self.assertIsNone(result) 1340 1341 1342class TestColumns(TestVolume): 1343 1344 def test_attachments_column_without_server_cache(self): 1345 _volume = volume_fakes.FakeVolume.create_one_volume() 1346 server_id = _volume.attachments[0]['server_id'] 1347 device = _volume.attachments[0]['device'] 1348 1349 col = volume.AttachmentsColumn(_volume.attachments, {}) 1350 self.assertEqual('Attached to %s on %s ' % (server_id, device), 1351 col.human_readable()) 1352 self.assertEqual(_volume.attachments, col.machine_readable()) 1353 1354 def test_attachments_column_with_server_cache(self): 1355 _volume = volume_fakes.FakeVolume.create_one_volume() 1356 1357 server_id = _volume.attachments[0]['server_id'] 1358 device = _volume.attachments[0]['device'] 1359 fake_server = mock.Mock() 1360 fake_server.name = 'fake-server-name' 1361 server_cache = {server_id: fake_server} 1362 1363 col = volume.AttachmentsColumn(_volume.attachments, server_cache) 1364 self.assertEqual( 1365 'Attached to %s on %s ' % ('fake-server-name', device), 1366 col.human_readable()) 1367 self.assertEqual(_volume.attachments, col.machine_readable()) 1368