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

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