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.handlers import VersionHandler, PodHandler, \
   6     ProjectHandler, TestcaseHandler, TestResultsHandler, DashboardHandler
   7 from 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             [
  28                 (r"/version", VersionHandler),
  29                 (r"/pods", PodHandler),
  30                 (r"/pods/([^/]+)", PodHandler),
  31                 (r"/projects", ProjectHandler),
  32                 (r"/projects/([^/]+)", ProjectHandler),
  33                 (r"/projects/([^/]+)/cases", TestcaseHandler),
  34                 (r"/projects/([^/]+)/cases/([^/]+)", TestcaseHandler),
  35                 (r"/results", TestResultsHandler),
  36                 (r"/results([^/]*)", TestResultsHandler),
  37                 (r"/results/([^/]*)", TestResultsHandler),
  38                 (r"/dashboard", DashboardHandler),
  39                 (r"/dashboard([^/]*)", DashboardHandler),
  40                 (r"/dashboard/([^/]*)", DashboardHandler),
  41             ],
  42             db=fake_pymongo,
  43             debug=True,
  44         )
  45
  46     def create_d(self, *args):
  47         return self.create(self.req_d, *args)
  48
  49     def create_e(self, *args):
  50         return self.create(self.req_e, *args)
  51
  52     def create(self, req=None, *args):
  53         return self.create_help(self.basePath, req, *args)
  54
  55     def create_help(self, uri, req, *args):
  56         if req:
  57             req = req.format()
  58         res = self.fetch(self._update_uri(uri, *args),
  59                          method='POST',
  60                          body=json.dumps(req),
  61                          headers=self.headers)
  62
  63         return self._get_return(res, self.create_res)
  64
  65     def get(self, *args):
  66         res = self.fetch(self._get_uri(*args),
  67                          method='GET',
  68                          headers=self.headers)
  69
  70         def inner():
  71             new_args, num = self._get_valid_args(*args)
  72             return self.get_res \
  73                 if num != self._need_arg_num(self.basePath) else self.list_res
  74         return self._get_return(res, inner())
  75
  76     def query(self, query):
  77         res = self.fetch(self._get_query_uri(query),
  78                          method='GET',
  79                          headers=self.headers)
  80         return self._get_return(res, self.list_res)
  81
  82     def update(self, new=None, *args):
  83         if new:
  84             new = new.format()
  85         res = self.fetch(self._get_uri(*args),
  86                          method='PUT',
  87                          body=json.dumps(new),
  88                          headers=self.headers)
  89         return self._get_return(res, self.update_res)
  90
  91     def delete(self, *args):
  92         res = self.fetch(self._get_uri(*args),
  93                          method='DELETE',
  94                          headers=self.headers)
  95         return res.code, res.body
  96
  97     @staticmethod
  98     def _get_valid_args(*args):
  99         new_args = tuple(['%s' % arg for arg in args if arg is not None])
 100         return new_args, len(new_args)
 101
 102     def _need_arg_num(self, uri):
 103         return uri.count('%s')
 104
 105     def _get_query_uri(self, query):
 106         return self.basePath + '?' + query
 107
 108     def _get_uri(self, *args):
 109         return self._update_uri(self.basePath, *args)
 110
 111     def _update_uri(self, uri, *args):
 112         r_uri = uri
 113         new_args, num = self._get_valid_args(*args)
 114         if num != self._need_arg_num(uri):
 115             r_uri += '/%s'
 116
 117         return r_uri % tuple(['%s' % arg for arg in new_args])
 118
 119     def _get_return(self, res, cls):
 120         code = res.code
 121         body = res.body
 122         return code, self._get_return_body(code, body, cls)
 123
 124     @staticmethod
 125     def _get_return_body(code, body, cls):
 126         return cls.from_dict(json.loads(body)) if code < 300 and cls else body
 127
 128     def assert_href(self, body):
 129         self.assertIn(self.basePath, body.href)
 130
 131     def assert_create_body(self, body, req=None, *args):
 132         if not req:
 133             req = self.req_d
 134         new_args = args + tuple([req.name])
 135         self.assertIn(self._get_uri(*new_args), body.href)
 136
 137     @staticmethod
 138     def _clear():
 139         fake_pymongo.pods.clear()
 140         fake_pymongo.projects.clear()
 141         fake_pymongo.testcases.clear()
 142         fake_pymongo.results.clear()