f23d3ae5e4
Convert all code to not require six library and instead use python 3.x logic. Change-Id: I9301046d6df0bcdcc76808f8a32a67388ddffb95
374 lines
15 KiB
Python
374 lines
15 KiB
Python
# 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()
|