utils/test/testapi/opnfv_testapi/common/check.py

   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 ##############################################################################
   9 import functools
  10
  11 from tornado import gen
  12 from tornado import web
  13
  14 from opnfv_testapi.common import message
  15 from opnfv_testapi.common import raises
  16
  17
  18 def authenticate(method):
  19     @web.asynchronous
  20     @gen.coroutine
  21     @functools.wraps(method)
  22     def wrapper(self, *args, **kwargs):
  23         if self.auth:
  24             try:
  25                 token = self.request.headers['X-Auth-Token']
  26             except KeyError:
  27                 raises.Unauthorized(message.unauthorized())
  28             query = {'access_token': token}
  29             check = yield self._eval_db_find_one(query, 'tokens')
  30             if not check:
  31                 raises.Forbidden(message.invalid_token())
  32         ret = yield gen.coroutine(method)(self, *args, **kwargs)
  33         raise gen.Return(ret)
  34     return wrapper
  35
  36
  37 def not_exist(xstep):
  38     @functools.wraps(xstep)
  39     def wrap(self, *args, **kwargs):
  40         query = kwargs.get('query')
  41         data = yield self._eval_db_find_one(query)
  42         if not data:
  43             raises.NotFound(message.not_found(self.table, query))
  44         ret = yield gen.coroutine(xstep)(self, data, *args, **kwargs)
  45         raise gen.Return(ret)
  46
  47     return wrap
  48
  49
  50 def no_body(xstep):
  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)
  56         raise gen.Return(ret)
  57
  58     return wrap
  59
  60
  61 def miss_fields(xstep):
  62     @functools.wraps(xstep)
  63     def wrap(self, *args, **kwargs):
  64         fields = kwargs.pop('miss_fields', [])
  65         if fields:
  66             for miss in 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)
  71         raise gen.Return(ret)
  72     return wrap
  73
  74
  75 def carriers_exist(xstep):
  76     @functools.wraps(xstep)
  77     def wrap(self, *args, **kwargs):
  78         carriers = kwargs.pop('carriers', {})
  79         if carriers:
  80             for table, query in carriers:
  81                 exist = yield self._eval_db_find_one(query(), table)
  82                 if not exist:
  83                     raises.Forbidden(message.not_found(table, query()))
  84         ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
  85         raise gen.Return(ret)
  86     return wrap
  87
  88
  89 def new_not_exists(xstep):
  90     @functools.wraps(xstep)
  91     def wrap(self, *args, **kwargs):
  92         query = kwargs.get('query')
  93         if query:
  94             to_data = yield self._eval_db_find_one(query())
  95             if to_data:
  96                 raises.Forbidden(message.exist(self.table, query()))
  97         ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
  98         raise gen.Return(ret)
  99     return wrap
 100
 101
 102 def updated_one_not_exist(xstep):
 103     @functools.wraps(xstep)
 104     def wrap(self, data, *args, **kwargs):
 105         db_keys = kwargs.pop('db_keys', [])
 106         query = self._update_query(db_keys, data)
 107         if query:
 108             to_data = yield self._eval_db_find_one(query)
 109             if to_data:
 110                 raises.Forbidden(message.exist(self.table, query))
 111         ret = yield gen.coroutine(xstep)(self, data, *args, **kwargs)
 112         raise gen.Return(ret)
 113     return wrap