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

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