Merge "Add unit test"
This commit is contained in:
Zuul
Gerrit Code Review
commit
fb23e914b1
|
|
@ -24,6 +24,7 @@ from oslo_config import fixture as oo_cfg
|
|||
from oslo_context import fixture as oo_ctx
|
||||
from oslo_serialization import jsonutils
|
||||
from oslotest import base as oslotest_base
|
||||
import testtools.matchers as matchers
|
||||
|
||||
from monasca_api.api.core import request
|
||||
from monasca_api import conf
|
||||
|
|
@ -131,3 +132,27 @@ class PolicyFixture(fixtures.Fixture):
|
|||
for rule in policies.list_rules():
|
||||
if rule.name not in rules:
|
||||
rules[rule.name] = rule.check_str
|
||||
|
||||
|
||||
class RESTResponseEquals(object):
|
||||
"""Match if the supplied data contains a single string containing a JSON
|
||||
object which decodes to match expected_data, excluding the contents of
|
||||
the 'links' key.
|
||||
"""
|
||||
|
||||
def __init__(self, expected_data):
|
||||
self.expected_data = expected_data
|
||||
|
||||
if u"links" in expected_data:
|
||||
del expected_data[u"links"]
|
||||
|
||||
def __str__(self):
|
||||
return 'RESTResponseEquals(%s)' % (self.expected_data,)
|
||||
|
||||
def match(self, actual):
|
||||
response_data = actual.json
|
||||
|
||||
if u"links" in response_data:
|
||||
del response_data[u"links"]
|
||||
|
||||
return matchers.Equals(self.expected_data).match(response_data)
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@
|
|||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
from falcon import errors
|
||||
from falcon import testing
|
||||
|
||||
from monasca_common.policy import policy_engine as policy
|
||||
|
|
@ -21,6 +22,55 @@ from monasca_api.api.core import request
|
|||
from monasca_api.tests import base
|
||||
import monasca_api.v2.reference.helpers as helpers
|
||||
|
||||
from monasca_common.rest import utils as rest_utils
|
||||
|
||||
|
||||
class TestHelpersFunction(base.BaseTestCase):
|
||||
|
||||
def test_from_json(self):
|
||||
body_json = {'test_body': 'test'}
|
||||
req = request.Request(
|
||||
testing.create_environ(
|
||||
body=rest_utils.as_json(body_json),
|
||||
)
|
||||
)
|
||||
response = helpers.from_json(req)
|
||||
self.assertEqual(body_json, response)
|
||||
|
||||
def test_from_json_incorrect_message(self):
|
||||
req = request.Request(
|
||||
testing.create_environ(
|
||||
body='incorrect message',
|
||||
)
|
||||
)
|
||||
self.assertRaises(errors.HTTPBadRequest, helpers.from_json, req)
|
||||
|
||||
def test_to_json(self):
|
||||
test_dict = {'test_body': 'test'}
|
||||
expected_json = '{"test_body":"test"}'
|
||||
response = helpers.to_json(test_dict)
|
||||
self.assertEqual(expected_json, response)
|
||||
|
||||
def test_validate_json_content_type(self):
|
||||
req = request.Request(
|
||||
testing.create_environ(
|
||||
headers={'Content-Type': 'application/json'}
|
||||
)
|
||||
)
|
||||
helpers.validate_json_content_type(req)
|
||||
|
||||
def test_validate_json_content_type_incorrect_content_type(self):
|
||||
req = request.Request(
|
||||
testing.create_environ(
|
||||
headers={'Content-Type': 'multipart/form-data'}
|
||||
)
|
||||
)
|
||||
self.assertRaises(errors.HTTPBadRequest, helpers.validate_json_content_type, req)
|
||||
|
||||
def test_validate_json_content_type_missing_content_type(self):
|
||||
req = request.Request(testing.create_environ())
|
||||
self.assertRaises(errors.HTTPBadRequest, helpers.validate_json_content_type, req)
|
||||
|
||||
|
||||
class TestGetXTenantOrTenantId(base.BaseApiTestCase):
|
||||
def setUp(self):
|
||||
|
|
|
|||
|
|
@ -0,0 +1,325 @@
|
|||
# Copyright 2019 FUJITSU LIMITED
|
||||
#
|
||||
# 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 json
|
||||
|
||||
import falcon
|
||||
import fixtures
|
||||
from oslo_config import cfg
|
||||
|
||||
from monasca_api.tests import base
|
||||
from monasca_api.v2.reference import metrics
|
||||
|
||||
|
||||
CONF = cfg.CONF
|
||||
|
||||
TENANT_ID = u"fedcba9876543210fedcba9876543210"
|
||||
|
||||
|
||||
class TestMetrics(base.BaseApiTestCase):
|
||||
def setUp(self):
|
||||
super(TestMetrics, self).setUp()
|
||||
|
||||
self.useFixture(fixtures.MockPatch(
|
||||
'monasca_api.common.messaging.kafka_publisher.KafkaPublisher'
|
||||
))
|
||||
self.metrics_repo_mock = self.useFixture(fixtures.MockPatch(
|
||||
'monasca_api.common.repositories.influxdb.metrics_repository.MetricsRepository'
|
||||
)).mock
|
||||
|
||||
# [messaging]
|
||||
self.conf_override(
|
||||
driver='monasca_api.common.messaging.'
|
||||
'kafka_publisher:KafkaPublisher',
|
||||
group='messaging')
|
||||
|
||||
self.metrics_resource = metrics.Metrics()
|
||||
|
||||
self.app.add_route('/v2.0/metrics',
|
||||
self.metrics_resource)
|
||||
|
||||
def test_list_metrics(self):
|
||||
expected_elements = \
|
||||
{'elements': [{'id': '0',
|
||||
'name': 'mon.fake_metric',
|
||||
'dimensions':
|
||||
{'hostname': 'host0',
|
||||
'db': 'vegeta'}},
|
||||
{'id': '1',
|
||||
'name': 'cpu.idle_perc',
|
||||
'dimensions':
|
||||
{'hostname': 'host0',
|
||||
'db': 'vegeta'}}
|
||||
]}
|
||||
|
||||
return_value = self.metrics_repo_mock.return_value
|
||||
return_value.list_metrics.return_value = expected_elements['elements']
|
||||
|
||||
response = self.simulate_request(path='/v2.0/metrics',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
method='GET')
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_elements))
|
||||
|
||||
def test_send_metrics(self):
|
||||
request_body = {
|
||||
"name": "mon.fake_metric",
|
||||
"dimensions": {
|
||||
"hostname": "host0",
|
||||
"db": "vegeta"
|
||||
},
|
||||
"timestamp": 1405630174123,
|
||||
"value": 1.0,
|
||||
"value_meta": {
|
||||
"key1": "value1",
|
||||
"key2": "value2"
|
||||
}}
|
||||
response = self.simulate_request(path='/v2.0/metrics',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID,
|
||||
'Content-Type': 'application/json'},
|
||||
body=json.dumps(request_body),
|
||||
method='POST')
|
||||
self.assertEqual(falcon.HTTP_204, response.status)
|
||||
|
||||
def test_send_incorrect_metric(self):
|
||||
request_body = {
|
||||
"name": "mon.fake_metric",
|
||||
"dimensions": 'oh no',
|
||||
"timestamp": 1405630174123,
|
||||
"value": 1.0}
|
||||
response = self.simulate_request(path='/v2.0/metrics',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID,
|
||||
'Content-Type': 'application/json'},
|
||||
body=json.dumps(request_body),
|
||||
method='POST')
|
||||
self.assertEqual(falcon.HTTP_422, response.status)
|
||||
|
||||
|
||||
class TestMeasurements(base.BaseApiTestCase):
|
||||
def setUp(self):
|
||||
super(TestMeasurements, self).setUp()
|
||||
|
||||
self.metrics_repo_mock = self.useFixture(fixtures.MockPatch(
|
||||
'monasca_api.common.repositories.influxdb.metrics_repository.MetricsRepository'
|
||||
)).mock
|
||||
|
||||
self.metrics_resource = metrics.MetricsMeasurements()
|
||||
|
||||
self.app.add_route('/v2.0/metrics/measurements',
|
||||
self.metrics_resource)
|
||||
|
||||
def test_get_measurements(self):
|
||||
expected_measurements = \
|
||||
{'elements': [
|
||||
{u'name': u'mon.fake_metric',
|
||||
u'columns': [u'timestamp',
|
||||
u'value',
|
||||
u'value_meta'],
|
||||
u'id': '0',
|
||||
u'dimensions': {
|
||||
u'hostname': u'devstack',
|
||||
u'service': u'monitoring'},
|
||||
u'measurements':
|
||||
[[u'2019-03-12T12:37:10.106Z', 98.5, {}],
|
||||
[u'2019-03-12T12:37:23.012Z', 98.8, {}],
|
||||
[u'2019-03-12T12:37:38.031Z', 68.7, {}],
|
||||
[u'2019-03-12T12:37:53.046Z', 55.3, {}],
|
||||
[u'2019-03-12T12:38:08.048Z', 52.8, {}]]}]}
|
||||
|
||||
return_value = self.metrics_repo_mock.return_value
|
||||
return_value.measurement_list.return_value = expected_measurements['elements']
|
||||
response = self.simulate_request(path='/v2.0/metrics/measurements',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
method='GET',
|
||||
query_string='name=mon.fake_metric&'
|
||||
'start_time=2015-03-01T00:00:01.000Z')
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_measurements))
|
||||
|
||||
def test_get_measurements_invalid_metric_name(self):
|
||||
response = self.simulate_request(path='/v2.0/metrics/measurements',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
method='GET',
|
||||
query_string='name=z'
|
||||
'start_time=2015-03-01T00:00:01.000Z')
|
||||
self.assertEqual(falcon.HTTP_422, response.status)
|
||||
|
||||
def test_get_measurements_missing_start_time(self):
|
||||
response = self.simulate_request(path='/v2.0/metrics/measurements',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
method='GET',
|
||||
query_string='name=mon.fake_metric')
|
||||
self.assertEqual(falcon.HTTP_422, response.status)
|
||||
|
||||
def test_get_measurements_missing_name(self):
|
||||
response = self.simulate_request(path='/v2.0/metrics/measurements',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
method='GET',
|
||||
query_string='start_time=2015-03-01T00:00:01.000Z')
|
||||
self.assertEqual(falcon.HTTP_422, response.status)
|
||||
|
||||
|
||||
class TestStatistics(base.BaseApiTestCase):
|
||||
def setUp(self):
|
||||
super(TestStatistics, self).setUp()
|
||||
|
||||
self.metrics_repo_mock = self.useFixture(fixtures.MockPatch(
|
||||
'monasca_api.common.repositories.influxdb.metrics_repository.MetricsRepository'
|
||||
)).mock
|
||||
|
||||
self.metrics_resource = metrics.MetricsStatistics()
|
||||
|
||||
self.app.add_route('/v2.0/metrics/statistics',
|
||||
self.metrics_resource)
|
||||
|
||||
def test_get_statistics(self):
|
||||
expected_statistics = \
|
||||
{u'elements': [{u'name': u'mon.fake_metric',
|
||||
u'columns':
|
||||
[u'timestamp',
|
||||
u'avg'],
|
||||
u'id': '0',
|
||||
u'dimensions':
|
||||
{u'hostname': u'devstack',
|
||||
u'service': u'monitoring'},
|
||||
u'statistics':
|
||||
[[u'2019-03-12T12:35:00Z', 49.25],
|
||||
[u'2019-03-12T12:40:00Z', 28.25],
|
||||
[u'2019-03-12T12:45:00Z', 27.5],
|
||||
[u'2019-03-12T12:50:00Z', 27],
|
||||
[u'2019-03-12T12:55:00Z', 28]]}]}
|
||||
return_value = self.metrics_repo_mock.return_value
|
||||
return_value.metrics_statistics.return_value = expected_statistics['elements']
|
||||
response = self.simulate_request(path='/v2.0/metrics/statistics',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
method='GET',
|
||||
query_string='name=mon.fake_metric&'
|
||||
'start_time=2015-03-01T00:00:01.000Z&'
|
||||
'statistics=avg')
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_statistics))
|
||||
|
||||
|
||||
class TestMetricsNames(base.BaseApiTestCase):
|
||||
def setUp(self):
|
||||
super(TestMetricsNames, self).setUp()
|
||||
|
||||
self.metrics_repo_mock = self.useFixture(fixtures.MockPatch(
|
||||
'monasca_api.common.repositories.influxdb.metrics_repository.MetricsRepository'
|
||||
)).mock
|
||||
|
||||
self.metrics_resource = metrics.MetricsNames()
|
||||
|
||||
self.app.add_route('/v2.0/metrics/names',
|
||||
self.metrics_resource)
|
||||
|
||||
def test_get_metrics_names(self):
|
||||
expected_metrics_names = \
|
||||
{u'elements': [
|
||||
{u'name': u'cpu.frequency_mhz'},
|
||||
{u'name': u'cpu.idle_perc'},
|
||||
{u'name': u'cpu.idle_time'},
|
||||
{u'name': u'cpu.percent'},
|
||||
{u'name': u'cpu.stolen_perc'},
|
||||
{u'name': u'cpu.system_perc'},
|
||||
{u'name': u'cpu.system_time'}]}
|
||||
|
||||
return_value = self.metrics_repo_mock.return_value
|
||||
return_value.list_metric_names.return_value = expected_metrics_names['elements']
|
||||
response = self.simulate_request(path='/v2.0/metrics/names',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
method='GET')
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_metrics_names))
|
||||
|
||||
|
||||
class TestDimensionNames(base.BaseApiTestCase):
|
||||
def setUp(self):
|
||||
super(TestDimensionNames, self).setUp()
|
||||
|
||||
self.metrics_repo_mock = self.useFixture(fixtures.MockPatch(
|
||||
'monasca_api.common.repositories.influxdb.metrics_repository.MetricsRepository'
|
||||
)).mock
|
||||
|
||||
self.metrics_resource = metrics.DimensionNames()
|
||||
|
||||
self.app.add_route('/v2.0/metrics/dimensions/names',
|
||||
self.metrics_resource)
|
||||
|
||||
def test_get_dimension_names(self):
|
||||
expected_dimension_names = \
|
||||
{u'elements': [
|
||||
{u'dimension_name': u'component'},
|
||||
{u'dimension_name': u'consumer_group'},
|
||||
{u'dimension_name': u'device'},
|
||||
{u'dimension_name': u'hostname'},
|
||||
{u'dimension_name': u'mode'}]}
|
||||
|
||||
return_value = self.metrics_repo_mock.return_value
|
||||
return_value.list_dimension_names.return_value = expected_dimension_names['elements']
|
||||
response = self.simulate_request(path='/v2.0/metrics/dimensions/names',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
method='GET')
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_dimension_names))
|
||||
|
||||
|
||||
class TestDimensionValues(base.BaseApiTestCase):
|
||||
def setUp(self):
|
||||
super(TestDimensionValues, self).setUp()
|
||||
|
||||
self.metrics_repo_mock = self.useFixture(fixtures.MockPatch(
|
||||
'monasca_api.common.repositories.influxdb.metrics_repository.MetricsRepository'
|
||||
)).mock
|
||||
|
||||
self.metrics_resource = metrics.DimensionValues()
|
||||
|
||||
self.app.add_route('/v2.0/metrics/dimensions/values',
|
||||
self.metrics_resource)
|
||||
|
||||
def test_get_dimension_values(self):
|
||||
expected_dimension_names = \
|
||||
{u'elements': [
|
||||
{u'dimension_value': u'dummy_dimension_value'}]}
|
||||
|
||||
return_value = self.metrics_repo_mock.return_value
|
||||
return_value.list_dimension_values.return_value = expected_dimension_names['elements']
|
||||
response = self.simulate_request(path='/v2.0/metrics/dimensions/values',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
method='GET',
|
||||
query_string='dimension_name=dummy_dimension_name')
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_dimension_names))
|
||||
|
|
@ -0,0 +1,603 @@
|
|||
# Copyright 2019 FUJITSU LIMITED
|
||||
#
|
||||
# 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 datetime
|
||||
import json
|
||||
|
||||
import falcon.testing
|
||||
import fixtures
|
||||
from oslo_config import cfg
|
||||
|
||||
from monasca_api.tests import base
|
||||
from monasca_api.v2.reference import notifications
|
||||
from monasca_api.v2.reference import notificationstype
|
||||
|
||||
|
||||
CONF = cfg.CONF
|
||||
|
||||
TENANT_ID = u"fedcba9876543210fedcba9876543210"
|
||||
|
||||
|
||||
class TestNotifications(base.BaseApiTestCase):
|
||||
def setUp(self):
|
||||
super(TestNotifications, self).setUp()
|
||||
|
||||
self.conf_override(
|
||||
notifications_driver='monasca_api.common.repositories.sqla.'
|
||||
'notifications_repository:NotificationsRepository',
|
||||
group='repositories')
|
||||
|
||||
self.notifications_repo_mock = self.useFixture(fixtures.MockPatch(
|
||||
'monasca_api.common.repositories.sqla.notifications_repository.NotificationsRepository'
|
||||
)).mock
|
||||
|
||||
self.notifications_type_repo_mock = self.useFixture(fixtures.MockPatch(
|
||||
'monasca_api.common.repositories.sqla.'
|
||||
'notification_method_type_repository.NotificationMethodTypeRepository'
|
||||
)).mock
|
||||
|
||||
self.notification_resource = notifications.Notifications()
|
||||
self.app.add_route(
|
||||
'/v2.0/notification-methods', self.notification_resource)
|
||||
self.app.add_route(
|
||||
'/v2.0/notification-methods/{notification_method_id}', self.notification_resource)
|
||||
|
||||
def test_create_notifications(self):
|
||||
request_body = \
|
||||
{
|
||||
"name": "Name",
|
||||
"type": "EMAIL",
|
||||
"address": "john@doe.com"
|
||||
}
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.find_notification_by_name.return_value = {}
|
||||
|
||||
return_value = self.notifications_type_repo_mock.return_value
|
||||
return_value.list_notification_method_types.return_value = \
|
||||
[u'EMAIL',
|
||||
u'PAGERDUTY',
|
||||
u'WEBHOOK']
|
||||
|
||||
response = self.simulate_request(path='/v2.0/notification-methods',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID,
|
||||
'Content-Type': 'application/json'},
|
||||
method='POST',
|
||||
body=json.dumps(request_body))
|
||||
self.assertEqual(falcon.HTTP_201, response.status)
|
||||
|
||||
def test_create_notifications_with_incorrect_type(self):
|
||||
request_body = \
|
||||
{
|
||||
"name": "Name",
|
||||
"type": "MagicTYPE",
|
||||
"address": "john@doe.com"
|
||||
}
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.find_notification_by_name.return_value = {}
|
||||
|
||||
return_value = self.notifications_type_repo_mock.return_value
|
||||
return_value.list_notification_method_types.return_value = \
|
||||
[u'EMAIL',
|
||||
u'PAGERDUTY',
|
||||
u'WEBHOOK']
|
||||
|
||||
response = self.simulate_request(path='/v2.0/notification-methods',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID,
|
||||
'Content-Type': 'application/json'},
|
||||
method='POST',
|
||||
body=json.dumps(request_body))
|
||||
self.assertEqual(falcon.HTTP_400, response.status)
|
||||
|
||||
def test_create_notifications_when_name_is_taken(self):
|
||||
request_body = \
|
||||
{
|
||||
"name": "Name",
|
||||
"type": "EMAIL",
|
||||
"address": "john@doe.com"
|
||||
}
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.find_notification_by_name.return_value = \
|
||||
{'name': u'Name',
|
||||
'id': u'1',
|
||||
'tenant_id': u'444',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'a@b.com',
|
||||
'created_at': datetime.datetime(2019, 3, 22, 9, 35, 25),
|
||||
'updated_at': datetime.datetime(2019, 3, 22, 9, 35, 25)}
|
||||
|
||||
return_value = self.notifications_type_repo_mock.return_value
|
||||
return_value.list_notification_method_types.return_value = \
|
||||
[u'EMAIL',
|
||||
u'PAGERDUTY',
|
||||
u'WEBHOOK']
|
||||
|
||||
response = self.simulate_request(path='/v2.0/notification-methods',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID,
|
||||
'Content-Type': 'application/json'},
|
||||
method='POST',
|
||||
body=json.dumps(request_body))
|
||||
self.assertEqual(falcon.HTTP_409, response.status)
|
||||
|
||||
def test_list_notifications(self):
|
||||
expected_elements = \
|
||||
{'elements': [
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'a@b.com',
|
||||
'links': [{
|
||||
'href': 'http://falconframework.org/v2.0/notification-methods/1',
|
||||
'rel': 'self'}]}]}
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.list_notifications.return_value = \
|
||||
[{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'4199b031d5fa401abf9afaf7e58890b7',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'a@b.com',
|
||||
'created_at': datetime.datetime(2019, 3, 22, 9, 35, 25),
|
||||
'updated_at': datetime.datetime(2019, 3, 22, 9, 35, 25)}]
|
||||
response = self.simulate_request(path='/v2.0/notification-methods',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
method='GET')
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_elements))
|
||||
|
||||
def test_list_notifications_with_sort_by(self):
|
||||
expected_elements = \
|
||||
{'elements': [
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'a@b.com',
|
||||
'links': [{
|
||||
'href': 'http://falconframework.org/v2.0/notification-methods/1',
|
||||
'rel': 'self'}]}]}
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.list_notifications.return_value = \
|
||||
[{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'4199b031d5fa401abf9afaf7e58890b7',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'a@b.com',
|
||||
'created_at': datetime.datetime(2019, 3, 22, 9, 35, 25),
|
||||
'updated_at': datetime.datetime(2019, 3, 22, 9, 35, 25)}]
|
||||
response = self.simulate_request(path='/v2.0/notification-methods',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
query_string='sort_by=name',
|
||||
method='GET')
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_elements))
|
||||
|
||||
def test_list_notifications_with_incorrect_sort_by(self):
|
||||
response = self.simulate_request(path='/v2.0/notification-methods',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
query_string='sort_by=random_string',
|
||||
method='GET')
|
||||
self.assertEqual(falcon.HTTP_422, response.status)
|
||||
|
||||
def test_list_notifications_with_offset(self):
|
||||
expected_elements = \
|
||||
{'elements': [
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'a@b.com',
|
||||
'links': [{
|
||||
'href': 'http://falconframework.org/v2.0/notification-methods/1',
|
||||
'rel': 'self'}]}]}
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.list_notifications.return_value = \
|
||||
[{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'4199b031d5fa401abf9afaf7e58890b7',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'a@b.com',
|
||||
'created_at': datetime.datetime(2019, 3, 22, 9, 35, 25),
|
||||
'updated_at': datetime.datetime(2019, 3, 22, 9, 35, 25)}]
|
||||
response = self.simulate_request(path='/v2.0/notification-methods',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
query_string='offset=10',
|
||||
method='GET')
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_elements))
|
||||
|
||||
def test_list_notifications_with_incorrect_offset(self):
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.list_notifications.return_value = \
|
||||
[{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'4199b031d5fa401abf9afaf7e58890b7',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'a@b.com',
|
||||
'created_at': datetime.datetime(2019, 3, 22, 9, 35, 25),
|
||||
'updated_at': datetime.datetime(2019, 3, 22, 9, 35, 25)}]
|
||||
result = self.simulate_request(path='/v2.0/notification-methods',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
query_string='offset=ten',
|
||||
method='GET')
|
||||
self.assertEqual(falcon.HTTP_422, result.status)
|
||||
|
||||
def test_get_notification_with_id(self):
|
||||
expected_elements = \
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'a@b.com'}
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.list_notification.return_value = \
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'4199b031d5fa401abf9afaf7e58890b7',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'a@b.com',
|
||||
'created_at': datetime.datetime(2019, 3, 22, 9, 35, 25),
|
||||
'updated_at': datetime.datetime(2019, 3, 22, 9, 35, 25)}
|
||||
response = self.simulate_request(path='/v2.0/notification-methods/1',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
method='GET')
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_elements))
|
||||
|
||||
def test_delete_notification(self):
|
||||
response = self.simulate_request(path='/v2.0/notification-methods/1',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID},
|
||||
method='DELETE')
|
||||
self.assertEqual(falcon.HTTP_204, response.status)
|
||||
|
||||
def test_put_notification(self):
|
||||
expected_elements = \
|
||||
{"id": "1",
|
||||
"name": "shy_name",
|
||||
"type": "EMAIL",
|
||||
"address": "james@bond.com",
|
||||
"period": 0}
|
||||
|
||||
return_value = self.notifications_type_repo_mock.return_value
|
||||
return_value.list_notification_method_types.return_value = \
|
||||
[u'EMAIL',
|
||||
u'PAGERDUTY',
|
||||
u'WEBHOOK']
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.find_notification_by_name.return_value = \
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'444',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'a@b.com',
|
||||
'created_at': datetime.datetime(2019, 3, 22, 9, 35, 25),
|
||||
'updated_at': datetime.datetime(2019, 3, 22, 9, 35, 25)}
|
||||
|
||||
request_body = \
|
||||
{"name": "shy_name",
|
||||
"type": "EMAIL",
|
||||
"address": "james@bond.com",
|
||||
"period": 0}
|
||||
|
||||
response = self.simulate_request(path='/v2.0/notification-methods/1',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID,
|
||||
'Content-Type': 'application/json'},
|
||||
method='PUT',
|
||||
body=json.dumps(request_body))
|
||||
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_elements))
|
||||
|
||||
def test_patch_notification_all_fields(self):
|
||||
expected_elements = \
|
||||
{"id": "1",
|
||||
"name": "shy_name",
|
||||
"type": "EMAIL",
|
||||
"address": "james@bond.com",
|
||||
"period": 0}
|
||||
|
||||
return_value = self.notifications_type_repo_mock.return_value
|
||||
return_value.list_notification_method_types.return_value = \
|
||||
[u'EMAIL',
|
||||
u'PAGERDUTY',
|
||||
u'WEBHOOK']
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.find_notification_by_name.return_value = \
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'444',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'a@b.com',
|
||||
'created_at': datetime.datetime(2019, 3, 22, 9, 35, 25),
|
||||
'updated_at': datetime.datetime(2019, 3, 22, 9, 35, 25)}
|
||||
|
||||
request_body = \
|
||||
{"name": "shy_name",
|
||||
"type": "EMAIL",
|
||||
"address": "james@bond.com",
|
||||
"period": 0}
|
||||
|
||||
response = self.simulate_request(path='/v2.0/notification-methods/1',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID,
|
||||
'Content-Type': 'application/json'},
|
||||
method='PATCH',
|
||||
body=json.dumps(request_body))
|
||||
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_elements))
|
||||
|
||||
def test_patch_notification_name_fields(self):
|
||||
expected_elements = \
|
||||
{"id": "1",
|
||||
"name": "shy_name",
|
||||
"type": "EMAIL",
|
||||
"address": "james@bond.com",
|
||||
"period": 0}
|
||||
|
||||
return_value = self.notifications_type_repo_mock.return_value
|
||||
return_value.list_notification_method_types.return_value = \
|
||||
[u'EMAIL',
|
||||
u'PAGERDUTY',
|
||||
u'WEBHOOK']
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.find_notification_by_name.return_value = \
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'444',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'james@bond.com',
|
||||
'created_at': datetime.datetime(2019, 3, 22, 9, 35, 25),
|
||||
'updated_at': datetime.datetime(2019, 3, 22, 9, 35, 25)}
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.list_notification.return_value = \
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'444',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'james@bond.com'}
|
||||
|
||||
request_body = {"name": "shy_name"}
|
||||
|
||||
response = self.simulate_request(path='/v2.0/notification-methods/1',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID,
|
||||
'Content-Type': 'application/json'},
|
||||
method='PATCH',
|
||||
body=json.dumps(request_body))
|
||||
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_elements))
|
||||
|
||||
def test_patch_notification_type_fields(self):
|
||||
expected_elements = \
|
||||
{"id": "1",
|
||||
"name": "notification",
|
||||
"type": "PAGERDUTY",
|
||||
"address": "james@bond.com",
|
||||
"period": 0}
|
||||
|
||||
return_value = self.notifications_type_repo_mock.return_value
|
||||
return_value.list_notification_method_types.return_value = \
|
||||
[u'EMAIL',
|
||||
u'PAGERDUTY',
|
||||
u'WEBHOOK']
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.find_notification_by_name.return_value = \
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'444',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'james@bond.com',
|
||||
'created_at': datetime.datetime(2019, 3, 22, 9, 35, 25),
|
||||
'updated_at': datetime.datetime(2019, 3, 22, 9, 35, 25)}
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.list_notification.return_value = \
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'444',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'james@bond.com'}
|
||||
|
||||
request_body = {"type": "PAGERDUTY"}
|
||||
|
||||
response = self.simulate_request(path='/v2.0/notification-methods/1',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID,
|
||||
'Content-Type': 'application/json'},
|
||||
method='PATCH',
|
||||
body=json.dumps(request_body))
|
||||
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_elements))
|
||||
|
||||
def test_patch_notification_address_fields(self):
|
||||
expected_elements = \
|
||||
{"id": "1",
|
||||
"name": "notification",
|
||||
"type": "EMAIL",
|
||||
"address": "a@b.com",
|
||||
"period": 0}
|
||||
|
||||
return_value = self.notifications_type_repo_mock.return_value
|
||||
return_value.list_notification_method_types.return_value = \
|
||||
[u'EMAIL',
|
||||
u'PAGERDUTY',
|
||||
u'WEBHOOK']
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.find_notification_by_name.return_value = \
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'444',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'james@bond.com',
|
||||
'created_at': datetime.datetime(2019, 3, 22, 9, 35, 25),
|
||||
'updated_at': datetime.datetime(2019, 3, 22, 9, 35, 25)}
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.list_notification.return_value = \
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'444',
|
||||
'type': u'EMAIL',
|
||||
'period': 0,
|
||||
'address': u'james@bond.com'}
|
||||
|
||||
request_body = {"address": "a@b.com"}
|
||||
|
||||
response = self.simulate_request(path='/v2.0/notification-methods/1',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID,
|
||||
'Content-Type': 'application/json'},
|
||||
method='PATCH',
|
||||
body=json.dumps(request_body))
|
||||
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_elements))
|
||||
|
||||
def test_patch_notification_period_fields(self):
|
||||
expected_elements = \
|
||||
{"id": "1",
|
||||
"name": "notification",
|
||||
"type": "WEBHOOK",
|
||||
"address": "http://jamesbond.com",
|
||||
"period": 60}
|
||||
|
||||
return_value = self.notifications_type_repo_mock.return_value
|
||||
return_value.list_notification_method_types.return_value = \
|
||||
[u'EMAIL',
|
||||
u'PAGERDUTY',
|
||||
u'WEBHOOK']
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.find_notification_by_name.return_value = \
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'444',
|
||||
'type': u'WEBHOOK',
|
||||
'period': 0,
|
||||
'address': u'http://jamesbond.com',
|
||||
'created_at': datetime.datetime(2019, 3, 22, 9, 35, 25),
|
||||
'updated_at': datetime.datetime(2019, 3, 22, 9, 35, 25)}
|
||||
|
||||
return_value = self.notifications_repo_mock.return_value
|
||||
return_value.list_notification.return_value = \
|
||||
{'name': u'notification',
|
||||
'id': u'1',
|
||||
'tenant_id': u'444',
|
||||
'type': u'WEBHOOK',
|
||||
'period': 0,
|
||||
'address': u'http://jamesbond.com'}
|
||||
|
||||
request_body = {"period": 60}
|
||||
|
||||
response = self.simulate_request(path='/v2.0/notification-methods/1',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID,
|
||||
'Content-Type': 'application/json'},
|
||||
method='PATCH',
|
||||
body=json.dumps(request_body))
|
||||
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_elements))
|
||||
|
||||
|
||||
class TestNotificationsType(base.BaseApiTestCase):
|
||||
def setUp(self):
|
||||
super(TestNotificationsType, self).setUp()
|
||||
|
||||
self.conf_override(
|
||||
notifications_driver='monasca_api.common.repositories.sqla.'
|
||||
'notifications_repository:NotificationsRepository',
|
||||
group='repositories')
|
||||
|
||||
self.notifications_type_repo_mock = self.useFixture(fixtures.MockPatch(
|
||||
'monasca_api.common.repositories.sqla.'
|
||||
'notification_method_type_repository.NotificationMethodTypeRepository'
|
||||
)).mock
|
||||
|
||||
self.notification_resource = notificationstype.NotificationsType()
|
||||
self.app.add_route(
|
||||
'/v2.0/notification-methods/types', self.notification_resource)
|
||||
|
||||
def test_get_notification_types(self):
|
||||
expected_notification_types = \
|
||||
{'elements': [
|
||||
{'type': 'EMAIL'},
|
||||
{'type': 'PAGERDUTY'},
|
||||
{'type': 'WEBHOOK'}]}
|
||||
return_value = self.notifications_type_repo_mock.return_value
|
||||
return_value.list_notification_method_types.return_value = \
|
||||
[u'EMAIL',
|
||||
u'PAGERDUTY',
|
||||
u'WEBHOOK']
|
||||
|
||||
response = self.simulate_request(path='/v2.0/notification-methods/types',
|
||||
headers={'X-Roles':
|
||||
CONF.security.default_authorized_roles[0],
|
||||
'X-Tenant-Id': TENANT_ID,
|
||||
'Content-Type': 'application/json'},
|
||||
method='GET')
|
||||
self.assertEqual(falcon.HTTP_200, response.status)
|
||||
self.assertThat(response, base.RESTResponseEquals(expected_notification_types))
|
||||
Loading…
Reference in New Issue