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
13 from tornado import web
15 from opnfv_testapi.common import constants
16 from opnfv_testapi.common import message
17 from opnfv_testapi.common import raises
18 from opnfv_testapi.common.config import CONF
19 from opnfv_testapi.db import api as dbapi
25 @functools.wraps(method)
26 def wrapper(self, *args, **kwargs):
27 ticket = self.get_query_argument('ticket', default=None)
29 client = cas.CASClient(version='2',
30 server_url=CONF.lfid_cas_url,
31 service_url=CONF.ui_url)
32 (user, attrs, _) = client.verify_ticket(ticket=ticket)
33 print 'login user: {}'.format(user)
36 'email': attrs.get('mail'),
37 'fullname': attrs.get('field_lf_full_name'),
38 'groups': constants.TESTAPI_USERS + attrs.get('group', [])
40 q_user = {'user': user}
41 db_user = yield dbapi.db_find_one(constants.USER_TABLE, q_user)
43 dbapi.db_save(constants.USER_TABLE, login_user)
45 dbapi.db_update(constants.USER_TABLE, q_user, login_user)
47 self.clear_cookie(constants.TESTAPI_ID)
48 self.set_secure_cookie(constants.TESTAPI_ID, user)
49 ret = yield gen.coroutine(method)(self, *args, **kwargs)
54 def authenticate(method):
57 @functools.wraps(method)
58 def wrapper(self, *args, **kwargs):
61 token = self.request.headers['X-Auth-Token']
63 raises.Unauthorized(message.unauthorized())
64 query = {'access_token': token}
65 check = yield dbapi.db_find_one('tokens', query)
67 raises.Forbidden(message.invalid_token())
68 ret = yield gen.coroutine(method)(self, *args, **kwargs)
74 @functools.wraps(xstep)
75 def wrap(self, *args, **kwargs):
76 query = kwargs.get('query')
77 data = yield dbapi.db_find_one(self.table, query)
79 raises.NotFound(message.not_found(self.table, query))
80 ret = yield gen.coroutine(xstep)(self, data, *args, **kwargs)
87 @functools.wraps(xstep)
88 def wrap(self, *args, **kwargs):
89 if self.json_args is None:
90 raises.BadRequest(message.no_body())
91 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
97 def miss_fields(xstep):
98 @functools.wraps(xstep)
99 def wrap(self, *args, **kwargs):
100 fields = kwargs.pop('miss_fields', [])
103 miss_data = self.json_args.get(miss)
104 if miss_data is None or miss_data == '':
105 raises.BadRequest(message.missing(miss))
106 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
107 raise gen.Return(ret)
111 def carriers_exist(xstep):
112 @functools.wraps(xstep)
113 def wrap(self, *args, **kwargs):
114 carriers = kwargs.pop('carriers', {})
116 for table, query in carriers:
117 exist = yield dbapi.db_find_one(table, query())
119 raises.Forbidden(message.not_found(table, query()))
120 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
121 raise gen.Return(ret)
125 def new_not_exists(xstep):
126 @functools.wraps(xstep)
127 def wrap(self, *args, **kwargs):
128 query = kwargs.get('query')
130 to_data = yield dbapi.db_find_one(self.table, query())
132 raises.Forbidden(message.exist(self.table, query()))
133 ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
134 raise gen.Return(ret)
138 def updated_one_not_exist(xstep):
139 @functools.wraps(xstep)
140 def wrap(self, data, *args, **kwargs):
141 db_keys = kwargs.pop('db_keys', [])
142 query = self._update_query(db_keys, data)
144 to_data = yield dbapi.db_find_one(self.table, query)
146 raises.Forbidden(message.exist(self.table, query))
147 ret = yield gen.coroutine(xstep)(self, data, *args, **kwargs)
148 raise gen.Return(ret)