Merge "Revert "Remove unittests monasca until story/2002978 is released""

This commit is contained in:
Zuul 2018-07-20 07:42:42 +00:00 committed by Gerrit Code Review
commit 42aab2095a
4 changed files with 586 additions and 0 deletions

View File

@ -0,0 +1,102 @@
#
# 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 mock
import six
import monascaclient
from heat.common import exception as heat_exception
from heat.engine.clients.os import monasca as client_plugin
from heat.tests import common
from heat.tests import utils
class MonascaNotificationConstraintTest(common.HeatTestCase):
def test_expected_exceptions(self):
self.assertEqual(
(heat_exception.EntityNotFound,),
client_plugin.MonascaNotificationConstraint.expected_exceptions,
"MonascaNotificationConstraint expected exceptions error")
def test_constraint(self):
constraint = client_plugin.MonascaNotificationConstraint()
client_mock = mock.MagicMock()
client_plugin_mock = mock.MagicMock()
client_plugin_mock.get_notification.return_value = None
client_mock.client_plugin.return_value = client_plugin_mock
self.assertIsNone(constraint.validate_with_client(client_mock,
'notification_1'))
client_plugin_mock.get_notification.assert_called_once_with(
'notification_1')
class MonascaClientPluginTest(common.HeatTestCase):
def test_client(self):
context = utils.dummy_context()
plugin = context.clients.client_plugin('monasca')
client = plugin.client()
self.assertIsNotNone(client.metrics)
@mock.patch.object(monascaclient.client, '_session')
def test_client_uses_session(self, mock_session):
context = mock.MagicMock()
monasca_client = client_plugin.MonascaClientPlugin(context=context)
self.assertIsNotNone(monasca_client._create())
class MonascaClientPluginNotificationTest(common.HeatTestCase):
sample_uuid = '477e8273-60a7-4c41-b683-fdb0bc7cd152'
sample_name = 'test-notification'
def _get_mock_notification(self):
notification = dict()
notification['id'] = self.sample_uuid
notification['name'] = self.sample_name
return notification
def setUp(self):
super(MonascaClientPluginNotificationTest, self).setUp()
self._client = mock.MagicMock()
self.client_plugin = client_plugin.MonascaClientPlugin(
context=mock.MagicMock()
)
@mock.patch.object(client_plugin.MonascaClientPlugin, 'client')
def test_get_notification(self, client_monasca):
mock_notification = self._get_mock_notification()
self._client.notifications.get.return_value = mock_notification
client_monasca.return_value = self._client
self.assertEqual(self.sample_uuid,
self.client_plugin.get_notification(
self.sample_uuid))
self._client.notifications.get.assert_called_once_with(
notification_id=self.sample_uuid)
@mock.patch.object(client_plugin.MonascaClientPlugin, 'client')
def test_get_notification_not_found(self, client_monasca):
self._client.notifications.get.side_effect = (
client_plugin.monasca_exc.NotFound)
client_monasca.return_value = self._client
ex = self.assertRaises(heat_exception.EntityNotFound,
self.client_plugin.get_notification,
self.sample_uuid)
msg = ("The Monasca Notification (%(name)s) could not be found." %
{'name': self.sample_uuid})
self.assertEqual(msg, six.text_type(ex))
self._client.notifications.get.assert_called_once_with(
notification_id=self.sample_uuid)

View File

View File

@ -0,0 +1,198 @@
#
# 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 mock
from heat.engine.clients.os import monasca as client_plugin
from heat.engine.resources.openstack.monasca import alarm_definition
from heat.engine import stack
from heat.engine import template
from heat.tests import common
from heat.tests import utils
sample_template = {
'heat_template_version': '2015-10-15',
'resources': {
'test_resource': {
'type': 'OS::Monasca::AlarmDefinition',
'properties': {
'name': 'sample_alarm_id',
'description': 'sample alarm def',
'expression': 'sample expression',
'match_by': ['match_by'],
'severity': 'low',
'ok_actions': ['sample_notification'],
'alarm_actions': ['sample_notification'],
'undetermined_actions': ['sample_notification'],
'actions_enabled': False
}
}
}
}
class MonascaAlarmDefinitionTest(common.HeatTestCase):
def setUp(self):
super(MonascaAlarmDefinitionTest, self).setUp()
self.ctx = utils.dummy_context()
self.stack = stack.Stack(
self.ctx, 'test_stack',
template.Template(sample_template)
)
self.test_resource = self.stack['test_resource']
# Mock client
self.test_client = mock.MagicMock()
self.test_resource.client = mock.MagicMock(
return_value=self.test_client)
# Mock client plugin
self.test_client_plugin = client_plugin.MonascaClientPlugin(self.ctx)
self.test_client_plugin._create = mock.MagicMock(
return_value=self.test_client)
self.test_resource.client_plugin = mock.MagicMock(
return_value=self.test_client_plugin)
self.test_client_plugin.get_notification = mock.MagicMock(
return_value='sample_notification')
def _get_mock_resource(self):
value = dict(id='477e8273-60a7-4c41-b683-fdb0bc7cd152')
return value
def test_resource_handle_create(self):
mock_alarm_create = self.test_client.alarm_definitions.create
mock_alarm_patch = self.test_client.alarm_definitions.patch
mock_resource = self._get_mock_resource()
mock_alarm_create.return_value = mock_resource
# validate the properties
self.assertEqual(
'sample_alarm_id',
self.test_resource.properties.get(
alarm_definition.MonascaAlarmDefinition.NAME))
self.assertEqual(
'sample alarm def',
self.test_resource.properties.get(
alarm_definition.MonascaAlarmDefinition.DESCRIPTION))
self.assertEqual(
'sample expression',
self.test_resource.properties.get(
alarm_definition.MonascaAlarmDefinition.EXPRESSION))
self.assertEqual(
['match_by'],
self.test_resource.properties.get(
alarm_definition.MonascaAlarmDefinition.MATCH_BY))
self.assertEqual(
'low',
self.test_resource.properties.get(
alarm_definition.MonascaAlarmDefinition.SEVERITY))
self.assertEqual(
['sample_notification'],
self.test_resource.properties.get(
alarm_definition.MonascaAlarmDefinition.OK_ACTIONS))
self.assertEqual(
['sample_notification'],
self.test_resource.properties.get(
alarm_definition.MonascaAlarmDefinition.ALARM_ACTIONS))
self.assertEqual(
['sample_notification'],
self.test_resource.properties.get(
alarm_definition.MonascaAlarmDefinition.UNDETERMINED_ACTIONS))
self.assertEqual(
False,
self.test_resource.properties.get(
alarm_definition.MonascaAlarmDefinition.ACTIONS_ENABLED))
self.test_resource.data_set = mock.Mock()
self.test_resource.handle_create()
# validate physical resource id
self.assertEqual(mock_resource['id'], self.test_resource.resource_id)
args = dict(
name='sample_alarm_id',
description='sample alarm def',
expression='sample expression',
match_by=['match_by'],
severity='low',
ok_actions=['sample_notification'],
alarm_actions=['sample_notification'],
undetermined_actions=['sample_notification']
)
mock_alarm_create.assert_called_once_with(**args)
mock_alarm_patch.assert_called_once_with(
alarm_id=self.test_resource.resource_id,
actions_enabled=False)
def test_resource_handle_update(self):
mock_alarm_patch = self.test_client.alarm_definitions.patch
self.test_resource.resource_id = '477e8273-60a7-4c41-b683-fdb0bc7cd151'
prop_diff = {
alarm_definition.MonascaAlarmDefinition.NAME:
'name-updated',
alarm_definition.MonascaAlarmDefinition.DESCRIPTION:
'description-updated',
alarm_definition.MonascaAlarmDefinition.ACTIONS_ENABLED:
True,
alarm_definition.MonascaAlarmDefinition.SEVERITY:
'medium',
alarm_definition.MonascaAlarmDefinition.OK_ACTIONS:
['sample_notification'],
alarm_definition.MonascaAlarmDefinition.ALARM_ACTIONS:
['sample_notification'],
alarm_definition.MonascaAlarmDefinition.UNDETERMINED_ACTIONS:
['sample_notification']}
self.test_resource.handle_update(json_snippet=None,
tmpl_diff=None,
prop_diff=prop_diff)
args = dict(
alarm_id=self.test_resource.resource_id,
name='name-updated',
description='description-updated',
actions_enabled=True,
severity='medium',
ok_actions=['sample_notification'],
alarm_actions=['sample_notification'],
undetermined_actions=['sample_notification']
)
mock_alarm_patch.assert_called_once_with(**args)
def test_resource_handle_delete(self):
mock_alarm_delete = self.test_client.alarm_definitions.delete
self.test_resource.resource_id = '477e8273-60a7-4c41-b683-fdb0bc7cd151'
mock_alarm_delete.return_value = None
self.assertIsNone(self.test_resource.handle_delete())
mock_alarm_delete.assert_called_once_with(
alarm_id=self.test_resource.resource_id
)
def test_resource_handle_delete_resource_id_is_none(self):
self.test_resource.resource_id = None
self.assertIsNone(self.test_resource.handle_delete())
def test_resource_handle_delete_not_found(self):
self.test_resource.resource_id = '477e8273-60a7-4c41-b683-fdb0bc7cd151'
mock_alarm_delete = self.test_client.alarm_definitions.delete
mock_alarm_delete.side_effect = client_plugin.monasca_exc.NotFound
self.assertIsNone(self.test_resource.handle_delete())

View File

@ -0,0 +1,286 @@
#
# 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 mock
import six
from heat.common import exception
from heat.engine.cfn import functions as cfn_funcs
from heat.engine.clients.os import monasca as client_plugin
from heat.engine.resources.openstack.monasca import notification
from heat.engine import stack
from heat.engine import template
from heat.tests import common
from heat.tests import utils
sample_template = {
'heat_template_version': '2015-10-15',
'resources': {
'test_resource': {
'type': 'OS::Monasca::Notification',
'properties': {
'name': 'test-notification',
'type': 'webhook',
'address': 'http://localhost:80/',
'period': 60
}
}
}
}
class MonascaNotificationTest(common.HeatTestCase):
def setUp(self):
super(MonascaNotificationTest, self).setUp()
self.ctx = utils.dummy_context()
self.stack = stack.Stack(
self.ctx, 'test_stack',
template.Template(sample_template)
)
self.test_resource = self.stack['test_resource']
# Mock client
self.test_client = mock.MagicMock()
self.test_resource.client = mock.MagicMock(
return_value=self.test_client)
# Mock client plugin
self.test_client_plugin = client_plugin.MonascaClientPlugin(self.ctx)
self.test_client_plugin._create = mock.MagicMock(
return_value=self.test_client)
self.test_resource.client_plugin = mock.MagicMock(
return_value=self.test_client_plugin)
def _get_mock_resource(self):
value = dict(id='477e8273-60a7-4c41-b683-fdb0bc7cd152')
return value
def test_validate_success_no_period(self):
self.test_resource.properties.data.pop('period')
self.test_resource.validate()
def test_validate_invalid_type_with_period(self):
self.test_resource.properties.data['type'] = self.test_resource.EMAIL
self.assertRaises(exception.StackValidationFailed,
self.test_resource.validate)
def test_validate_no_scheme_address_for_get_attr(self):
self.test_resource.properties.data['type'] = self.test_resource.WEBHOOK
self.patchobject(cfn_funcs, 'GetAtt', return_value=None)
get_att = cfn_funcs.GetAtt(self.stack, 'Fn::GetAtt',
["ResourceA", "abc"])
self.test_resource.properties.data['address'] = get_att
self.assertIsNone(self.test_resource.validate())
def test_validate_no_scheme_address_for_webhook(self):
self.test_resource.properties.data['type'] = self.test_resource.WEBHOOK
self.test_resource.properties.data['address'] = 'abc@def.com'
ex = self.assertRaises(exception.StackValidationFailed,
self.test_resource.validate)
self.assertEqual('Address "abc@def.com" doesn\'t have '
'required URL scheme', six.text_type(ex))
def test_validate_no_netloc_address_for_webhook(self):
self.test_resource.properties.data['type'] = self.test_resource.WEBHOOK
self.test_resource.properties.data['address'] = 'https://'
ex = self.assertRaises(exception.StackValidationFailed,
self.test_resource.validate)
self.assertEqual('Address "https://" doesn\'t have '
'required network location', six.text_type(ex))
def test_validate_prohibited_address_for_webhook(self):
self.test_resource.properties.data['type'] = self.test_resource.WEBHOOK
self.test_resource.properties.data['address'] = 'ftp://127.0.0.1'
ex = self.assertRaises(exception.StackValidationFailed,
self.test_resource.validate)
self.assertEqual('Address "ftp://127.0.0.1" doesn\'t satisfies '
'allowed schemes: http, https', six.text_type(ex))
def test_validate_incorrect_address_for_email(self):
self.test_resource.properties.data['type'] = self.test_resource.EMAIL
self.test_resource.properties.data['address'] = 'abc#def.com'
self.test_resource.properties.data.pop('period')
ex = self.assertRaises(exception.StackValidationFailed,
self.test_resource.validate)
self.assertEqual('Address "abc#def.com" doesn\'t satisfies allowed '
'format for "email" type of "type" property',
six.text_type(ex))
def test_validate_invalid_address_parsing(self):
self.test_resource.properties.data['type'] = self.test_resource.WEBHOOK
self.test_resource.properties.data['address'] = "https://example.com]"
ex = self.assertRaises(exception.StackValidationFailed,
self.test_resource.validate)
self.assertEqual('Address "https://example.com]" should have correct '
'format required by "webhook" type of "type" '
'property', six.text_type(ex))
def test_resource_handle_create(self):
mock_notification_create = self.test_client.notifications.create
mock_resource = self._get_mock_resource()
mock_notification_create.return_value = mock_resource
# validate the properties
self.assertEqual(
'test-notification',
self.test_resource.properties.get(
notification.MonascaNotification.NAME))
self.assertEqual(
'webhook',
self.test_resource.properties.get(
notification.MonascaNotification.TYPE))
self.assertEqual(
'http://localhost:80/',
self.test_resource.properties.get(
notification.MonascaNotification.ADDRESS))
self.assertEqual(
60, self.test_resource.properties.get(
notification.MonascaNotification.PERIOD))
self.test_resource.data_set = mock.Mock()
self.test_resource.handle_create()
args = dict(
name='test-notification',
type='webhook',
address='http://localhost:80/',
period=60
)
mock_notification_create.assert_called_once_with(**args)
# validate physical resource id
self.assertEqual(mock_resource['id'], self.test_resource.resource_id)
def test_resource_handle_create_default_period(self):
self.test_resource.properties.data.pop('period')
mock_notification_create = self.test_client.notifications.create
self.test_resource.handle_create()
args = dict(
name='test-notification',
type='webhook',
address='http://localhost:80/',
period=60
)
mock_notification_create.assert_called_once_with(**args)
def test_resource_handle_create_no_period(self):
self.test_resource.properties.data.pop('period')
self.test_resource.properties.data['type'] = 'email'
self.test_resource.properties.data['address'] = 'abc@def.com'
mock_notification_create = self.test_client.notifications.create
self.test_resource.handle_create()
args = dict(
name='test-notification',
type='email',
address='abc@def.com'
)
mock_notification_create.assert_called_once_with(**args)
def test_resource_handle_update(self):
mock_notification_update = self.test_client.notifications.update
self.test_resource.resource_id = '477e8273-60a7-4c41-b683-fdb0bc7cd151'
prop_diff = {notification.MonascaNotification.ADDRESS:
'http://localhost:1234/',
notification.MonascaNotification.NAME: 'name-updated',
notification.MonascaNotification.TYPE: 'webhook',
notification.MonascaNotification.PERIOD: 0}
self.test_resource.handle_update(json_snippet=None,
tmpl_diff=None,
prop_diff=prop_diff)
args = dict(
notification_id=self.test_resource.resource_id,
name='name-updated',
type='webhook',
address='http://localhost:1234/',
period=0
)
mock_notification_update.assert_called_once_with(**args)
def test_resource_handle_update_default_period(self):
mock_notification_update = self.test_client.notifications.update
self.test_resource.resource_id = '477e8273-60a7-4c41-b683-fdb0bc7cd151'
self.test_resource.properties.data.pop('period')
prop_diff = {notification.MonascaNotification.ADDRESS:
'http://localhost:1234/',
notification.MonascaNotification.NAME: 'name-updated',
notification.MonascaNotification.TYPE: 'webhook'}
self.test_resource.handle_update(json_snippet=None,
tmpl_diff=None,
prop_diff=prop_diff)
args = dict(
notification_id=self.test_resource.resource_id,
name='name-updated',
type='webhook',
address='http://localhost:1234/',
period=60
)
mock_notification_update.assert_called_once_with(**args)
def test_resource_handle_update_no_period(self):
mock_notification_update = self.test_client.notifications.update
self.test_resource.resource_id = '477e8273-60a7-4c41-b683-fdb0bc7cd151'
self.test_resource.properties.data.pop('period')
prop_diff = {notification.MonascaNotification.ADDRESS:
'abc@def.com',
notification.MonascaNotification.NAME: 'name-updated',
notification.MonascaNotification.TYPE: 'email'}
self.test_resource.handle_update(json_snippet=None,
tmpl_diff=None,
prop_diff=prop_diff)
args = dict(
notification_id=self.test_resource.resource_id,
name='name-updated',
type='email',
address='abc@def.com'
)
mock_notification_update.assert_called_once_with(**args)
def test_resource_handle_delete(self):
mock_notification_delete = self.test_client.notifications.delete
self.test_resource.resource_id = '477e8273-60a7-4c41-b683-fdb0bc7cd151'
mock_notification_delete.return_value = None
self.assertIsNone(self.test_resource.handle_delete())
mock_notification_delete.assert_called_once_with(
notification_id=self.test_resource.resource_id
)
def test_resource_handle_delete_resource_id_is_none(self):
self.test_resource.resource_id = None
self.assertIsNone(self.test_resource.handle_delete())
def test_resource_handle_delete_not_found(self):
self.test_resource.resource_id = '477e8273-60a7-4c41-b683-fdb0bc7cd151'
mock_notification_delete = self.test_client.notifications.delete
mock_notification_delete.side_effect = (
client_plugin.monasca_exc.NotFound)
self.assertIsNone(self.test_resource.handle_delete())