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