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.resources.pod_handlers import PodCLHandler, PodGURHandler
   7 from opnfv_testapi.resources.project_handlers import ProjectCLHandler, \
   8     ProjectGURHandler
   9 from opnfv_testapi.resources.handlers import VersionHandler
  10 from opnfv_testapi.resources.testcase_handlers import TestcaseCLHandler, \
  11     TestcaseGURHandler
  12 from opnfv_testapi.resources.result_handlers import ResultsCLHandler, \
  13     ResultsGURHandler
  14 from opnfv_testapi.resources.dashboard_handlers import DashboardHandler
  15 from opnfv_testapi.resources.models import CreateResponse
  16 import fake_pymongo
  17
  18
  19 class TestBase(AsyncHTTPTestCase):
  20     headers = {'Content-Type': 'application/json; charset=UTF-8'}
  21
  22     def setUp(self):
  23         self.basePath = ''
  24         self.create_res = 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 Application(
  35             [
  36                 (r"/versions", VersionHandler),
  37                 (r"/api/v1/pods", PodCLHandler),
  38                 (r"/api/v1/pods/([^/]+)", PodGURHandler),
  39                 (r"/api/v1/projects", ProjectCLHandler),
  40                 (r"/api/v1/projects/([^/]+)", ProjectGURHandler),
  41                 (r"/api/v1/projects/([^/]+)/cases", TestcaseCLHandler),
  42                 (r"/api/v1/projects/([^/]+)/cases/([^/]+)",
  43                  TestcaseGURHandler),
  44                 (r"/api/v1/results", ResultsCLHandler),
  45                 (r"/api/v1/results/([^/]+)", ResultsGURHandler),
  46                 (r"/dashboard/v1/results", DashboardHandler),
  47             ],
  48             db=fake_pymongo,
  49             debug=True,
  50         )
  51
  52     def create_d(self, *args):
  53         return self.create(self.req_d, *args)
  54
  55     def create_e(self, *args):
  56         return self.create(self.req_e, *args)
  57
  58     def create(self, req=None, *args):
  59         return self.create_help(self.basePath, req, *args)
  60
  61     def create_help(self, uri, req, *args):
  62         if req:
  63             req = req.format()
  64         res = self.fetch(self._update_uri(uri, *args),
  65                          method='POST',
  66                          body=json.dumps(req),
  67                          headers=self.headers)
  68
  69         return self._get_return(res, self.create_res)
  70
  71     def get(self, *args):
  72         res = self.fetch(self._get_uri(*args),
  73                          method='GET',
  74                          headers=self.headers)
  75
  76         def inner():
  77             new_args, num = self._get_valid_args(*args)
  78             return self.get_res \
  79                 if num != self._need_arg_num(self.basePath) else self.list_res
  80         return self._get_return(res, inner())
  81
  82     def query(self, query):
  83         res = self.fetch(self._get_query_uri(query),
  84                          method='GET',
  85                          headers=self.headers)
  86         return self._get_return(res, self.list_res)
  87
  88     def update(self, new=None, *args):
  89         if new:
  90             new = new.format()
  91         res = self.fetch(self._get_uri(*args),
  92                          method='PUT',
  93                          body=json.dumps(new),
  94                          headers=self.headers)
  95         return self._get_return(res, self.update_res)
  96
  97     def delete(self, *args):
  98         res = self.fetch(self._get_uri(*args),
  99                          method='DELETE',
 100                          headers=self.headers)
 101         return res.code, res.body
 102
 103     @staticmethod
 104     def _get_valid_args(*args):
 105         new_args = tuple(['%s' % arg for arg in args if arg is not None])
 106         return new_args, len(new_args)
 107
 108     def _need_arg_num(self, uri):
 109         return uri.count('%s')
 110
 111     def _get_query_uri(self, query):
 112         return self.basePath + '?' + query
 113
 114     def _get_uri(self, *args):
 115         return self._update_uri(self.basePath, *args)
 116
 117     def _update_uri(self, uri, *args):
 118         r_uri = uri
 119         new_args, num = self._get_valid_args(*args)
 120         if num != self._need_arg_num(uri):
 121             r_uri += '/%s'
 122
 123         return r_uri % tuple(['%s' % arg for arg in new_args])
 124
 125     def _get_return(self, res, cls):
 126         code = res.code
 127         body = res.body
 128         return code, self._get_return_body(code, body, cls)
 129
 130     @staticmethod
 131     def _get_return_body(code, body, cls):
 132         return cls.from_dict(json.loads(body)) if code < 300 and cls else body
 133
 134     def assert_href(self, body):
 135         self.assertIn(self.basePath, body.href)
 136
 137     def assert_create_body(self, body, req=None, *args):
 138         if not req:
 139             req = self.req_d
 140         new_args = args + tuple([req.name])
 141         self.assertIn(self._get_uri(*new_args), body.href)
 142
 143     @staticmethod
 144     def _clear():
 145         fake_pymongo.pods.clear()
 146         fake_pymongo.projects.clear()
 147         fake_pymongo.testcases.clear()
 148         fake_pymongo.results.clear()