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