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