Infra
/
lemur
5
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Marshmallowing notifications. (#308)

This commit is contained in:
kevgliss 2016-05-10 11:27:57 -07:00
parent 008d608ec4
commit f9655213b3
8 changed files with 145 additions and 185 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
lemur/certificates/schemas.py
View File

@ -57,6 +57,7 @@ class CertificateOutputSchema(LemurOutputSchema):
description = fields.String()
issuer = fields.String()
name = fields.String()
common_name = fields.String()
not_after = fields.DateTime()
not_before = fields.DateTime()
owner = fields.Email()

34
lemur/notifications/schemas.py Normal file
View File

@ -0,0 +1,34 @@
"""
.. module: lemur.notifications.schemas
:platform: unix
:copyright: (c) 2015 by Netflix Inc., see AUTHORS for more
:license: Apache, see LICENSE for more details.
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
"""
from marshmallow import fields
from lemur.common.schema import LemurInputSchema, LemurOutputSchema
from lemur.schemas import PluginSchema, AssociatedCertificateSchema
class NotificationInputSchema(LemurInputSchema):
label = fields.String(required=True)
options = fields.Dict(required=True)
description = fields.String()
active = fields.Boolean()
plugin = fields.Nested(PluginSchema, required=True)
certificates = fields.Nested(AssociatedCertificateSchema, many=True)
class NotificationOutputSchema(LemurOutputSchema):
id = fields.Integer()
label = fields.String()
description = fields.String()
options = fields.Dict(dump_to='notification_options')
active = fields.Boolean()
plugin = fields.Nested(PluginSchema)
certificates = fields.Nested(AssociatedCertificateSchema, many=True)
notification_input_schema = NotificationInputSchema()
notification_output_schema = NotificationOutputSchema()
notifications_output_schema = NotificationOutputSchema(many=True)

11
lemur/notifications/service.py
View File

@ -319,10 +319,6 @@ def get_all():
def render(args):
sort_by = args.pop('sort_by')
sort_dir = args.pop('sort_dir')
page = args.pop('page')
count = args.pop('count')
filt = args.pop('filter')
certificate_id = args.pop('certificate_id', None)
@ -341,9 +337,4 @@ def render(args):
else:
query = database.filter(query, Notification, terms)
query = database.find_all(query, Notification, args)
if sort_by and sort_dir:
query = database.sort(query, Notification, sort_by, sort_dir)
return database.paginate(query, page, count)
return database.sort_and_page(query, Notification, args)

103
lemur/notifications/views.py
View File

@ -7,64 +7,27 @@
.. moduleauthor:: Kevin Glisson <kglisson@netflix.com>
"""
from flask import Blueprint
from flask.ext.restful import Api, reqparse, fields
from flask.ext.restful import Api, reqparse
from lemur.notifications import service
from lemur.notifications.schemas import notification_input_schema, notification_output_schema, notifications_output_schema
from lemur.auth.service import AuthenticatedResource
from lemur.common.utils import paginated_parser, marshal_items
from lemur.common.utils import paginated_parser
from lemur.common.schema import validate_schema
mod = Blueprint('notifications', __name__)
api = Api(mod)
FIELDS = {
'description': fields.String,
'notificationOptions': fields.Raw(attribute='options'),
'pluginName': fields.String(attribute='plugin_name'),
'label': fields.String,
'active': fields.Boolean,
'id': fields.Integer,
}
def notification(value, name):
"""
Validates a given notification exits
:param value:
:param name:
:return:
"""
n = service.get(value)
if not n:
raise ValueError("Unable to find notification specified")
return n
def notification_list(value, name):
"""
Validates a given notification exists and returns a list
:param value:
:param name:
:return:
"""
notifications = []
for v in value:
try:
notifications.append(notification(v['id'], 'id'))
except ValueError:
pass
return notifications
class NotificationsList(AuthenticatedResource):
""" Defines the 'notifications' endpoint """
def __init__(self):
self.reqparse = reqparse.RequestParser()
super(NotificationsList, self).__init__()
@marshal_items(FIELDS)
@validate_schema(None, notifications_output_schema)
def get(self):
"""
.. http:get:: /notifications
@ -144,8 +107,8 @@ class NotificationsList(AuthenticatedResource):
args = parser.parse_args()
return service.render(args)
@marshal_items(FIELDS)
def post(self):
@validate_schema(notification_input_schema, notification_output_schema)
def post(self, data=None):
"""
.. http:post:: /notifications
@ -251,18 +214,12 @@ class NotificationsList(AuthenticatedResource):
:reqheader Authorization: OAuth token to authenticate
:statuscode 200: no error
"""
self.reqparse.add_argument('label', type=str, location='json', required=True)
self.reqparse.add_argument('plugin', type=dict, location='json', required=True)
self.reqparse.add_argument('description', type=str, location='json')
self.reqparse.add_argument('certificates', type=list, default=[], location='json')
args = self.reqparse.parse_args()
return service.create(
args['label'],
args['plugin']['slug'],
args['plugin']['pluginOptions'],
args['description'],
args['certificates']
data['label'],
data['plugin']['slug'],
data['plugin']['pluginOptions'],
data['description'],
data['certificates']
)
@ -271,7 +228,7 @@ class Notifications(AuthenticatedResource):
self.reqparse = reqparse.RequestParser()
super(Notifications, self).__init__()
@marshal_items(FIELDS)
@validate_schema(None, notification_output_schema)
def get(self, notification_id):
"""
.. http:get:: /notifications/1
@ -338,8 +295,8 @@ class Notifications(AuthenticatedResource):
"""
return service.get(notification_id)
@marshal_items(FIELDS)
def put(self, notification_id):
@validate_schema(notification_input_schema, notification_output_schema)
def put(self, notification_id, data=None):
"""
.. http:put:: /notifications/1
@ -375,20 +332,13 @@ class Notifications(AuthenticatedResource):
:reqheader Authorization: OAuth token to authenticate
:statuscode 200: no error
"""
self.reqparse.add_argument('label', type=str, location='json', required=True)
self.reqparse.add_argument('notificationOptions', type=list, location='json')
self.reqparse.add_argument('active', type=bool, location='json')
self.reqparse.add_argument('certificates', type=list, default=[], location='json')
self.reqparse.add_argument('description', type=str, location='json')
args = self.reqparse.parse_args()
return service.update(
notification_id,
args['label'],
args['notificationOptions'],
args['description'],
args['active'],
args['certificates']
data['label'],
data['notificationOptions'],
data['description'],
data['active'],
data['certificates']
)
def delete(self, notification_id):
@ -401,8 +351,8 @@ class CertificateNotifications(AuthenticatedResource):
def __init__(self):
super(CertificateNotifications, self).__init__()
@marshal_items(FIELDS)
def get(self, certificate_id):
@validate_schema(None, notifications_output_schema)
def get(self, certificate_id, data=None):
"""
.. http:get:: /certificates/1/notifications
@ -476,11 +426,8 @@ class CertificateNotifications(AuthenticatedResource):
:reqheader Authorization: OAuth token to authenticate
:statuscode 200: no error
"""
parser = paginated_parser.copy()
parser.add_argument('active', type=bool, location='args')
args = parser.parse_args()
args['certificate_id'] = certificate_id
return service.render(args)
data['certificate_id'] = certificate_id
return service.render(data)
api.add_resource(NotificationsList, '/notifications', endpoint='notifications')

2
lemur/schemas.py
View File

@ -86,6 +86,8 @@ class AssociatedCertificateSchema(LemurInputSchema):
class PluginSchema(LemurInputSchema):
plugin_options = fields.Dict()
slug = fields.String()
title = fields.String()
description = fields.String()
@post_load
def get_object(self, data, many=False):

0
lemur/sources/schemas.py Normal file
View File

3
lemur/static/app/angular/notifications/view/view.js vendored
View File

@ -24,9 +24,6 @@ angular.module('lemur')
getData: function ($defer, params) {
NotificationApi.getList(params.url()).then(
function (data) {
_.each(data, function (notification) {
NotificationService.getPlugin(notification);
});
params.total(data.total);
$defer.resolve(data);
}

176
lemur/tests/test_notifications.py
View File

@ -1,117 +1,105 @@
from lemur.notifications.service import * # noqa
import pytest
from lemur.notifications.views import * # noqa
def test_crud(session):
notification = create('testnotify', 'email-notification', {}, 'notify1', [])
assert notification.id > 0
notification = update(notification.id, 'testnotify2', {}, 'notify2', True, [])
assert notification.label == 'testnotify2'
assert len(get_all()) == 1
delete(1)
assert len(get_all()) == 0
from .vectors import VALID_ADMIN_HEADER_TOKEN, VALID_USER_HEADER_TOKEN
def test_notification_get(client):
assert client.get(api.url_for(Notifications, notification_id=1)).status_code == 401
def test_notification_input_schema(client, notification):
from lemur.notifications.schemas import NotificationInputSchema
input_data = {
'label': 'notification1',
'options': {},
'description': 'my notification',
'active': True,
'plugin': {
'slug': 'email-notification'
}
}
data, errors = NotificationInputSchema().load(input_data)
assert not errors
def test_notification_post(client):
assert client.post(api.url_for(Notifications, notification_id=1), data={}).status_code == 405
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 404),
(VALID_ADMIN_HEADER_TOKEN, 404),
('', 401)
])
def test_notification_get(client, token, status):
assert client.get(api.url_for(Notifications, notification_id=1), headers=token).status_code == status
def test_notification_put(client):
assert client.put(api.url_for(Notifications, notification_id=1), data={}).status_code == 401
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
('', 405)
])
def test_notification_post_(client, token, status):
assert client.post(api.url_for(Notifications, notification_id=1), data={}, headers=token).status_code == status
def test_notification_delete(client):
assert client.delete(api.url_for(Notifications, notification_id=1)).status_code == 401
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 400),
(VALID_ADMIN_HEADER_TOKEN, 400),
('', 401)
])
def test_notification_put(client, token, status):
assert client.put(api.url_for(Notifications, notification_id=1), data={}, headers=token).status_code == status
def test_notification_patch(client):
assert client.patch(api.url_for(Notifications, notification_id=1), data={}).status_code == 405
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 200),
(VALID_ADMIN_HEADER_TOKEN, 200),
('', 401)
])
def test_notification_delete(client, token, status):
assert client.delete(api.url_for(Notifications, notification_id=1), headers=token).status_code == status
VALID_USER_HEADER_TOKEN = {
'Authorization': 'Basic ' + 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpYXQiOjE0MzUyMzMzNjksInN1YiI6MSwiZXhwIjoxNTIxNTQ2OTY5fQ.1qCi0Ip7mzKbjNh0tVd3_eJOrae3rNa_9MCVdA4WtQI'}
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
('', 405)
])
def test_notification_patch(client, token, status):
assert client.patch(api.url_for(Notifications, notification_id=1), data={}, headers=token).status_code == status
def test_auth_notification_get(client):
assert client.get(api.url_for(Notifications, notification_id=1), headers=VALID_USER_HEADER_TOKEN).status_code == 200
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 400),
(VALID_ADMIN_HEADER_TOKEN, 400),
('', 401)
])
def test_notification_list_post_(client, token, status):
assert client.post(api.url_for(NotificationsList), data={}, headers=token).status_code == status
def test_auth_notification_post_(client):
assert client.post(api.url_for(Notifications, notification_id=1), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 405
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 200),
(VALID_ADMIN_HEADER_TOKEN, 200),
('', 401)
])
def test_notification_list_get(client, token, status):
assert client.get(api.url_for(NotificationsList), headers=token).status_code == status
def test_auth_notification_put(client):
assert client.put(api.url_for(Notifications, notification_id=1), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 400
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
('', 405)
])
def test_notification_list_delete(client, token, status):
assert client.delete(api.url_for(NotificationsList), headers=token).status_code == status
def test_auth_notification_delete(client):
assert client.delete(api.url_for(Notifications, notification_id=1), headers=VALID_USER_HEADER_TOKEN).status_code == 200
def test_auth_notification_patch(client):
assert client.patch(api.url_for(Notifications, notification_id=1), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 405
VALID_ADMIN_HEADER_TOKEN = {
'Authorization': 'Basic ' + 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJpYXQiOjE0MzUyNTAyMTgsInN1YiI6MiwiZXhwIjoxNTIxNTYzODE4fQ.6mbq4-Ro6K5MmuNiTJBB153RDhlM5LGJBjI7GBKkfqA'}
def test_admin_notification_get(client):
assert client.get(api.url_for(Notifications, notification_id=1), headers=VALID_ADMIN_HEADER_TOKEN).status_code == 200
def test_admin_notification_post(client):
assert client.post(api.url_for(Notifications, notification_id=1), data={}, headers=VALID_ADMIN_HEADER_TOKEN).status_code == 405
def test_admin_notification_put(client):
assert client.put(api.url_for(Notifications, notification_id=1), data={}, headers=VALID_ADMIN_HEADER_TOKEN).status_code == 400
def test_admin_notification_delete(client):
assert client.delete(api.url_for(Notifications, notification_id=1), headers=VALID_ADMIN_HEADER_TOKEN).status_code == 200
def test_admin_notification_patch(client):
assert client.patch(api.url_for(Notifications, notification_id=1), data={}, headers=VALID_ADMIN_HEADER_TOKEN).status_code == 405
def test_notifications_get(client):
assert client.get(api.url_for(NotificationsList)).status_code == 401
def test_notifications_post(client):
assert client.post(api.url_for(NotificationsList), data={}).status_code == 401
def test_notifications_put(client):
assert client.put(api.url_for(NotificationsList), data={}).status_code == 405
def test_notifications_delete(client):
assert client.delete(api.url_for(NotificationsList)).status_code == 405
def test_notifications_patch(client):
assert client.patch(api.url_for(NotificationsList), data={}).status_code == 405
def test_auth_notifications_get(client):
assert client.get(api.url_for(NotificationsList), headers=VALID_USER_HEADER_TOKEN).status_code == 200
def test_auth_notifications_post(client):
assert client.post(api.url_for(NotificationsList), data={}, headers=VALID_USER_HEADER_TOKEN).status_code == 400
def test_admin_notifications_get(client):
resp = client.get(api.url_for(NotificationsList), headers=VALID_ADMIN_HEADER_TOKEN)
assert resp.status_code == 200
assert resp.json == {'items': [], 'total': 0}
@pytest.mark.parametrize("token,status", [
(VALID_USER_HEADER_TOKEN, 405),
(VALID_ADMIN_HEADER_TOKEN, 405),
('', 405)
])
def test_notification_list_patch(client, token, status):
assert client.patch(api.url_for(NotificationsList), data={}, headers=token).status_code == status