x/surveil
Code Issues Proposed changes

Use WSME for the Host controller

Browse Source 
Change-Id: I5bd02a5ef69946ff8896b79510a89640be846e79
This commit is contained in:
aviau 2014-08-21 14:02:27 -04:00
parent fd075316cb
commit 3d60cf9c6b
5 changed files with 160 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
doc/source/webapi/v1.rst
View File

@ -19,6 +19,9 @@ Hosts
.. rest-controller:: surveil.api.controllers.v1.hosts:HostController
:webprefix: /v1/hosts
.. autotype:: surveil.api.controllers.v1.datamodel.host.Host
:members:
Commands
========

54
surveil/api/controllers/v1/datamodel/host.py Normal file
View File

@ -0,0 +1,54 @@
# Copyright 2014 - 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.controllers.v1.datamodel import types
class Host(types.Base):
host_name = wsme.wsattr(wtypes.text, mandatory=True)
"""The name of the host"""
address = wsme.wsattr(wtypes.text, mandatory=True)
"""The address of the host. Normally, this is an IP address."""
max_check_attempts = wsme.wsattr(wtypes.text, mandatory=True)
check_period = wsme.wsattr(wtypes.text, mandatory=True)
"""The time period during which active checks of this host can be made."""
contacts = wsme.wsattr(wtypes.text, mandatory=True)
"""A list of the short names of the contacts that should be notified."""
contact_groups = wsme.wsattr(wtypes.text, mandatory=True)
"""List of the short names of the contact groups that should be notified"""
notification_interval = wsme.wsattr(wtypes.text, mandatory=True)
notification_period = wsme.wsattr(wtypes.text, mandatory=True)
@classmethod
def sample(cls):
return cls(
host_name="bogus-router",
address="192.168.1.254",
max_check_attempts="5",
check_period="24x7",
contacts="admin,carl",
contact_groups="router-admins",
notification_interval="30",
notification_period="24x7",
)

56
surveil/api/controllers/v1/hosts.py
View File

@ -12,10 +12,11 @@
# License for the specific language governing permissions and limitations
# under the License.
import json
import pecan
from pecan import rest
import wsmeext.pecan as wsme_pecan
from surveil.api.controllers.v1.datamodel import host
class HostController(rest.RestController):
@ -24,33 +25,36 @@ class HostController(rest.RestController):
pecan.request.context['host_id'] = host_id
self._id = host_id
@pecan.expose("json")
@wsme_pecan.wsexpose(host.Host)
def get(self):
"""Returns a specific host."""
host = pecan.request.mongo_connection.shinken.hosts.find_one(
h = pecan.request.mongo_connection.shinken.hosts.find_one(
{"host_name": self._id}
)
if host:
del host['_id']
return host
return host.Host(**h)
@wsme_pecan.wsexpose(None, body=host.Host, status_code=204)
def put(self, data):
"""Modify this host.
:param data: a host within the request body.
"""
host_dict = data.as_dict()
if "host_name" not in host_dict.keys():
host_dict['host_name'] = self._id
@pecan.expose()
def put(self):
"""Modify this host."""
body = json.loads(pecan.request.body.decode())
pecan.request.mongo_connection.shinken.hosts.update(
{"host_name": self._id},
body
host_dict
)
pecan.response.status = 204
@pecan.expose()
@wsme_pecan.wsexpose(None, status_code=204)
def delete(self):
"""Delete this host."""
pecan.request.mongo_connection.shinken.hosts.remove(
{"host_name": self._id}
)
pecan.response.status = 204
class HostsController(rest.RestController):
@ -59,12 +63,20 @@ class HostsController(rest.RestController):
def _lookup(self, host_id, *remainder):
return HostController(host_id), remainder
@pecan.expose("json")
@wsme_pecan.wsexpose([host.Host])
def get_all(self):
"""Returns all host."""
hosts = [host for host in
pecan.request.mongo_connection.shinken.hosts.find()]
for host in hosts:
del host['_id']
"""Returns all hosts."""
hosts = [h for h
in pecan.request.mongo_connection.shinken.hosts.find()]
return hosts
return [host.Host(**h) for h in hosts]
@wsme_pecan.wsexpose(host.Host, body=host.Host, status_code=201)
def post(self, data):
"""Create a new host.
:param data: a host within the request body.
"""
pecan.request.mongo_connection.shinken.hosts.insert(
data.as_dict()
)

123
surveil/tests/api/controllers/v1/test_hosts.py
View File

@ -20,85 +20,98 @@ from surveil.tests.api import functionalTest
class TestRootController(functionalTest.FunctionalTest):
def test_get_all_hosts(self):
hosts = [
{u"use": u"generic-host", u"contact_groups": u"admins",
u"host_name": u"testhost1", u"address": u"www.google.ca"},
{u"use": u"generic-host", u"contact_groups": u"admins",
u"host_name": u"testhost2", u"address": u"www.google.ca"},
{u"use": u"generic-host", u"contact_groups": u"admins",
u"host_name": u"testhost3", u"address": u"www.google.ca"}
def setUp(self):
super(TestRootController, self).setUp()
self.hosts = [
{
"host_name": "bogus-router", "address": "192.168.1.254",
"max_check_attempts": "5", "check_period": "24x7",
"contacts": "admin,carl", "contact_groups": "router-admins",
"notification_interval": "30", "notification_period": "24x7"
},
{
"host_name": "bogus-router2", "address": "192.168.1.254",
"max_check_attempts": "5", "check_period": "24x7",
"contacts": "admin,carl", "contact_groups": "router-admins",
"notification_interval": "30", "notification_period": "24x7"
},
{
"host_name": "bogus-router333", "address": "192.168.1.254",
"max_check_attempts": "5", "check_period": "24x7",
"contacts": "admin,carl", "contact_groups": "router-admins",
"notification_interval": "30", "notification_period": "24x7"
},
]
self.mongoconnection.shinken.hosts.insert(copy.deepcopy(hosts))
self.mongoconnection.shinken.hosts.insert(
copy.deepcopy(self.hosts)
)
def test_get_all_hosts(self):
response = self.app.get('/v1/hosts')
self.assert_count_equal_backport(
json.loads(response.body.decode()),
hosts
self.hosts
)
self.assertEqual(response.status_int, 200)
def test_get_specific_host(self):
hosts = [
{u"use": u"generic-host", u"contact_groups": u"admins",
u"host_name": u"testhost1", u"address": u"www.google.ca"},
{u"use": u"generic-host", u"contact_groups": u"admins",
u"host_name": u"testhost2", u"address": u"www.google.ca"},
{u"use": u"generic-host", u"contact_groups": u"admins",
u"host_name": u"testhost3", u"address": u"www.google.ca"}
]
self.mongoconnection.shinken.hosts.insert(copy.deepcopy(hosts))
response = self.app.get('/v1/hosts/testhost2')
response = self.app.get('/v1/hosts/bogus-router333')
self.assert_count_equal_backport(
json.loads(response.body.decode()),
hosts[1]
self.hosts[2]
)
self.assertEqual(response.status_int, 200)
def test_update_host(self):
hosts = [{u"use": u"generic-host", u"contact_groups": u"admins",
u"host_name": u"testhost1", u"address": u"www.google.ca"}]
self.mongoconnection.shinken.hosts.insert(copy.deepcopy(hosts))
put_body = {u"use": u"generic-host", u"contact_groups": u"admins",
u"host_name": u"testhost1", u"address": u"test.com"}
response = self.app.put_json("/v1/hosts/testhost1", params=put_body)
expected_host = {u"use": u"generic-host", u"contact_groups": u"admins",
u"host_name": u"testhost1", u"address": u"test.com"}
mongo_hosts = self.mongoconnection.shinken.hosts.find_one(
{"host_name": "testhost1"}
put_host = {
u"host_name": u"bogus-router333",
u"address": u"newputaddress",
u"max_check_attempts": u"222225",
u"check_period": u"newtimeperiod",
u"contacts": u"aaa,bbb",
u"contact_groups": u"newgroup",
u"notification_interval": u"newnotificationinterval",
u"notification_period": u"newnotificationperiod"
}
response = self.app.put_json(
"/v1/hosts/bogus-router333", params=put_host
)
del mongo_hosts['_id']
self.assertEqual(expected_host, mongo_hosts)
mongo_host = self.mongoconnection.shinken.hosts.find_one(
{'host_name': 'bogus-router333'}
)
del mongo_host['_id']
self.assertEqual(put_host, mongo_host)
self.assertEqual(response.status_int, 204)
def test_delete_host(self):
hosts = [
{u"use": u"generic-host", u"contact_groups": u"admins",
u"host_name": u"testhost1", u"address": u"www.google.ca"},
{u"use": u"generic-host", u"contact_groups": u"admins",
u"host_name": u"testhost2", u"address": u"www.google.ca"}
]
self.mongoconnection.shinken.hosts.insert(copy.deepcopy(hosts))
response = self.app.delete('/v1/hosts/bogus-router')
response = self.app.delete('/v1/hosts/testhost2')
expected_hosts = [
{u"use": u"generic-host", u"contact_groups": u"admins",
u"host_name": u"testhost1", u"address": u"www.google.ca"}
]
mongo_hosts = [host for host
in self.mongoconnection.shinken.hosts.find()]
for host in mongo_hosts:
del host['_id']
self.assertEqual(expected_hosts, mongo_hosts)
self.assertEqual(2, len(mongo_hosts))
self.assertEqual(response.status_int, 204)
def test_add_host(self):
new_host = {
"host_name": "testpost",
"address": "192.168.1.254",
"max_check_attempts": "5",
"check_period": "24x7",
"contacts": "admin,carl",
"contact_groups": "router-admins",
"notification_interval": "30",
"notification_period": "24x7"
}
response = self.app.post_json("/v1/hosts", params=new_host)
hosts = [h for h in self.mongoconnection.shinken.hosts.find()]
for h in hosts:
del h["_id"]
self.assertTrue(new_host in hosts)
self.assertEqual(response.status_int, 201)

1
surveil/tests/base.py
View File

@ -17,6 +17,7 @@ import unittest
class BaseTestCase(unittest.TestCase):
maxDiff = None
def assert_count_equal_backport(self, item1, item2):
if sys.version_info[0] >= 3: