1# 2# Licensed under the Apache License, Version 2.0 (the "License"); you may 3# not use this file except in compliance with the License. You may obtain 4# a copy of the License at 5# 6# http://www.apache.org/licenses/LICENSE-2.0 7# 8# Unless required by applicable law or agreed to in writing, software 9# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 10# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 11# License for the specific language governing permissions and limitations 12# under the License. 13# 14 15from unittest import mock 16from unittest.mock import call 17 18from osc_lib.cli import format_columns 19from osc_lib import exceptions 20from osc_lib import utils 21 22from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes 23from openstackclient.volume.v2 import consistency_group 24 25 26class TestConsistencyGroup(volume_fakes.TestVolume): 27 28 def setUp(self): 29 super(TestConsistencyGroup, self).setUp() 30 31 # Get a shortcut to the TransferManager Mock 32 self.consistencygroups_mock = ( 33 self.app.client_manager.volume.consistencygroups) 34 self.consistencygroups_mock.reset_mock() 35 36 self.cgsnapshots_mock = ( 37 self.app.client_manager.volume.cgsnapshots) 38 self.cgsnapshots_mock.reset_mock() 39 40 self.volumes_mock = ( 41 self.app.client_manager.volume.volumes) 42 self.volumes_mock.reset_mock() 43 44 self.types_mock = self.app.client_manager.volume.volume_types 45 self.types_mock.reset_mock() 46 47 48class TestConsistencyGroupAddVolume(TestConsistencyGroup): 49 50 _consistency_group = ( 51 volume_fakes.FakeConsistencyGroup.create_one_consistency_group()) 52 53 def setUp(self): 54 super(TestConsistencyGroupAddVolume, self).setUp() 55 56 self.consistencygroups_mock.get.return_value = ( 57 self._consistency_group) 58 # Get the command object to test 59 self.cmd = \ 60 consistency_group.AddVolumeToConsistencyGroup(self.app, None) 61 62 def test_add_one_volume_to_consistency_group(self): 63 volume = volume_fakes.FakeVolume.create_one_volume() 64 self.volumes_mock.get.return_value = volume 65 arglist = [ 66 self._consistency_group.id, 67 volume.id, 68 ] 69 verifylist = [ 70 ('consistency_group', self._consistency_group.id), 71 ('volumes', [volume.id]), 72 ] 73 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 74 75 result = self.cmd.take_action(parsed_args) 76 77 # Set expected values 78 kwargs = { 79 'add_volumes': volume.id, 80 } 81 self.consistencygroups_mock.update.assert_called_once_with( 82 self._consistency_group.id, 83 **kwargs 84 ) 85 self.assertIsNone(result) 86 87 def test_add_multiple_volumes_to_consistency_group(self): 88 volumes = volume_fakes.FakeVolume.create_volumes(count=2) 89 self.volumes_mock.get = volume_fakes.FakeVolume.get_volumes(volumes) 90 arglist = [ 91 self._consistency_group.id, 92 volumes[0].id, 93 volumes[1].id, 94 ] 95 verifylist = [ 96 ('consistency_group', self._consistency_group.id), 97 ('volumes', [volumes[0].id, volumes[1].id]), 98 ] 99 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 100 101 result = self.cmd.take_action(parsed_args) 102 103 # Set expected values 104 kwargs = { 105 'add_volumes': volumes[0].id + ',' + volumes[1].id, 106 } 107 self.consistencygroups_mock.update.assert_called_once_with( 108 self._consistency_group.id, 109 **kwargs 110 ) 111 self.assertIsNone(result) 112 113 @mock.patch.object(consistency_group.LOG, 'error') 114 def test_add_multiple_volumes_to_consistency_group_with_exception( 115 self, mock_error): 116 volume = volume_fakes.FakeVolume.create_one_volume() 117 arglist = [ 118 self._consistency_group.id, 119 volume.id, 120 'unexist_volume', 121 ] 122 verifylist = [ 123 ('consistency_group', self._consistency_group.id), 124 ('volumes', [volume.id, 'unexist_volume']), 125 ] 126 127 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 128 129 find_mock_result = [volume, 130 exceptions.CommandError, 131 self._consistency_group] 132 with mock.patch.object(utils, 'find_resource', 133 side_effect=find_mock_result) as find_mock: 134 result = self.cmd.take_action(parsed_args) 135 mock_error.assert_called_with("1 of 2 volumes failed to add.") 136 self.assertIsNone(result) 137 find_mock.assert_any_call(self.consistencygroups_mock, 138 self._consistency_group.id) 139 find_mock.assert_any_call(self.volumes_mock, 140 volume.id) 141 find_mock.assert_any_call(self.volumes_mock, 142 'unexist_volume') 143 self.assertEqual(3, find_mock.call_count) 144 self.consistencygroups_mock.update.assert_called_once_with( 145 self._consistency_group.id, add_volumes=volume.id 146 ) 147 148 149class TestConsistencyGroupCreate(TestConsistencyGroup): 150 151 volume_type = volume_fakes.FakeType.create_one_type() 152 new_consistency_group = ( 153 volume_fakes.FakeConsistencyGroup.create_one_consistency_group()) 154 consistency_group_snapshot = ( 155 volume_fakes. 156 FakeConsistencyGroupSnapshot. 157 create_one_consistency_group_snapshot() 158 ) 159 160 columns = ( 161 'availability_zone', 162 'created_at', 163 'description', 164 'id', 165 'name', 166 'status', 167 'volume_types', 168 ) 169 data = ( 170 new_consistency_group.availability_zone, 171 new_consistency_group.created_at, 172 new_consistency_group.description, 173 new_consistency_group.id, 174 new_consistency_group.name, 175 new_consistency_group.status, 176 new_consistency_group.volume_types, 177 ) 178 179 def setUp(self): 180 super(TestConsistencyGroupCreate, self).setUp() 181 self.consistencygroups_mock.create.return_value = ( 182 self.new_consistency_group) 183 self.consistencygroups_mock.create_from_src.return_value = ( 184 self.new_consistency_group) 185 self.consistencygroups_mock.get.return_value = ( 186 self.new_consistency_group) 187 self.types_mock.get.return_value = self.volume_type 188 self.cgsnapshots_mock.get.return_value = ( 189 self.consistency_group_snapshot) 190 191 # Get the command object to test 192 self.cmd = consistency_group.CreateConsistencyGroup(self.app, None) 193 194 def test_consistency_group_create(self): 195 arglist = [ 196 '--volume-type', self.volume_type.id, 197 '--description', self.new_consistency_group.description, 198 '--availability-zone', 199 self.new_consistency_group.availability_zone, 200 self.new_consistency_group.name, 201 ] 202 verifylist = [ 203 ('volume_type', self.volume_type.id), 204 ('description', self.new_consistency_group.description), 205 ('availability_zone', 206 self.new_consistency_group.availability_zone), 207 ('name', self.new_consistency_group.name), 208 ] 209 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 210 211 columns, data = self.cmd.take_action(parsed_args) 212 213 self.types_mock.get.assert_called_once_with( 214 self.volume_type.id) 215 self.consistencygroups_mock.get.assert_not_called() 216 self.consistencygroups_mock.create.assert_called_once_with( 217 self.volume_type.id, 218 name=self.new_consistency_group.name, 219 description=self.new_consistency_group.description, 220 availability_zone=self.new_consistency_group.availability_zone, 221 ) 222 223 self.assertEqual(self.columns, columns) 224 self.assertEqual(self.data, data) 225 226 def test_consistency_group_create_without_name(self): 227 arglist = [ 228 '--volume-type', self.volume_type.id, 229 '--description', self.new_consistency_group.description, 230 '--availability-zone', 231 self.new_consistency_group.availability_zone, 232 ] 233 verifylist = [ 234 ('volume_type', self.volume_type.id), 235 ('description', self.new_consistency_group.description), 236 ('availability_zone', 237 self.new_consistency_group.availability_zone), 238 ] 239 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 240 241 columns, data = self.cmd.take_action(parsed_args) 242 243 self.types_mock.get.assert_called_once_with( 244 self.volume_type.id) 245 self.consistencygroups_mock.get.assert_not_called() 246 self.consistencygroups_mock.create.assert_called_once_with( 247 self.volume_type.id, 248 name=None, 249 description=self.new_consistency_group.description, 250 availability_zone=self.new_consistency_group.availability_zone, 251 ) 252 253 self.assertEqual(self.columns, columns) 254 self.assertItemEqual(self.data, data) 255 256 def test_consistency_group_create_from_source(self): 257 arglist = [ 258 '--consistency-group-source', self.new_consistency_group.id, 259 '--description', self.new_consistency_group.description, 260 self.new_consistency_group.name, 261 ] 262 verifylist = [ 263 ('consistency_group_source', self.new_consistency_group.id), 264 ('description', self.new_consistency_group.description), 265 ('name', self.new_consistency_group.name), 266 ] 267 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 268 269 columns, data = self.cmd.take_action(parsed_args) 270 271 self.types_mock.get.assert_not_called() 272 self.consistencygroups_mock.get.assert_called_once_with( 273 self.new_consistency_group.id) 274 self.consistencygroups_mock.create_from_src.assert_called_with( 275 None, 276 self.new_consistency_group.id, 277 name=self.new_consistency_group.name, 278 description=self.new_consistency_group.description, 279 ) 280 281 self.assertEqual(self.columns, columns) 282 self.assertItemEqual(self.data, data) 283 284 def test_consistency_group_create_from_snapshot(self): 285 arglist = [ 286 '--consistency-group-snapshot', self.consistency_group_snapshot.id, 287 '--description', self.new_consistency_group.description, 288 self.new_consistency_group.name, 289 ] 290 verifylist = [ 291 ('consistency_group_snapshot', self.consistency_group_snapshot.id), 292 ('description', self.new_consistency_group.description), 293 ('name', self.new_consistency_group.name), 294 ] 295 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 296 297 columns, data = self.cmd.take_action(parsed_args) 298 299 self.types_mock.get.assert_not_called() 300 self.cgsnapshots_mock.get.assert_called_once_with( 301 self.consistency_group_snapshot.id) 302 self.consistencygroups_mock.create_from_src.assert_called_with( 303 self.consistency_group_snapshot.id, 304 None, 305 name=self.new_consistency_group.name, 306 description=self.new_consistency_group.description, 307 ) 308 309 self.assertEqual(self.columns, columns) 310 self.assertItemEqual(self.data, data) 311 312 313class TestConsistencyGroupDelete(TestConsistencyGroup): 314 315 consistency_groups =\ 316 volume_fakes.FakeConsistencyGroup.create_consistency_groups(count=2) 317 318 def setUp(self): 319 super(TestConsistencyGroupDelete, self).setUp() 320 321 self.consistencygroups_mock.get = volume_fakes.FakeConsistencyGroup.\ 322 get_consistency_groups(self.consistency_groups) 323 self.consistencygroups_mock.delete.return_value = None 324 325 # Get the command object to mock 326 self.cmd = consistency_group.DeleteConsistencyGroup(self.app, None) 327 328 def test_consistency_group_delete(self): 329 arglist = [ 330 self.consistency_groups[0].id 331 ] 332 verifylist = [ 333 ("consistency_groups", [self.consistency_groups[0].id]) 334 ] 335 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 336 337 result = self.cmd.take_action(parsed_args) 338 339 self.consistencygroups_mock.delete.assert_called_with( 340 self.consistency_groups[0].id, False) 341 self.assertIsNone(result) 342 343 def test_consistency_group_delete_with_force(self): 344 arglist = [ 345 '--force', 346 self.consistency_groups[0].id, 347 ] 348 verifylist = [ 349 ('force', True), 350 ("consistency_groups", [self.consistency_groups[0].id]) 351 ] 352 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 353 354 result = self.cmd.take_action(parsed_args) 355 356 self.consistencygroups_mock.delete.assert_called_with( 357 self.consistency_groups[0].id, True) 358 self.assertIsNone(result) 359 360 def test_delete_multiple_consistency_groups(self): 361 arglist = [] 362 for b in self.consistency_groups: 363 arglist.append(b.id) 364 verifylist = [ 365 ('consistency_groups', arglist), 366 ] 367 368 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 369 result = self.cmd.take_action(parsed_args) 370 371 calls = [] 372 for b in self.consistency_groups: 373 calls.append(call(b.id, False)) 374 self.consistencygroups_mock.delete.assert_has_calls(calls) 375 self.assertIsNone(result) 376 377 def test_delete_multiple_consistency_groups_with_exception(self): 378 arglist = [ 379 self.consistency_groups[0].id, 380 'unexist_consistency_group', 381 ] 382 verifylist = [ 383 ('consistency_groups', arglist), 384 ] 385 386 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 387 388 find_mock_result = [self.consistency_groups[0], 389 exceptions.CommandError] 390 with mock.patch.object(utils, 'find_resource', 391 side_effect=find_mock_result) as find_mock: 392 try: 393 self.cmd.take_action(parsed_args) 394 self.fail('CommandError should be raised.') 395 except exceptions.CommandError as e: 396 self.assertEqual('1 of 2 consistency groups failed to delete.', 397 str(e)) 398 399 find_mock.assert_any_call(self.consistencygroups_mock, 400 self.consistency_groups[0].id) 401 find_mock.assert_any_call(self.consistencygroups_mock, 402 'unexist_consistency_group') 403 404 self.assertEqual(2, find_mock.call_count) 405 self.consistencygroups_mock.delete.assert_called_once_with( 406 self.consistency_groups[0].id, False 407 ) 408 409 410class TestConsistencyGroupList(TestConsistencyGroup): 411 412 consistency_groups = ( 413 volume_fakes.FakeConsistencyGroup.create_consistency_groups(count=2)) 414 415 columns = [ 416 'ID', 417 'Status', 418 'Name', 419 ] 420 columns_long = [ 421 'ID', 422 'Status', 423 'Availability Zone', 424 'Name', 425 'Description', 426 'Volume Types', 427 ] 428 data = [] 429 for c in consistency_groups: 430 data.append(( 431 c.id, 432 c.status, 433 c.name, 434 )) 435 data_long = [] 436 for c in consistency_groups: 437 data_long.append(( 438 c.id, 439 c.status, 440 c.availability_zone, 441 c.name, 442 c.description, 443 format_columns.ListColumn(c.volume_types) 444 )) 445 446 def setUp(self): 447 super(TestConsistencyGroupList, self).setUp() 448 449 self.consistencygroups_mock.list.return_value = self.consistency_groups 450 # Get the command to test 451 self.cmd = consistency_group.ListConsistencyGroup(self.app, None) 452 453 def test_consistency_group_list_without_options(self): 454 arglist = [] 455 verifylist = [ 456 ("all_projects", False), 457 ("long", False), 458 ] 459 460 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 461 columns, data = self.cmd.take_action(parsed_args) 462 463 self.consistencygroups_mock.list.assert_called_once_with( 464 detailed=True, search_opts={'all_tenants': False}) 465 self.assertEqual(self.columns, columns) 466 self.assertListItemEqual(self.data, list(data)) 467 468 def test_consistency_group_list_with_all_project(self): 469 arglist = [ 470 "--all-projects" 471 ] 472 verifylist = [ 473 ("all_projects", True), 474 ("long", False), 475 ] 476 477 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 478 columns, data = self.cmd.take_action(parsed_args) 479 480 self.consistencygroups_mock.list.assert_called_once_with( 481 detailed=True, search_opts={'all_tenants': True}) 482 self.assertEqual(self.columns, columns) 483 self.assertListItemEqual(self.data, list(data)) 484 485 def test_consistency_group_list_with_long(self): 486 arglist = [ 487 "--long", 488 ] 489 verifylist = [ 490 ("all_projects", False), 491 ("long", True), 492 ] 493 494 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 495 columns, data = self.cmd.take_action(parsed_args) 496 497 self.consistencygroups_mock.list.assert_called_once_with( 498 detailed=True, search_opts={'all_tenants': False}) 499 self.assertEqual(self.columns_long, columns) 500 self.assertListItemEqual(self.data_long, list(data)) 501 502 503class TestConsistencyGroupRemoveVolume(TestConsistencyGroup): 504 505 _consistency_group = ( 506 volume_fakes.FakeConsistencyGroup.create_one_consistency_group()) 507 508 def setUp(self): 509 super(TestConsistencyGroupRemoveVolume, self).setUp() 510 511 self.consistencygroups_mock.get.return_value = ( 512 self._consistency_group) 513 # Get the command object to test 514 self.cmd = \ 515 consistency_group.RemoveVolumeFromConsistencyGroup(self.app, None) 516 517 def test_remove_one_volume_from_consistency_group(self): 518 volume = volume_fakes.FakeVolume.create_one_volume() 519 self.volumes_mock.get.return_value = volume 520 arglist = [ 521 self._consistency_group.id, 522 volume.id, 523 ] 524 verifylist = [ 525 ('consistency_group', self._consistency_group.id), 526 ('volumes', [volume.id]), 527 ] 528 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 529 530 result = self.cmd.take_action(parsed_args) 531 532 # Set expected values 533 kwargs = { 534 'remove_volumes': volume.id, 535 } 536 self.consistencygroups_mock.update.assert_called_once_with( 537 self._consistency_group.id, 538 **kwargs 539 ) 540 self.assertIsNone(result) 541 542 def test_remove_multi_volumes_from_consistency_group(self): 543 volumes = volume_fakes.FakeVolume.create_volumes(count=2) 544 self.volumes_mock.get = volume_fakes.FakeVolume.get_volumes(volumes) 545 arglist = [ 546 self._consistency_group.id, 547 volumes[0].id, 548 volumes[1].id, 549 ] 550 verifylist = [ 551 ('consistency_group', self._consistency_group.id), 552 ('volumes', [volumes[0].id, volumes[1].id]), 553 ] 554 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 555 556 result = self.cmd.take_action(parsed_args) 557 558 # Set expected values 559 kwargs = { 560 'remove_volumes': volumes[0].id + ',' + volumes[1].id, 561 } 562 self.consistencygroups_mock.update.assert_called_once_with( 563 self._consistency_group.id, 564 **kwargs 565 ) 566 self.assertIsNone(result) 567 568 @mock.patch.object(consistency_group.LOG, 'error') 569 def test_remove_multiple_volumes_from_consistency_group_with_exception( 570 self, mock_error): 571 volume = volume_fakes.FakeVolume.create_one_volume() 572 arglist = [ 573 self._consistency_group.id, 574 volume.id, 575 'unexist_volume', 576 ] 577 verifylist = [ 578 ('consistency_group', self._consistency_group.id), 579 ('volumes', [volume.id, 'unexist_volume']), 580 ] 581 582 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 583 584 find_mock_result = [volume, 585 exceptions.CommandError, 586 self._consistency_group] 587 with mock.patch.object(utils, 'find_resource', 588 side_effect=find_mock_result) as find_mock: 589 result = self.cmd.take_action(parsed_args) 590 mock_error.assert_called_with("1 of 2 volumes failed to remove.") 591 self.assertIsNone(result) 592 find_mock.assert_any_call(self.consistencygroups_mock, 593 self._consistency_group.id) 594 find_mock.assert_any_call(self.volumes_mock, 595 volume.id) 596 find_mock.assert_any_call(self.volumes_mock, 597 'unexist_volume') 598 self.assertEqual(3, find_mock.call_count) 599 self.consistencygroups_mock.update.assert_called_once_with( 600 self._consistency_group.id, remove_volumes=volume.id 601 ) 602 603 604class TestConsistencyGroupSet(TestConsistencyGroup): 605 606 consistency_group = ( 607 volume_fakes.FakeConsistencyGroup.create_one_consistency_group()) 608 609 def setUp(self): 610 super(TestConsistencyGroupSet, self).setUp() 611 612 self.consistencygroups_mock.get.return_value = ( 613 self.consistency_group) 614 # Get the command object to test 615 self.cmd = consistency_group.SetConsistencyGroup(self.app, None) 616 617 def test_consistency_group_set_name(self): 618 new_name = 'new_name' 619 arglist = [ 620 '--name', new_name, 621 self.consistency_group.id, 622 ] 623 verifylist = [ 624 ('name', new_name), 625 ('description', None), 626 ('consistency_group', self.consistency_group.id), 627 ] 628 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 629 630 result = self.cmd.take_action(parsed_args) 631 632 # Set expected values 633 kwargs = { 634 'name': new_name, 635 } 636 self.consistencygroups_mock.update.assert_called_once_with( 637 self.consistency_group.id, 638 **kwargs 639 ) 640 self.assertIsNone(result) 641 642 def test_consistency_group_set_description(self): 643 new_description = 'new_description' 644 arglist = [ 645 '--description', new_description, 646 self.consistency_group.id, 647 ] 648 verifylist = [ 649 ('name', None), 650 ('description', new_description), 651 ('consistency_group', self.consistency_group.id), 652 ] 653 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 654 655 result = self.cmd.take_action(parsed_args) 656 657 # Set expected values 658 kwargs = { 659 'description': new_description, 660 } 661 self.consistencygroups_mock.update.assert_called_once_with( 662 self.consistency_group.id, 663 **kwargs 664 ) 665 self.assertIsNone(result) 666 667 668class TestConsistencyGroupShow(TestConsistencyGroup): 669 columns = ( 670 'availability_zone', 671 'created_at', 672 'description', 673 'id', 674 'name', 675 'status', 676 'volume_types', 677 ) 678 679 def setUp(self): 680 super(TestConsistencyGroupShow, self).setUp() 681 682 self.consistency_group = ( 683 volume_fakes.FakeConsistencyGroup.create_one_consistency_group()) 684 self.data = ( 685 self.consistency_group.availability_zone, 686 self.consistency_group.created_at, 687 self.consistency_group.description, 688 self.consistency_group.id, 689 self.consistency_group.name, 690 self.consistency_group.status, 691 self.consistency_group.volume_types, 692 ) 693 self.consistencygroups_mock.get.return_value = self.consistency_group 694 self.cmd = consistency_group.ShowConsistencyGroup(self.app, None) 695 696 def test_consistency_group_show(self): 697 arglist = [ 698 self.consistency_group.id 699 ] 700 verifylist = [ 701 ("consistency_group", self.consistency_group.id) 702 ] 703 parsed_args = self.check_parser(self.cmd, arglist, verifylist) 704 columns, data = self.cmd.take_action(parsed_args) 705 self.consistencygroups_mock.get.assert_called_once_with( 706 self.consistency_group.id) 707 self.assertEqual(self.columns, columns) 708 self.assertItemEqual(self.data, data) 709