1 ##############################################################################
2 # Copyright (c) 2017 ZTE Corp
3 # feng.xiaowei@zte.com.cn
4 # All rights reserved. This program and the accompanying materials
5 # are made available under the terms of the Apache License, Version 2.0
6 # which accompanies this distribution, and is available at
7 # http://www.apache.org/licenses/LICENSE-2.0
8 ##############################################################################
11 from tornado import gen
12 from tornado import web
14 from opnfv_testapi.common import message
15 from opnfv_testapi.common import raises
16 from opnfv_testapi.db import api as dbapi
19 def authenticate(method):
22 @functools.wraps(method)
23 def wrapper(self, *args, **kwargs):
26 token = self.request.headers['X-Auth-Token']
28 raises.Unauthorized(message.unauthorized())
29 query = {'access_token': token}
30 check = yield dbapi.db_find_one('tokens', query)
32 raises.Forbidden(message.invalid_token())
33 ret = yield gen.coroutine(method)(self, *args, **kwargs)
39 @functools.wraps(xstep)
40 def wrap(self, *args, **kwargs):
41 query = kwargs.get('query')
42 data = yield dbapi.db_find_one(self.table, query)
44 raises.NotFound(message.not_found(self.table, query))
45 ret = yield gen.coroutine(xstep)(self, data, *args, **kwargs)
52 @functools.wraps(xstep)
53 def wrap(self, *args, **kwargs):
54 if self.json_args is None:
55 raises.BadRequest(message.no_body())
56 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
62 def miss_fields(xstep):
63 @functools.wraps(xstep)
64 def wrap(self, *args, **kwargs):
65 fields = kwargs.pop('miss_fields', [])
68 miss_data = self.json_args.get(miss)
69 if miss_data is None or miss_data == '':
70 raises.BadRequest(message.missing(miss))
71 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
76 def carriers_exist(xstep):
77 @functools.wraps(xstep)
78 def wrap(self, *args, **kwargs):
79 carriers = kwargs.pop('carriers', {})
81 for table, query in carriers:
82 exist = yield dbapi.db_find_one(table, query())
84 raises.Forbidden(message.not_found(table, query()))
85 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
90 def new_not_exists(xstep):
91 @functools.wraps(xstep)
92 def wrap(self, *args, **kwargs):
93 query = kwargs.get('query')
95 to_data = yield dbapi.db_find_one(self.table, query())
97 raises.Forbidden(message.exist(self.table, query()))
98 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
103 def updated_one_not_exist(xstep):
104 @functools.wraps(xstep)
105 def wrap(self, data, *args, **kwargs):
106 db_keys = kwargs.pop('db_keys', [])
107 query = self._update_query(db_keys, data)
109 to_data = yield dbapi.db_find_one(self.table, query)
111 raises.Forbidden(message.exist(self.table, query))
112 ret = yield gen.coroutine(xstep)(self, data, *args, **kwargs)
113 raise gen.Return(ret)