2 from tornado.web import Application
3 from tornado.testing import AsyncHTTPTestCase
5 from resources.pod_handlers import PodCLHandler, PodGURHandler
6 from resources.project_handlers import ProjectCLHandler, ProjectGURHandler
7 from resources.handlers import VersionHandler, \
8 TestcaseHandler, TestResultsHandler, DashboardHandler
9 from resources.models import CreateResponse
13 class TestBase(AsyncHTTPTestCase):
14 headers = {'Content-Type': 'application/json; charset=UTF-8'}
18 self.create_res = CreateResponse
21 self.update_res = None
24 self.addCleanup(self._clear)
25 super(TestBase, self).setUp()
30 (r"/versions", VersionHandler),
31 (r"/api/v1/pods", PodCLHandler),
32 (r"/api/v1/pods/([^/]+)", PodGURHandler),
33 (r"/api/v1/projects", ProjectCLHandler),
34 (r"/api/v1/projects/([^/]+)", ProjectGURHandler),
35 (r"/api/v1/projects/([^/]+)/cases", TestcaseHandler),
36 (r"/api/v1/projects/([^/]+)/cases/([^/]+)", TestcaseHandler),
37 (r"/api/v1/results", TestResultsHandler),
38 (r"/api/v1/results([^/]*)", TestResultsHandler),
39 (r"/api/v1/results/([^/]*)", TestResultsHandler),
40 (r"/dashboard/v1/results", DashboardHandler),
41 (r"/dashboard/v1/results([^/]*)", DashboardHandler),
47 def create_d(self, *args):
48 return self.create(self.req_d, *args)
50 def create_e(self, *args):
51 return self.create(self.req_e, *args)
53 def create(self, req=None, *args):
54 return self.create_help(self.basePath, req, *args)
56 def create_help(self, uri, req, *args):
59 res = self.fetch(self._update_uri(uri, *args),
64 return self._get_return(res, self.create_res)
67 res = self.fetch(self._get_uri(*args),
72 new_args, num = self._get_valid_args(*args)
74 if num != self._need_arg_num(self.basePath) else self.list_res
75 return self._get_return(res, inner())
77 def query(self, query):
78 res = self.fetch(self._get_query_uri(query),
81 return self._get_return(res, self.list_res)
83 def update(self, new=None, *args):
86 res = self.fetch(self._get_uri(*args),
90 return self._get_return(res, self.update_res)
92 def delete(self, *args):
93 res = self.fetch(self._get_uri(*args),
96 return res.code, res.body
99 def _get_valid_args(*args):
100 new_args = tuple(['%s' % arg for arg in args if arg is not None])
101 return new_args, len(new_args)
103 def _need_arg_num(self, uri):
104 return uri.count('%s')
106 def _get_query_uri(self, query):
107 return self.basePath + '?' + query
109 def _get_uri(self, *args):
110 return self._update_uri(self.basePath, *args)
112 def _update_uri(self, uri, *args):
114 new_args, num = self._get_valid_args(*args)
115 if num != self._need_arg_num(uri):
118 return r_uri % tuple(['%s' % arg for arg in new_args])
120 def _get_return(self, res, cls):
123 return code, self._get_return_body(code, body, cls)
126 def _get_return_body(code, body, cls):
127 return cls.from_dict(json.loads(body)) if code < 300 and cls else body
129 def assert_href(self, body):
130 self.assertIn(self.basePath, body.href)
132 def assert_create_body(self, body, req=None, *args):
135 new_args = args + tuple([req.name])
136 self.assertIn(self._get_uri(*new_args), body.href)
140 fake_pymongo.pods.clear()
141 fake_pymongo.projects.clear()
142 fake_pymongo.testcases.clear()
143 fake_pymongo.results.clear()