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