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