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 ##############################################################################
9 from datetime import datetime
13 from bson.objectid import ObjectId
15 from tornado import testing
17 from opnfv_testapi.models import base_models
18 from opnfv_testapi.models import pod_models
19 from opnfv_testapi.tests.unit import fake_pymongo
22 class TestBase(testing.AsyncHTTPTestCase):
23 headers = {'Content-Type': 'application/json; charset=UTF-8'}
28 self.create_res = base_models.CreateResponse
31 self.update_res = None
32 self.pod_d = pod_models.Pod(name='zte-pod1',
38 create_date=str(datetime.now()))
39 self.pod_e = pod_models.Pod(name='zte-pod2',
45 create_date=str(datetime.now()))
48 self.addCleanup(self._clear)
49 super(TestBase, self).setUp()
50 fake_pymongo.users.insert({"user": "ValidUser",
51 'email': 'validuser@lf.com',
52 'fullname': 'Valid User',
54 'opnfv-testapi-users',
55 'opnfv-gerrit-functest-submitters',
56 'opnfv-gerrit-qtip-contributors']
60 self.db_patcher.stop()
61 self.config_patcher.stop()
63 def _patch_server(self):
65 config = path.join(path.dirname(__file__),
66 '../../../../etc/config.ini')
67 self.config_patcher = mock.patch(
68 'argparse.ArgumentParser.parse_known_args',
69 return_value=(argparse.Namespace(config_file=config), None))
70 self.db_patcher = mock.patch('opnfv_testapi.db.api.DB',
72 self.config_patcher.start()
73 self.db_patcher.start()
76 from opnfv_testapi.cmd import server
77 return server.make_app()
79 def create_d(self, *args):
80 return self.create(self.req_d, *args)
82 def create_e(self, *args):
83 return self.create(self.req_e, *args)
85 def create(self, req=None, *args):
86 return self.create_help(self.basePath, req, *args)
88 def create_help(self, uri, req, *args):
89 return self.post_direct_url(self._update_uri(uri, *args), req)
91 def post_direct_url(self, url, req):
92 if req and not isinstance(req, str) and hasattr(req, 'format'):
99 return self._get_return(res, self.create_res)
101 def get(self, *args):
102 res = self.fetch(self._get_uri(*args),
104 headers=self.headers)
107 new_args, num = self._get_valid_args(*args)
108 return self.get_res \
109 if num != self._need_arg_num(self.basePath) else self.list_res
110 return self._get_return(res, inner())
112 def query(self, query):
113 res = self.fetch(self._get_query_uri(query),
115 headers=self.headers)
116 return self._get_return(res, self.list_res)
118 def update_direct_url(self, url, new=None):
119 if new and hasattr(new, 'format'):
121 res = self.fetch(url,
123 body=json.dumps(new),
124 headers=self.headers)
125 return self._get_return(res, self.update_res)
127 def update(self, new=None, *args):
128 return self.update_direct_url(self._get_uri(*args), new)
130 def delete_direct_url(self, url, body):
132 res = self.fetch(url,
134 body=json.dumps(body),
135 headers=self.headers,
136 allow_nonstandard_methods=True)
138 res = self.fetch(url,
140 headers=self.headers)
142 return res.code, res.body
144 def delete(self, *args):
145 return self.delete_direct_url(self._get_uri(*args), None)
148 def _get_valid_args(*args):
149 new_args = tuple(['%s' % arg for arg in args if arg is not None])
150 return new_args, len(new_args)
152 def _need_arg_num(self, uri):
153 return uri.count('%s')
155 def _get_query_uri(self, query):
156 return self.basePath + '?' + query if query else self.basePath
158 def _get_uri(self, *args):
159 return self._update_uri(self.basePath, *args)
161 def _update_uri(self, uri, *args):
163 new_args, num = self._get_valid_args(*args)
164 if num != self._need_arg_num(uri):
167 return r_uri % tuple(['%s' % arg for arg in new_args])
169 def _get_return(self, res, cls):
173 return code, self._get_return_body(code, body, cls)
178 def _get_return_body(code, body, cls):
179 return cls.from_dict(json.loads(body)) if code < 300 and cls else body
181 def assert_href(self, body):
182 self.assertIn(self.basePath, body.href)
184 def assert_create_body(self, body, req=None, *args):
189 if inspect.isclass(req):
190 resource_name = req.name
191 elif isinstance(req, dict):
192 resource_name = req['name']
193 elif isinstance(req, str):
194 resource_name = json.loads(req)['name']
195 new_args = args + tuple([resource_name])
196 self.assertIn(self._get_uri(*new_args), body.href)
200 fake_pymongo.pods.clear()
201 fake_pymongo.projects.clear()
202 fake_pymongo.testcases.clear()
203 fake_pymongo.results.clear()
204 fake_pymongo.scenarios.clear()