x/surveil
Code Issues Proposed changes

Config API: added notificationways

Browse Source 
Change-Id: I6a9ae1f2456623ad19e71d5bce68f1177f267ca1
This commit is contained in:
Vincent Fournier 2015-05-06 11:09:09 -04:00
parent 9cd8f2e48d
commit bbe0ecc768
6 changed files with 348 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
doc/source/webapi/v2/config.rst
View File

@ -54,6 +54,11 @@ Business impact modulations
.. rest-controller:: surveil.api.controllers.v2.config.businessimpactmodulations:BusinessImpactModulationsController
:webprefix: /v2/config/businessimpactmodulations
Notification ways
=================
.. rest-controller:: surveil.api.controllers.v2.config.notificationways:NotificationWaysController
:webprefix: /v2/config/notificationways
types documentation
===================
@ -69,3 +74,6 @@ types documentation
.. autotype:: surveil.api.datamodel.config.businessimpactmodulation.BuisnessImpactModulation
:members:
.. autotype:: surveil.api.datamodel.config.notificationway.NotificationWay
:members:

4
surveil/api/controllers/v2/config/__init__.py
View File

@ -18,6 +18,7 @@ from surveil.api.controllers.v2.config import contactgroups
from surveil.api.controllers.v2.config import contacts
from surveil.api.controllers.v2.config import hostgroups
from surveil.api.controllers.v2.config import hosts
from surveil.api.controllers.v2.config import notificationways
from surveil.api.controllers.v2.config import realms
from surveil.api.controllers.v2.config import reload_config
from surveil.api.controllers.v2.config import servicegroup
@ -41,6 +42,5 @@ class ConfigController(rest.RestController):
contactgroups = contactgroups.ContactGroupsController()
businessimpactmodulations = (businessimpactmodulations.
BusinessImpactModulationsController())
# notificationways = NotificationWayController()
notificationways = notificationways.NotificationWaysController()
# engine = EngineController()

72
surveil/api/controllers/v2/config/notificationways.py Normal file
View File

@ -0,0 +1,72 @@
# Copyright 2015 - Savoir-Faire Linux inc.
#
# 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 pecan
from pecan import rest
import wsmeext.pecan as wsme_pecan
from surveil.api.datamodel.config import notificationway
from surveil.api.handlers.config import notificationway_handler
from surveil.common import util
class NotificationWaysController(rest.RestController):
@util.policy_enforce(['authenticated'])
@wsme_pecan.wsexpose([notificationway.NotificationWay])
def get_all(self):
"""Returns all notification ways."""
handler = notificationway_handler.NotificationWayHandler(pecan.request)
notificationsway = handler.get_all()
return notificationsway
@util.policy_enforce(['authenticated'])
@wsme_pecan.wsexpose(notificationway.NotificationWay, unicode)
def get_one(self, notificationway_name):
"""Returns a specific notification way."""
handler = notificationway_handler.NotificationWayHandler(pecan.request)
notificationway = handler.get(notificationway_name)
return notificationway
@util.policy_enforce(['authenticated'])
@wsme_pecan.wsexpose(body=notificationway.NotificationWay, status_code=201)
def post(self, data):
"""Create a new notification way.
:param data: a notification way within the request body.
"""
handler = notificationway_handler.NotificationWayHandler(pecan.request)
handler.create(data)
@util.policy_enforce(['authenticated'])
@wsme_pecan.wsexpose(
notificationway.NotificationWay,
unicode,
status_code=204
)
def delete(self, notificationway_name):
"""Returns a specific notification way."""
handler = notificationway_handler.NotificationWayHandler(pecan.request)
handler.delete(notificationway_name)
@util.policy_enforce(['authenticated'])
@wsme_pecan.wsexpose(notificationway.NotificationWay,
unicode,
body=notificationway.NotificationWay,
status_code=204)
def put(self, notificationway_name, notificationway):
"""Update a specific notification way."""
handler = notificationway_handler.NotificationWayHandler(pecan.request)
handler.update(notificationway_name, notificationway)

41
surveil/api/datamodel/config/notificationway.py Normal file
View File

@ -0,0 +1,41 @@
# Copyright 2015 - Savoir-Faire Linux inc.
#
# 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 wsme
import wsme.types as wtypes
from surveil.api.datamodel import types
class NotificationWay(types.Base):
notificationway_name = wsme.wsattr(wtypes.text, mandatory=True)
host_notification_period = wsme.wsattr(wtypes.text, mandatory=True)
service_notification_period = wsme.wsattr(wtypes.text, mandatory=True)
host_notification_options = wsme.wsattr(wtypes.text, mandatory=True)
service_notification_options = wsme.wsattr(wtypes.text, mandatory=True)
host_notification_commands = wsme.wsattr(wtypes.text, mandatory=True)
service_notification_commands = wsme.wsattr(wtypes.text, mandatory=True)
min_business_impact = wsme.wsattr(int, mandatory=False)
@classmethod
def sample(cls):
return cls(
notificationway_name="email_in_day",
host_notification_period="24x7",
service_notification_period="24x7",
host_notification_options="d,u,r,f,s",
service_notification_options="w,u,c,r,f",
service_notification_commands="notify-service",
host_notification_commands="notify-host"
)

63
surveil/api/handlers/config/notificationway_handler.py Normal file
View File

@ -0,0 +1,63 @@
# Copyright 2015 - Savoir-Faire Linux inc.
#
# 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.
from surveil.api.datamodel.config import notificationway
from surveil.api.handlers import handler
class NotificationWayHandler(handler.Handler):
def get(self, notificationway_name):
"""Return a notification way."""
g = self.request.mongo_connection.shinken.notificationways.find_one(
{"notificationway_name": notificationway_name}, {'_id': 0}
)
return notificationway.NotificationWay(**g)
def update(self, notificationway_name, notificationway):
"""Modify an existing notification way."""
notificationway_dict = notificationway.as_dict()
if "notificationway_name" not in notificationway_dict.keys():
notificationway_dict['notificationway_name'] = notificationway_name
self.request.mongo_connection.shinken.notificationways.update(
{"notificationway_name": notificationway_name},
notificationway_dict
)
def delete(self, notificationway_name):
"""Delete existing notification way."""
self.request.mongo_connection.shinken.notificationways.remove(
{"notificationway_name": notificationway_name}
)
def create(self, notificationway):
"""Create a new notification way."""
self.request.mongo_connection.shinken.notificationways.insert(
notificationway.as_dict()
)
def get_all(self):
"""Return all notification way."""
notificationways = [
g for g in self.request.mongo_connection
.shinken.notificationways.find(
{"register": {"$ne": "0"}},
{'_id': 0}
)]
notificationways = [
notificationway.NotificationWay(**g) for g in notificationways
]
return notificationways

162
surveil/tests/api/controllers/v2/config/test_notificationways.py Normal file
View File

@ -0,0 +1,162 @@
# Copyright 2015 - Savoir-Faire Linux inc.
#
# 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 json
from surveil.tests.api import functionalTest
class TestNotificationWayController(functionalTest.FunctionalTest):
def setUp(self):
super(TestNotificationWayController, self).setUp()
self.notificationway = [
{
'notificationway_name': 'email_in_day',
'host_notification_period': '24x7',
'service_notification_period': '24x7',
'host_notification_options': 'd,u',
'service_notification_options': 'w,c,r',
'host_notification_commands': 'notify-service',
'service_notification_commands': 'notify-host'
},
{
'notificationway_name': 'email_all_time',
'host_notification_period': '24x7',
'service_notification_period': '24x7',
'host_notification_options': 'd,r,f,u',
'service_notification_options': 'w,f,c,r',
'host_notification_commands': 'notify-service',
'service_notification_commands': 'notify-host',
'min_business_impact': 5
}
]
self.mongoconnection.shinken.notificationways.insert(
copy.deepcopy(self.notificationway)
)
def test_get_all_notificationways(self):
response = self.get('/v2/config/notificationways')
self.assertItemsEqual(
json.loads(response.body.decode()),
[
{
'notificationway_name': 'email_in_day',
'host_notification_period': '24x7',
'service_notification_period': '24x7',
'host_notification_options': 'd,u',
'service_notification_options': 'w,c,r',
'host_notification_commands': 'notify-service',
'service_notification_commands': 'notify-host'
},
{
'notificationway_name': 'email_all_time',
'host_notification_period': '24x7',
'service_notification_period': '24x7',
'host_notification_options': 'd,r,f,u',
'service_notification_options': 'w,f,c,r',
'host_notification_commands': 'notify-service',
'service_notification_commands': 'notify-host',
'min_business_impact': 5
}
]
)
self.assertEqual(response.status_int, 200)
def test_get_one_notificationway(self):
response = self.get('/v2/config/notificationways/email_all_time')
self.assertEqual(
json.loads(response.body.decode()),
{
'notificationway_name': 'email_all_time',
'host_notification_period': '24x7',
'service_notification_period': '24x7',
'host_notification_options': 'd,r,f,u',
'service_notification_options': 'w,f,c,r',
'host_notification_commands': 'notify-service',
'service_notification_commands': 'notify-host',
'min_business_impact': 5
}
)
def test_create_notificationway(self):
notificationway = {
'notificationway_name': 'test_create_notification',
'host_notification_period': '24x7',
'service_notification_period': '24x7',
'host_notification_options': 'd,r,f,u',
'service_notification_options': 'w,f,c,r',
'host_notification_commands': 'notify-service',
'service_notification_commands': 'notify-host',
'min_business_impact': 5
}
self.post_json('/v2/config/notificationways', notificationway)
self.assertIsNotNone(
self.mongoconnection.
shinken.notificationways.find_one(notificationway)
)
def test_delete_notificationway(self):
notificationway = {
'notificationway_name': 'email_all_time',
}
self.assertIsNotNone(
self.mongoconnection.shinken.notificationways.find_one(
notificationway
)
)
self.delete('/v2/config/notificationways/email_all_time')
self.assertIsNone(
self.mongoconnection.shinken.notificationways.find_one(
notificationway
)
)
def test_put_notificationway(self):
self.assertEqual(
self.mongoconnection.shinken.notificationways.find_one(
{'notificationway_name': 'email_all_time'}
)['min_business_impact'],
5
)
self.put_json(
'/v2/config/notificationways/email_all_time',
{
'notificationway_name': 'email_all_time',
'host_notification_period': '24x7',
'service_notification_period': '24x7',
'host_notification_options': 'd,r,f,u',
'service_notification_options': 'w,f,c,r',
'host_notification_commands': 'notify-service',
'service_notification_commands': 'notify-host',
'min_business_impact': 3
}
)
self.assertEqual(
self.mongoconnection.shinken.notificationways.find_one(
{'notificationway_name': 'email_all_time'}
)['min_business_impact'],
3
)