utils/test/testapi/opnfv_testapi/tests/unit/test_base.py

   1 ##############################################################################
   2 # Copyright (c) 2016 ZTE Corporation
   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 json
  10
  11 from tornado import testing
  12 from tornado import web
  13
  14 import fake_pymongo
  15 from opnfv_testapi.resources import models
  16 from opnfv_testapi.router import url_mappings
  17
  18
  19 class TestBase(testing.AsyncHTTPTestCase):
  20     headers = {'Content-Type': 'application/json; charset=UTF-8'}
  21
  22     def setUp(self):
  23         self.basePath = ''
  24         self.create_res = models.CreateResponse
  25         self.get_res = None
  26         self.list_res = None
  27         self.update_res = None
  28         self.req_d = None
  29         self.req_e = None
  30         self.addCleanup(self._clear)
  31         super(TestBase, self).setUp()
  32
  33     def get_app(self):
  34         return web.Application(
  35             url_mappings.mappings,
  36             db=fake_pymongo,
  37             debug=True,
  38             auth=False
  39         )
  40
  41     def create_d(self, *args):
  42         return self.create(self.req_d, *args)
  43
  44     def create_e(self, *args):
  45         return self.create(self.req_e, *args)
  46
  47     def create(self, req=None, *args):
  48         return self.create_help(self.basePath, req, *args)
  49
  50     def create_help(self, uri, req, *args):
  51         if req and not isinstance(req, str) and hasattr(req, 'format'):
  52             req = req.format()
  53         res = self.fetch(self._update_uri(uri, *args),
  54                          method='POST',
  55                          body=json.dumps(req),
  56                          headers=self.headers)
  57
  58         return self._get_return(res, self.create_res)
  59
  60     def get(self, *args):
  61         res = self.fetch(self._get_uri(*args),
  62                          method='GET',
  63                          headers=self.headers)
  64
  65         def inner():
  66             new_args, num = self._get_valid_args(*args)
  67             return self.get_res \
  68                 if num != self._need_arg_num(self.basePath) else self.list_res
  69         return self._get_return(res, inner())
  70
  71     def query(self, query):
  72         res = self.fetch(self._get_query_uri(query),
  73                          method='GET',
  74                          headers=self.headers)
  75         return self._get_return(res, self.list_res)
  76
  77     def update(self, new=None, *args):
  78         if new:
  79             new = new.format()
  80         res = self.fetch(self._get_uri(*args),
  81                          method='PUT',
  82                          body=json.dumps(new),
  83                          headers=self.headers)
  84         return self._get_return(res, self.update_res)
  85
  86     def delete(self, *args):
  87         res = self.fetch(self._get_uri(*args),
  88                          method='DELETE',
  89                          headers=self.headers)
  90         return res.code, res.body
  91
  92     @staticmethod
  93     def _get_valid_args(*args):
  94         new_args = tuple(['%s' % arg for arg in args if arg is not None])
  95         return new_args, len(new_args)
  96
  97     def _need_arg_num(self, uri):
  98         return uri.count('%s')
  99
 100     def _get_query_uri(self, query):
 101         return self.basePath + '?' + query if query else self.basePath
 102
 103     def _get_uri(self, *args):
 104         return self._update_uri(self.basePath, *args)
 105
 106     def _update_uri(self, uri, *args):
 107         r_uri = uri
 108         new_args, num = self._get_valid_args(*args)
 109         if num != self._need_arg_num(uri):
 110             r_uri += '/%s'
 111
 112         return r_uri % tuple(['%s' % arg for arg in new_args])
 113
 114     def _get_return(self, res, cls):
 115         code = res.code
 116         body = res.body
 117         return code, self._get_return_body(code, body, cls)
 118
 119     @staticmethod
 120     def _get_return_body(code, body, cls):
 121         return cls.from_dict(json.loads(body)) if code < 300 and cls else body
 122
 123     def assert_href(self, body):
 124         self.assertIn(self.basePath, body.href)
 125
 126     def assert_create_body(self, body, req=None, *args):
 127         import inspect
 128         if not req:
 129             req = self.req_d
 130         resource_name = ''
 131         if inspect.isclass(req):
 132             resource_name = req.name
 133         elif isinstance(req, dict):
 134             resource_name = req['name']
 135         elif isinstance(req, str):
 136             resource_name = json.loads(req)['name']
 137         new_args = args + tuple([resource_name])
 138         self.assertIn(self._get_uri(*new_args), body.href)
 139
 140     @staticmethod
 141     def _clear():
 142         fake_pymongo.pods.clear()
 143         fake_pymongo.projects.clear()
 144         fake_pymongo.testcases.clear()
 145         fake_pymongo.results.clear()
 146         fake_pymongo.scenarios.clear()