# Licensed under the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. You may obtain # a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. import copy import io from unittest import mock from openstack import exceptions as sdk_exc from osc_lib import exceptions as exc from osc_lib import utils from senlinclient.common.i18n import _ from senlinclient.tests.unit.v1 import fakes from senlinclient.v1 import receiver as osc_receiver class TestReceiver(fakes.TestClusteringv1): def setUp(self): super(TestReceiver, self).setUp() self.mock_client = self.app.client_manager.clustering class TestReceiverList(TestReceiver): columns = ['id', 'name', 'type', 'cluster_id', 'action', 'created_at'] defaults = { 'global_project': False, 'marker': None, 'limit': None, 'sort': None, } def setUp(self): super(TestReceiverList, self).setUp() self.cmd = osc_receiver.ListReceiver(self.app, None) fake_receiver = mock.Mock( action="CLUSTER_SCALE_OUT", actor={}, channel={ "alarm_url": "http://node1:8778/v1/webhooks/e03dd2e5-8f2e-4ec1" "-8c6a-74ba891e5422/trigger?V=1&count=1" }, cluster_id="ae63a10b-4a90-452c-aef1-113a0b255ee3", created_at="2015-06-27T05:09:43", domain_id="Default", id="573aa1ba-bf45-49fd-907d-6b5d6e6adfd3", name="cluster_inflate", params={"count": "1"}, project_id="6e18cc2bdbeb48a5b3cad2dc499f6804", type="webhook", updated_at=None, user_id="b4ad2d6e18cc2b9c48049f6dbe8a5b3c" ) fake_receiver.to_dict = mock.Mock(return_value={}) self.mock_client.receivers = mock.Mock(return_value=[fake_receiver]) def test_receiver_list_defaults(self): arglist = [] parsed_args = self.check_parser(self.cmd, arglist, []) columns, data = self.cmd.take_action(parsed_args) self.mock_client.receivers.assert_called_with(**self.defaults) self.assertEqual(self.columns, columns) def test_receiver_list_full_id(self): arglist = ['--full-id'] parsed_args = self.check_parser(self.cmd, arglist, []) columns, data = self.cmd.take_action(parsed_args) self.mock_client.receivers.assert_called_with(**self.defaults) self.assertEqual(self.columns, columns) def test_receiver_list_limit(self): kwargs = copy.deepcopy(self.defaults) kwargs['limit'] = '3' arglist = ['--limit', '3'] parsed_args = self.check_parser(self.cmd, arglist, []) columns, data = self.cmd.take_action(parsed_args) self.mock_client.receivers.assert_called_with(**kwargs) self.assertEqual(self.columns, columns) def test_receiver_list_sort(self): kwargs = copy.deepcopy(self.defaults) kwargs['sort'] = 'name:asc' arglist = ['--sort', 'name:asc'] parsed_args = self.check_parser(self.cmd, arglist, []) columns, data = self.cmd.take_action(parsed_args) self.mock_client.receivers.assert_called_with(**kwargs) self.assertEqual(self.columns, columns) def test_receiver_list_sort_invalid_key(self): kwargs = copy.deepcopy(self.defaults) kwargs['sort'] = 'bad_key' arglist = ['--sort', 'bad_key'] parsed_args = self.check_parser(self.cmd, arglist, []) self.mock_client.receivers.side_effect = sdk_exc.HttpException() self.assertRaises(sdk_exc.HttpException, self.cmd.take_action, parsed_args) def test_receiver_list_sort_invalid_direction(self): kwargs = copy.deepcopy(self.defaults) kwargs['sort'] = 'name:bad_direction' arglist = ['--sort', 'name:bad_direction'] parsed_args = self.check_parser(self.cmd, arglist, []) self.mock_client.receivers.side_effect = sdk_exc.HttpException() self.assertRaises(sdk_exc.HttpException, self.cmd.take_action, parsed_args) def test_receiver_list_filter(self): kwargs = copy.deepcopy(self.defaults) kwargs['name'] = 'my_receiver' arglist = ['--filter', 'name=my_receiver'] parsed_args = self.check_parser(self.cmd, arglist, []) columns, data = self.cmd.take_action(parsed_args) self.mock_client.receivers.assert_called_with(**kwargs) self.assertEqual(self.columns, columns) def test_receiver_list_marker(self): kwargs = copy.deepcopy(self.defaults) kwargs['marker'] = 'a9448bf6' arglist = ['--marker', 'a9448bf6'] parsed_args = self.check_parser(self.cmd, arglist, []) columns, data = self.cmd.take_action(parsed_args) self.mock_client.receivers.assert_called_with(**kwargs) self.assertEqual(self.columns, columns) class TestReceiverShow(TestReceiver): def setUp(self): super(TestReceiverShow, self).setUp() self.cmd = osc_receiver.ShowReceiver(self.app, None) fake_receiver = mock.Mock( action="CLUSTER_SCALE_OUT", actor={}, channel={ "alarm_url": "http://node1:8778/v1/webhooks/e03dd2e5-8f2e-4ec1" "-8c6a-74ba891e5422/trigger?V=1&count=1" }, cluster_id="ae63a10b-4a90-452c-aef1-113a0b255ee3", created_at="2015-06-27T05:09:43", domain_id="Default", id="573aa1ba-bf45-49fd-907d-6b5d6e6adfd3", name="cluster_inflate", params={"count": "1"}, project_id="6e18cc2bdbeb48a5b3cad2dc499f6804", type="webhook", updated_at=None, user_id="b4ad2d6e18cc2b9c48049f6dbe8a5b3c" ) fake_receiver.to_dict = mock.Mock(return_value={}) self.mock_client.get_receiver = mock.Mock(return_value=fake_receiver) def test_receiver_show(self): arglist = ['my_receiver'] parsed_args = self.check_parser(self.cmd, arglist, []) self.cmd.take_action(parsed_args) self.mock_client.get_receiver.assert_called_with('my_receiver') def test_receiver_show_not_found(self): arglist = ['my_receiver'] parsed_args = self.check_parser(self.cmd, arglist, []) self.mock_client.get_receiver.side_effect = sdk_exc.ResourceNotFound() error = self.assertRaises(exc.CommandError, self.cmd.take_action, parsed_args) self.assertEqual('Receiver not found: my_receiver', str(error)) class TestReceiverCreate(TestReceiver): args = { "action": "CLUSTER_SCALE_OUT", "cluster_id": "my_cluster", "name": "my_receiver", "params": { "count": "1" }, "type": "webhook" } def setUp(self): super(TestReceiverCreate, self).setUp() self.cmd = osc_receiver.CreateReceiver(self.app, None) fake_receiver = mock.Mock( action="CLUSTER_SCALE_OUT", actor={}, channel={ "alarm_url": "http://node1:8778/v1/webhooks/e03dd2e5-8f2e-4ec1" "-8c6a-74ba891e5422/trigger?V=1&count=1" }, cluster_id="ae63a10b-4a90-452c-aef1-113a0b255ee3", created_at="2015-06-27T05:09:43", domain_id="Default", id="573aa1ba-bf45-49fd-907d-6b5d6e6adfd3", name="cluster_inflate", params={"count": "1"}, project_id="6e18cc2bdbeb48a5b3cad2dc499f6804", type="webhook", updated_at=None, user_id="b4ad2d6e18cc2b9c48049f6dbe8a5b3c" ) fake_receiver.to_dict = mock.Mock(return_value={}) self.mock_client.create_receiver = mock.Mock( return_value=fake_receiver) self.mock_client.get_receiver = mock.Mock( return_value=fake_receiver) def test_receiver_create_webhook(self): arglist = ['my_receiver', '--action', 'CLUSTER_SCALE_OUT', '--cluster', 'my_cluster', '--params', 'count=1', '--type', 'webhook'] parsed_args = self.check_parser(self.cmd, arglist, []) self.cmd.take_action(parsed_args) self.mock_client.create_receiver.assert_called_with(**self.args) def test_receiver_create_webhook_failed(self): arglist = ['my_receiver', '--action', 'CLUSTER_SCALE_OUT', '--params', 'count=1', '--type', 'webhook'] parsed_args = self.check_parser(self.cmd, arglist, []) error = self.assertRaises(exc.CommandError, self.cmd.take_action, parsed_args) self.assertIn(_('cluster and action parameters are required to create ' 'webhook type of receiver'), str(error)) def test_receiver_create_non_webhook(self): arglist = ['my_receiver', '--params', 'count=1', '--type', 'foo'] parsed_args = self.check_parser(self.cmd, arglist, []) self.cmd.take_action(parsed_args) args = copy.deepcopy(self.args) args['type'] = 'foo' args['cluster_id'] = None args['action'] = None self.mock_client.create_receiver.assert_called_with(**args) class TestReceiverUpdate(TestReceiver): args = { "action": "CLUSTER_SCALE_OUT", "name": "test_receiver", "params": { "count": "2" }, } def setUp(self): super(TestReceiverUpdate, self).setUp() self.cmd = osc_receiver.UpdateReceiver(self.app, None) fake_receiver = mock.Mock( action="CLUSTER_SCALE_IN", actor={}, channel={ "alarm_url": "http://node1:8778/v1/webhooks/e03dd2e5-8f2e-4ec1" "-8c6a-74ba891e5422/trigger?V=1&count=1" }, created_at="2015-06-27T05:09:43", domain_id="Default", id="573aa1ba-bf45-49fd-907d-6b5d6e6adfd3", params={"count": "1"}, project_id="6e18cc2bdbeb48a5b3cad2dc499f6804", updated_at=None, user_id="b4ad2d6e18cc2b9c48049f6dbe8a5b3c" ) fake_receiver.name = "cluster_inflate" fake_receiver.action = "CLUSTER_SCALE_IN" fake_receiver.params = {"count": "1"} fake_receiver.to_dict = mock.Mock(return_value={}) self.mock_client.update_receiver = mock.Mock( return_value=fake_receiver) self.mock_client.get_receiver = mock.Mock(return_value=fake_receiver) self.mock_client.find_receiver = mock.Mock(return_value=fake_receiver) utils.get_dict_properties = mock.Mock(return_value='') def test_receiver_update_defaults(self): arglist = ['--name', 'test_receiver', '--action', 'CLUSTER_SCALE_OUT', '--params', 'count=2', '573aa1ba'] parsed_args = self.check_parser(self.cmd, arglist, []) self.cmd.take_action(parsed_args) self.mock_client.update_receiver.assert_called_with( "573aa1ba-bf45-49fd-907d-6b5d6e6adfd3", **self.args) def test_receiver_update_not_found(self): arglist = ['--name', 'test_receiver', '--action', 'CLUSTER_SCALE_OUT', '--params', 'count=2', '573aa1b2'] parsed_args = self.check_parser(self.cmd, arglist, []) self.mock_client.find_receiver.return_value = None error = self.assertRaises(exc.CommandError, self.cmd.take_action, parsed_args) self.assertIn('Receiver not found: 573aa1b2', str(error)) class TestReceiverDelete(TestReceiver): def setUp(self): super(TestReceiverDelete, self).setUp() self.cmd = osc_receiver.DeleteReceiver(self.app, None) self.mock_client.delete_receiver = mock.Mock() def test_receiver_delete(self): arglist = ['receiver1', 'receiver2', 'receiver3'] parsed_args = self.check_parser(self.cmd, arglist, []) self.cmd.take_action(parsed_args) self.mock_client.delete_receiver.assert_has_calls( [mock.call('receiver1', False), mock.call('receiver2', False), mock.call('receiver3', False)] ) def test_receiver_delete_force(self): arglist = ['receiver1', 'receiver2', 'receiver3', '--force'] parsed_args = self.check_parser(self.cmd, arglist, []) self.cmd.take_action(parsed_args) self.mock_client.delete_receiver.assert_has_calls( [mock.call('receiver1', False), mock.call('receiver2', False), mock.call('receiver3', False)] ) def test_receiver_delete_not_found(self): arglist = ['my_receiver'] self.mock_client.delete_receiver.side_effect = ( sdk_exc.ResourceNotFound) parsed_args = self.check_parser(self.cmd, arglist, []) error = self.assertRaises(exc.CommandError, self.cmd.take_action, parsed_args) self.assertIn('Failed to delete 1 of the 1 specified receiver(s).', str(error)) def test_receiver_delete_one_found_one_not_found(self): arglist = ['receiver1', 'receiver2'] self.mock_client.delete_receiver.side_effect = ( [None, sdk_exc.ResourceNotFound] ) parsed_args = self.check_parser(self.cmd, arglist, []) error = self.assertRaises(exc.CommandError, self.cmd.take_action, parsed_args) self.mock_client.delete_receiver.assert_has_calls( [mock.call('receiver1', False), mock.call('receiver2', False)] ) self.assertEqual('Failed to delete 1 of the 2 specified receiver(s).', str(error)) @mock.patch('sys.stdin', spec=io.StringIO) def test_receiver_delete_prompt_yes(self, mock_stdin): arglist = ['my_receiver'] mock_stdin.isatty.return_value = True mock_stdin.readline.return_value = 'y' parsed_args = self.check_parser(self.cmd, arglist, []) self.cmd.take_action(parsed_args) mock_stdin.readline.assert_called_with() self.mock_client.delete_receiver.assert_called_with('my_receiver', False) @mock.patch('sys.stdin', spec=io.StringIO) def test_receiver_delete_prompt_no(self, mock_stdin): arglist = ['my_receiver'] mock_stdin.isatty.return_value = True mock_stdin.readline.return_value = 'n' parsed_args = self.check_parser(self.cmd, arglist, []) self.cmd.take_action(parsed_args) mock_stdin.readline.assert_called_with() self.mock_client.delete_receiver.assert_not_called()