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