1 ##############################################################################
2 # Copyright (c) 2016 ZTE Corporation
3 # feng.xiaowei@zte.com.cn
4 # All rights reserved. This program and the accompanying materials
5 # are made available under the terms of the Apache License, Version 2.0
6 # which accompanies this distribution, and is available at
7 # http://www.apache.org/licenses/LICENSE-2.0
8 ##############################################################################
11 from tornado.web import Application
12 from tornado.testing import AsyncHTTPTestCase
14 from opnfv_testapi.router import url_mappings
15 from opnfv_testapi.resources.models import CreateResponse
19 class TestBase(AsyncHTTPTestCase):
20 headers = {'Content-Type': 'application/json; charset=UTF-8'}
24 self.create_res = CreateResponse
27 self.update_res = None
30 self.addCleanup(self._clear)
31 super(TestBase, self).setUp()
35 url_mappings.mappings,
40 def create_d(self, *args):
41 return self.create(self.req_d, *args)
43 def create_e(self, *args):
44 return self.create(self.req_e, *args)
46 def create(self, req=None, *args):
47 return self.create_help(self.basePath, req, *args)
49 def create_help(self, uri, req, *args):
50 if req and not isinstance(req, str):
51 req = json.dumps(req.format())
52 res = self.fetch(self._update_uri(uri, *args),
54 body=req if req else json.dumps(None),
57 return self._get_return(res, self.create_res)
60 res = self.fetch(self._get_uri(*args),
65 new_args, num = self._get_valid_args(*args)
67 if num != self._need_arg_num(self.basePath) else self.list_res
68 return self._get_return(res, inner())
70 def query(self, query):
71 res = self.fetch(self._get_query_uri(query),
74 return self._get_return(res, self.list_res)
76 def update(self, new=None, *args):
79 res = self.fetch(self._get_uri(*args),
83 return self._get_return(res, self.update_res)
85 def delete(self, *args):
86 res = self.fetch(self._get_uri(*args),
89 return res.code, res.body
92 def _get_valid_args(*args):
93 new_args = tuple(['%s' % arg for arg in args if arg is not None])
94 return new_args, len(new_args)
96 def _need_arg_num(self, uri):
97 return uri.count('%s')
99 def _get_query_uri(self, query):
100 return self.basePath + '?' + query
102 def _get_uri(self, *args):
103 return self._update_uri(self.basePath, *args)
105 def _update_uri(self, uri, *args):
107 new_args, num = self._get_valid_args(*args)
108 if num != self._need_arg_num(uri):
111 return r_uri % tuple(['%s' % arg for arg in new_args])
113 def _get_return(self, res, cls):
116 return code, self._get_return_body(code, body, cls)
119 def _get_return_body(code, body, cls):
120 return cls.from_dict(json.loads(body)) if code < 300 and cls else body
122 def assert_href(self, body):
123 self.assertIn(self.basePath, body.href)
125 def assert_create_body(self, body, req=None, *args):
130 if inspect.isclass(req):
131 resource_name = req.name
132 elif isinstance(req, dict):
133 resource_name = req['name']
134 elif isinstance(req, str):
135 resource_name = json.loads(req)['name']
136 new_args = args + tuple([resource_name])
137 self.assertIn(self._get_uri(*new_args), body.href)
141 fake_pymongo.pods.clear()
142 fake_pymongo.projects.clear()
143 fake_pymongo.testcases.clear()
144 fake_pymongo.results.clear()
145 fake_pymongo.scenarios.clear()