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 ##############################################################################
13 from tornado import testing
15 from opnfv_testapi.resources import models
16 from opnfv_testapi.tests.unit import fake_pymongo
19 class TestBase(testing.AsyncHTTPTestCase):
20 headers = {'Content-Type': 'application/json; charset=UTF-8'}
25 self.create_res = models.CreateResponse
28 self.update_res = None
31 self.addCleanup(self._clear)
32 super(TestBase, self).setUp()
35 self.db_patcher.stop()
36 self.config_patcher.stop()
38 def _patch_server(self):
40 config = path.join(path.dirname(__file__), '../common/normal.ini')
41 self.config_patcher = mock.patch(
42 'argparse.ArgumentParser.parse_known_args',
43 return_value=(argparse.Namespace(config_file=config), None))
44 self.db_patcher = mock.patch('opnfv_testapi.cmd.server.get_db',
46 self.config_patcher.start()
47 self.db_patcher.start()
49 def set_config_file(self):
50 self.config_file = 'normal.ini'
57 from opnfv_testapi.cmd import server
58 return server.make_app()
60 def create_d(self, *args):
61 return self.create(self.req_d, *args)
63 def create_e(self, *args):
64 return self.create(self.req_e, *args)
66 def create(self, req=None, *args):
67 return self.create_help(self.basePath, req, *args)
69 def create_help(self, uri, req, *args):
70 if req and not isinstance(req, str) and hasattr(req, 'format'):
72 res = self.fetch(self._update_uri(uri, *args),
77 return self._get_return(res, self.create_res)
80 res = self.fetch(self._get_uri(*args),
85 new_args, num = self._get_valid_args(*args)
87 if num != self._need_arg_num(self.basePath) else self.list_res
88 return self._get_return(res, inner())
90 def query(self, query):
91 res = self.fetch(self._get_query_uri(query),
94 return self._get_return(res, self.list_res)
96 def update(self, new=None, *args):
99 res = self.fetch(self._get_uri(*args),
101 body=json.dumps(new),
102 headers=self.headers)
103 return self._get_return(res, self.update_res)
105 def delete(self, *args):
106 res = self.fetch(self._get_uri(*args),
108 headers=self.headers)
109 return res.code, res.body
112 def _get_valid_args(*args):
113 new_args = tuple(['%s' % arg for arg in args if arg is not None])
114 return new_args, len(new_args)
116 def _need_arg_num(self, uri):
117 return uri.count('%s')
119 def _get_query_uri(self, query):
120 return self.basePath + '?' + query if query else self.basePath
122 def _get_uri(self, *args):
123 return self._update_uri(self.basePath, *args)
125 def _update_uri(self, uri, *args):
127 new_args, num = self._get_valid_args(*args)
128 if num != self._need_arg_num(uri):
131 return r_uri % tuple(['%s' % arg for arg in new_args])
133 def _get_return(self, res, cls):
136 return code, self._get_return_body(code, body, cls)
139 def _get_return_body(code, body, cls):
140 return cls.from_dict(json.loads(body)) if code < 300 and cls else body
142 def assert_href(self, body):
143 self.assertIn(self.basePath, body.href)
145 def assert_create_body(self, body, req=None, *args):
150 if inspect.isclass(req):
151 resource_name = req.name
152 elif isinstance(req, dict):
153 resource_name = req['name']
154 elif isinstance(req, str):
155 resource_name = json.loads(req)['name']
156 new_args = args + tuple([resource_name])
157 self.assertIn(self._get_uri(*new_args), body.href)
161 fake_pymongo.pods.clear()
162 fake_pymongo.projects.clear()
163 fake_pymongo.testcases.clear()
164 fake_pymongo.results.clear()
165 fake_pymongo.scenarios.clear()