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
18 def authenticate(method):
21 @functools.wraps(method)
22 def wrapper(self, *args, **kwargs):
25 token = self.request.headers['X-Auth-Token']
27 raises.Unauthorized(message.unauthorized())
28 query = {'access_token': token}
29 check = yield self._eval_db_find_one(query, 'tokens')
31 raises.Forbidden(message.invalid_token())
32 ret = yield gen.coroutine(method)(self, *args, **kwargs)
38 @functools.wraps(xstep)
39 def wrap(self, *args, **kwargs):
40 query = kwargs.get('query')
41 data = yield self._eval_db_find_one(query)
43 raises.NotFound(message.not_found(self.table, query))
44 ret = yield gen.coroutine(xstep)(self, data, *args, **kwargs)
51 @functools.wraps(xstep)
52 def wrap(self, *args, **kwargs):
53 if self.json_args is None:
54 raises.BadRequest(message.no_body())
55 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
61 def miss_fields(xstep):
62 @functools.wraps(xstep)
63 def wrap(self, *args, **kwargs):
64 fields = kwargs.get('miss_fields')
67 miss_data = self.json_args.get(miss)
68 if miss_data is None or miss_data == '':
69 raises.BadRequest(message.missing(miss))
70 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
75 def carriers_exist(xstep):
76 @functools.wraps(xstep)
77 def wrap(self, *args, **kwargs):
78 carriers = kwargs.get('carriers')
80 for table, query in carriers:
81 exist = yield self._eval_db_find_one(query(), table)
83 raises.Forbidden(message.not_found(table, query()))
84 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
89 def new_not_exists(xstep):
90 @functools.wraps(xstep)
91 def wrap(self, *args, **kwargs):
92 query = kwargs.get('query')
94 to_data = yield self._eval_db_find_one(query())
96 raises.Forbidden(message.exist(self.table, query()))
97 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
102 def updated_one_not_exist(xstep):
103 @functools.wraps(xstep)
104 def wrap(self, data, *args, **kwargs):
105 db_keys = kwargs.get('db_keys')
106 query = self._update_query(db_keys, data)
108 to_data = yield self._eval_db_find_one(query)
110 raises.Forbidden(message.exist(self.table, query))
111 ret = yield gen.coroutine(xstep)(self, data, *args, **kwargs)
112 raise gen.Return(ret)