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 ##############################################################################
12 from tornado import gen
14 from opnfv_testapi.common import constants
15 from opnfv_testapi.common import message
16 from opnfv_testapi.common import raises
17 from opnfv_testapi.common.config import CONF
18 from opnfv_testapi.db import api as dbapi
21 def is_authorized(method):
22 @functools.wraps(method)
23 def wrapper(self, *args, **kwargs):
24 if CONF.api_authenticate and self.table in ['pods']:
25 testapi_id = self.get_secure_cookie(constants.TESTAPI_ID)
27 raises.Unauthorized(message.not_login())
28 user_info = yield dbapi.db_find_one('users', {'user': testapi_id})
30 raises.Unauthorized(message.not_lfid())
31 kwargs['owner'] = testapi_id
32 ret = yield gen.coroutine(method)(self, *args, **kwargs)
37 def valid_token(method):
38 @functools.wraps(method)
39 def wrapper(self, *args, **kwargs):
40 if self.auth and self.table == 'results':
42 token = self.request.headers['X-Auth-Token']
44 raises.Unauthorized(message.unauthorized())
45 query = {'access_token': token}
46 check = yield dbapi.db_find_one('tokens', query)
48 raises.Forbidden(message.invalid_token())
49 ret = yield gen.coroutine(method)(self, *args, **kwargs)
55 @functools.wraps(xstep)
56 def wrap(self, *args, **kwargs):
57 query = kwargs.get('query')
58 data = yield dbapi.db_find_one(self.table, query)
60 raises.NotFound(message.not_found(self.table, query))
61 ret = yield gen.coroutine(xstep)(self, data, *args, **kwargs)
68 @functools.wraps(xstep)
69 def wrap(self, *args, **kwargs):
70 if self.json_args is None:
71 raises.BadRequest(message.no_body())
72 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
78 def miss_fields(xstep):
79 @functools.wraps(xstep)
80 def wrap(self, *args, **kwargs):
81 fields = kwargs.pop('miss_fields', [])
84 miss_data = self.json_args.get(miss)
85 if miss_data is None or miss_data == '':
86 raises.BadRequest(message.missing(miss))
87 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
92 def carriers_exist(xstep):
93 @functools.wraps(xstep)
94 def wrap(self, *args, **kwargs):
95 carriers = kwargs.pop('carriers', {})
97 for table, query in carriers:
98 exist = yield dbapi.db_find_one(table, query())
100 raises.Forbidden(message.not_found(table, query()))
101 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
102 raise gen.Return(ret)
106 def values_check(xstep):
107 @functools.wraps(xstep)
108 def wrap(self, *args, **kwargs):
109 checks = kwargs.pop('values_check', {})
111 for field, check, options in checks:
112 if not check(field, options):
113 raises.BadRequest(message.invalid_value(field, options))
114 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
115 raise gen.Return(ret)
119 def new_not_exists(xstep):
120 @functools.wraps(xstep)
121 def wrap(self, *args, **kwargs):
122 query = kwargs.get('query')
125 if self.table == 'pods':
126 if query_data.get('name') is not None:
127 query_data['name'] = re.compile(query_data.get('name'),
129 to_data = yield dbapi.db_find_one(self.table, query_data)
131 raises.Forbidden(message.exist(self.table, query()))
132 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
133 raise gen.Return(ret)
137 def updated_one_not_exist(xstep):
138 @functools.wraps(xstep)
139 def wrap(self, data, *args, **kwargs):
140 db_keys = kwargs.pop('db_keys', [])
141 query = self._update_query(db_keys, data)
143 to_data = yield dbapi.db_find_one(self.table, query)
145 raises.Forbidden(message.exist(self.table, query))
146 ret = yield gen.coroutine(xstep)(self, data, *args, **kwargs)
147 raise gen.Return(ret)