1#   Licensed under the Apache License, Version 2.0 (the "License"); you may
2#   not use this file except in compliance with the License. You may obtain
3#   a copy of the License at
4#
5#        http://www.apache.org/licenses/LICENSE-2.0
6#
7#   Unless required by applicable law or agreed to in writing, software
8#   distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
9#   WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
10#   License for the specific language governing permissions and limitations
11#   under the License.
12#
13
14import copy
15from unittest import mock
16
17from osc_lib import exceptions as exc
18
19from heatclient import exc as heat_exc
20from heatclient.osc.v1 import resource
21from heatclient.tests.unit.osc.v1 import fakes as orchestration_fakes
22from heatclient.v1 import resources as v1_resources
23
24
25class TestResource(orchestration_fakes.TestOrchestrationv1):
26    def setUp(self):
27        super(TestResource, self).setUp()
28        self.resource_client = self.app.client_manager.orchestration.resources
29
30
31class TestStackResourceShow(TestResource):
32
33    response = {
34        'attributes': {},
35        'creation_time': '2016-02-01T20:20:53',
36        'description': 'a resource',
37        'links': [
38            {'rel': 'stack',
39             "href": "http://heat.example.com:8004/my_stack/12"}
40        ],
41        'logical_resource_id': 'my_resource',
42        'physical_resource_id': '1234',
43        'required_by': [],
44        'resource_name': 'my_resource',
45        'resource_status': 'CREATE_COMPLETE',
46        'resource_status_reason': 'state changed',
47        'resource_type': 'OS::Heat::None',
48        'updated_time': '2016-02-01T20:20:53',
49    }
50
51    def setUp(self):
52        super(TestStackResourceShow, self).setUp()
53        self.cmd = resource.ResourceShow(self.app, None)
54        self.resource_client.get.return_value = v1_resources.Resource(
55            None, self.response)
56
57    def test_resource_show(self):
58        arglist = ['my_stack', 'my_resource']
59        parsed_args = self.check_parser(self.cmd, arglist, [])
60
61        columns, data = self.cmd.take_action(parsed_args)
62
63        self.resource_client.get.assert_called_with('my_stack', 'my_resource',
64                                                    with_attr=None)
65        for key in self.response:
66            self.assertIn(key, columns)
67            self.assertIn(self.response[key], data)
68
69    def test_resource_show_with_attr(self):
70        arglist = ['my_stack', 'my_resource',
71                   '--with-attr', 'foo', '--with-attr', 'bar']
72        parsed_args = self.check_parser(self.cmd, arglist, [])
73
74        columns, data = self.cmd.take_action(parsed_args)
75
76        self.resource_client.get.assert_called_with('my_stack', 'my_resource',
77                                                    with_attr=['foo', 'bar'])
78        for key in self.response:
79            self.assertIn(key, columns)
80            self.assertIn(self.response[key], data)
81
82    def test_resource_show_not_found(self):
83        arglist = ['my_stack', 'bad_resource']
84        self.resource_client.get.side_effect = heat_exc.HTTPNotFound
85        parsed_args = self.check_parser(self.cmd, arglist, [])
86
87        error = self.assertRaises(exc.CommandError,
88                                  self.cmd.take_action, parsed_args)
89        self.assertEqual('Stack or resource not found: my_stack bad_resource',
90                         str(error))
91
92
93class TestStackResourceList(TestResource):
94
95    response = {
96        'attributes': {},
97        'creation_time': '2016-02-01T20:20:53',
98        'description': 'a resource',
99        'links': [
100            {'rel': 'stack',
101             "href": "http://heat.example.com:8004/my_stack/12"}
102        ],
103        'logical_resource_id': '1234',
104        'physical_resource_id': '1234',
105        'required_by': [],
106        'resource_name': 'my_resource',
107        'resource_status': 'CREATE_COMPLETE',
108        'resource_status_reason': 'state changed',
109        'resource_type': 'OS::Heat::None',
110        'updated_time': '2016-02-01T20:20:53',
111    }
112
113    columns = ['resource_name', 'physical_resource_id', 'resource_type',
114               'resource_status', 'updated_time']
115
116    data = ['my_resource', '1234', 'OS::Heat::None',
117            'CREATE_COMPLETE', '2016-02-01T20:20:53']
118
119    def setUp(self):
120        super(TestStackResourceList, self).setUp()
121        self.cmd = resource.ResourceList(self.app, None)
122        self.resource_client.list.return_value = [
123            v1_resources.Resource(None, self.response)]
124
125    def test_resource_list(self):
126        arglist = ['my_stack']
127        parsed_args = self.check_parser(self.cmd, arglist, [])
128
129        columns, data = self.cmd.take_action(parsed_args)
130
131        self.resource_client.list.assert_called_with(
132            'my_stack',
133            filters={},
134            with_detail=False,
135            nested_depth=None)
136        self.assertEqual(self.columns, columns)
137        self.assertEqual(tuple(self.data), list(data)[0])
138
139    def test_resource_list_not_found(self):
140        arglist = ['bad_stack']
141        self.resource_client.list.side_effect = heat_exc.HTTPNotFound
142        parsed_args = self.check_parser(self.cmd, arglist, [])
143
144        self.assertRaises(exc.CommandError, self.cmd.take_action, parsed_args)
145
146    def test_resource_list_with_detail(self):
147        arglist = ['my_stack', '--long']
148        cols = copy.deepcopy(self.columns)
149        cols.append('stack_name')
150        out = copy.deepcopy(self.data)
151        out.append('my_stack')
152        parsed_args = self.check_parser(self.cmd, arglist, [])
153
154        columns, data = self.cmd.take_action(parsed_args)
155
156        self.resource_client.list.assert_called_with(
157            'my_stack',
158            filters={},
159            with_detail=True,
160            nested_depth=None)
161        self.assertEqual(cols, columns)
162        self.assertEqual(tuple(out), list(data)[0])
163
164    def test_resource_list_nested_depth(self):
165        arglist = ['my_stack', '--nested-depth', '3']
166        cols = copy.deepcopy(self.columns)
167        cols.append('stack_name')
168        out = copy.deepcopy(self.data)
169        out.append('my_stack')
170        parsed_args = self.check_parser(self.cmd, arglist, [])
171
172        columns, data = self.cmd.take_action(parsed_args)
173
174        self.resource_client.list.assert_called_with(
175            'my_stack',
176            filters={},
177            with_detail=False,
178            nested_depth=3)
179        self.assertEqual(cols, columns)
180        self.assertEqual(tuple(out), list(data)[0])
181
182    def test_resource_list_no_resource_name(self):
183        arglist = ['my_stack']
184        resp = copy.deepcopy(self.response)
185        del resp['resource_name']
186        cols = copy.deepcopy(self.columns)
187        cols[0] = 'logical_resource_id'
188        out = copy.deepcopy(self.data)
189        out[1] = '1234'
190        self.resource_client.list.return_value = [
191            v1_resources.Resource(None, resp)]
192        parsed_args = self.check_parser(self.cmd, arglist, [])
193
194        columns, data = self.cmd.take_action(parsed_args)
195
196        self.resource_client.list.assert_called_with(
197            'my_stack',
198            filters={},
199            with_detail=False,
200            nested_depth=None)
201        self.assertEqual(cols, columns)
202
203    def test_resource_list_filter(self):
204        arglist = ['my_stack', '--filter', 'name=my_resource']
205        out = copy.deepcopy(self.data)
206        parsed_args = self.check_parser(self.cmd, arglist, [])
207
208        columns, data = self.cmd.take_action(parsed_args)
209
210        self.resource_client.list.assert_called_with(
211            'my_stack',
212            filters=dict(name='my_resource'),
213            with_detail=False,
214            nested_depth=None)
215        self.assertEqual(tuple(out), list(data)[0])
216
217
218class TestResourceMetadata(TestResource):
219
220    def setUp(self):
221        super(TestResourceMetadata, self).setUp()
222        self.cmd = resource.ResourceMetadata(self.app, None)
223        self.resource_client.metadata.return_value = {}
224
225    def test_resource_metadata(self):
226        arglist = ['my_stack', 'my_resource']
227        parsed_args = self.check_parser(self.cmd, arglist, [])
228        self.cmd.take_action(parsed_args)
229        self.resource_client.metadata.assert_called_with(**{
230            'stack_id': 'my_stack',
231            'resource_name': 'my_resource'
232        })
233
234    def test_resource_metadata_yaml(self):
235        arglist = ['my_stack', 'my_resource', '--format', 'yaml']
236        parsed_args = self.check_parser(self.cmd, arglist, [])
237        self.cmd.take_action(parsed_args)
238        self.resource_client.metadata.assert_called_with(**{
239            'stack_id': 'my_stack',
240            'resource_name': 'my_resource'
241        })
242
243    def test_resource_metadata_error(self):
244        arglist = ['my_stack', 'my_resource']
245        parsed_args = self.check_parser(self.cmd, arglist, [])
246        self.resource_client.metadata.side_effect = heat_exc.HTTPNotFound
247        error = self.assertRaises(exc.CommandError,
248                                  self.cmd.take_action,
249                                  parsed_args)
250        self.assertEqual('Stack my_stack or resource my_resource not found.',
251                         str(error))
252
253
254class TestResourceSignal(TestResource):
255
256    def setUp(self):
257        super(TestResourceSignal, self).setUp()
258        self.cmd = resource.ResourceSignal(self.app, None)
259
260    def test_resource_signal(self):
261        arglist = ['my_stack', 'my_resource']
262        parsed_args = self.check_parser(self.cmd, arglist, [])
263        self.cmd.take_action(parsed_args)
264        self.resource_client.signal.assert_called_with(**{
265            'stack_id': 'my_stack',
266            'resource_name': 'my_resource'
267        })
268
269    def test_resource_signal_error(self):
270        arglist = ['my_stack', 'my_resource']
271        parsed_args = self.check_parser(self.cmd, arglist, [])
272        self.resource_client.signal.side_effect = heat_exc.HTTPNotFound
273        error = self.assertRaises(exc.CommandError,
274                                  self.cmd.take_action,
275                                  parsed_args)
276        self.assertEqual('Stack my_stack or resource my_resource not found.',
277                         str(error))
278
279    def test_resource_signal_data(self):
280        arglist = ['my_stack', 'my_resource',
281                   '--data', '{"message":"Content"}']
282        parsed_args = self.check_parser(self.cmd, arglist, [])
283        self.cmd.take_action(parsed_args)
284        self.resource_client.signal.assert_called_with(**{
285            'data': {u'message': u'Content'},
286            'stack_id': 'my_stack',
287            'resource_name': 'my_resource'
288        })
289
290    def test_resource_signal_data_not_json(self):
291        arglist = ['my_stack', 'my_resource', '--data', '{']
292        parsed_args = self.check_parser(self.cmd, arglist, [])
293        error = self.assertRaises(exc.CommandError,
294                                  self.cmd.take_action,
295                                  parsed_args)
296        self.assertIn('Data should be in JSON format', str(error))
297
298    def test_resource_signal_data_and_file_error(self):
299        arglist = ['my_stack', 'my_resource',
300                   '--data', '{}', '--data-file', 'file']
301        parsed_args = self.check_parser(self.cmd, arglist, [])
302        error = self.assertRaises(exc.CommandError,
303                                  self.cmd.take_action,
304                                  parsed_args)
305        self.assertEqual('Should only specify one of data or data-file',
306                         str(error))
307
308    @mock.patch('six.moves.urllib.request.urlopen')
309    def test_resource_signal_file(self, urlopen):
310        data = mock.Mock()
311        data.read.side_effect = ['{"message":"Content"}']
312        urlopen.return_value = data
313
314        arglist = ['my_stack', 'my_resource', '--data-file', 'test_file']
315        parsed_args = self.check_parser(self.cmd, arglist, [])
316        self.cmd.take_action(parsed_args)
317        self.resource_client.signal.assert_called_with(**{
318            'data': {u'message': u'Content'},
319            'stack_id': 'my_stack',
320            'resource_name': 'my_resource'
321        })
322
323
324class TestResourceMarkUnhealthy(TestResource):
325    def setUp(self):
326        super(TestResourceMarkUnhealthy, self).setUp()
327        self.cmd = resource.ResourceMarkUnhealthy(self.app, None)
328        self.resource_client.mark_unhealthy = mock.Mock()
329
330    def test_resource_mark_unhealthy(self):
331        arglist = ['my_stack', 'my_resource', 'reason']
332        parsed_args = self.check_parser(self.cmd, arglist, [])
333        self.cmd.take_action(parsed_args)
334        self.resource_client.mark_unhealthy.assert_called_with(**{
335            "stack_id": "my_stack",
336            "resource_name": "my_resource",
337            "mark_unhealthy": True,
338            "resource_status_reason": "reason"
339        })
340
341    def test_resource_mark_unhealthy_reset(self):
342        arglist = ['my_stack', 'my_resource', '--reset']
343        parsed_args = self.check_parser(self.cmd, arglist, [])
344        self.cmd.take_action(parsed_args)
345        self.resource_client.mark_unhealthy.assert_called_with(**{
346            "stack_id": "my_stack",
347            "resource_name": "my_resource",
348            "mark_unhealthy": False,
349            "resource_status_reason": ""
350        })
351
352    def test_resource_mark_unhealthy_not_found(self):
353        arglist = ['my_stack', 'my_resource', '--reset']
354        self.resource_client.mark_unhealthy.side_effect = (
355            heat_exc.HTTPNotFound)
356        parsed_args = self.check_parser(self.cmd, arglist, [])
357
358        error = self.assertRaises(exc.CommandError,
359                                  self.cmd.take_action, parsed_args)
360        self.assertEqual('Stack or resource not found: my_stack my_resource',
361                         str(error))
362