From 0cf6b232ac9cf128ee9183a27c08f4f74ab2e2e6 Mon Sep 17 00:00:00 2001 From: grakiss Date: Thu, 28 Sep 2017 03:47:54 -0400 Subject: add api&web services for cvp JIRA: DOVETAIL-512 add api&web services for cvp Change-Id: I9ef9525e980fe61dc3108035ef9a3ff8783b2697 Signed-off-by: grakiss --- .../tests/unit/resources/test_base.py | 161 +++++++++++++++++++++ 1 file changed, 161 insertions(+) create mode 100644 cvp/opnfv_testapi/tests/unit/resources/test_base.py (limited to 'cvp/opnfv_testapi/tests/unit/resources/test_base.py') diff --git a/cvp/opnfv_testapi/tests/unit/resources/test_base.py b/cvp/opnfv_testapi/tests/unit/resources/test_base.py new file mode 100644 index 00000000..dcec4e95 --- /dev/null +++ b/cvp/opnfv_testapi/tests/unit/resources/test_base.py @@ -0,0 +1,161 @@ +############################################################################## +# Copyright (c) 2016 ZTE Corporation +# feng.xiaowei@zte.com.cn +# All rights reserved. This program and the accompanying materials +# are made available under the terms of the Apache License, Version 2.0 +# which accompanies this distribution, and is available at +# http://www.apache.org/licenses/LICENSE-2.0 +############################################################################## +import json +from os import path + +import mock +from tornado import testing + +from opnfv_testapi.resources import models +from opnfv_testapi.tests.unit import fake_pymongo + + +class TestBase(testing.AsyncHTTPTestCase): + headers = {'Content-Type': 'application/json; charset=UTF-8'} + + def setUp(self): + self._patch_server() + self.basePath = '' + self.create_res = models.CreateResponse + self.get_res = None + self.list_res = None + self.update_res = None + self.req_d = None + self.req_e = None + self.addCleanup(self._clear) + super(TestBase, self).setUp() + + def tearDown(self): + self.db_patcher.stop() + self.config_patcher.stop() + + def _patch_server(self): + import argparse + config = path.join(path.dirname(__file__), '../common/normal.ini') + self.config_patcher = mock.patch( + 'argparse.ArgumentParser.parse_known_args', + return_value=(argparse.Namespace(config_file=config), None)) + self.db_patcher = mock.patch('opnfv_testapi.db.api.DB', + fake_pymongo) + self.config_patcher.start() + self.db_patcher.start() + + def set_config_file(self): + self.config_file = 'normal.ini' + + def get_app(self): + from opnfv_testapi.cmd import server + return server.make_app() + + def create_d(self, *args): + return self.create(self.req_d, *args) + + def create_e(self, *args): + return self.create(self.req_e, *args) + + def create(self, req=None, *args): + return self.create_help(self.basePath, req, *args) + + def create_help(self, uri, req, *args): + if req and not isinstance(req, str) and hasattr(req, 'format'): + req = req.format() + res = self.fetch(self._update_uri(uri, *args), + method='POST', + body=json.dumps(req), + headers=self.headers) + + return self._get_return(res, self.create_res) + + def get(self, *args): + res = self.fetch(self._get_uri(*args), + method='GET', + headers=self.headers) + + def inner(): + new_args, num = self._get_valid_args(*args) + return self.get_res \ + if num != self._need_arg_num(self.basePath) else self.list_res + return self._get_return(res, inner()) + + def query(self, query): + res = self.fetch(self._get_query_uri(query), + method='GET', + headers=self.headers) + return self._get_return(res, self.list_res) + + def update(self, new=None, *args): + if new: + new = new.format() + res = self.fetch(self._get_uri(*args), + method='PUT', + body=json.dumps(new), + headers=self.headers) + return self._get_return(res, self.update_res) + + def delete(self, *args): + res = self.fetch(self._get_uri(*args), + method='DELETE', + headers=self.headers) + return res.code, res.body + + @staticmethod + def _get_valid_args(*args): + new_args = tuple(['%s' % arg for arg in args if arg is not None]) + return new_args, len(new_args) + + def _need_arg_num(self, uri): + return uri.count('%s') + + def _get_query_uri(self, query): + return self.basePath + '?' + query if query else self.basePath + + def _get_uri(self, *args): + return self._update_uri(self.basePath, *args) + + def _update_uri(self, uri, *args): + r_uri = uri + new_args, num = self._get_valid_args(*args) + if num != self._need_arg_num(uri): + r_uri += '/%s' + + return r_uri % tuple(['%s' % arg for arg in new_args]) + + def _get_return(self, res, cls): + code = res.code + body = res.body + return code, self._get_return_body(code, body, cls) + + @staticmethod + def _get_return_body(code, body, cls): + return cls.from_dict(json.loads(body)) if code < 300 and cls else body + + def assert_href(self, body): + self.assertIn(self.basePath, body.href) + + def assert_create_body(self, body, req=None, *args): + import inspect + if not req: + req = self.req_d + resource_name = '' + if inspect.isclass(req): + resource_name = req.name + elif isinstance(req, dict): + resource_name = req['name'] + elif isinstance(req, str): + resource_name = json.loads(req)['name'] + new_args = args + tuple([resource_name]) + self.assertIn(self._get_uri(*new_args), body.href) + + @staticmethod + def _clear(): + fake_pymongo.pods.clear() + fake_pymongo.projects.clear() + fake_pymongo.testcases.clear() + fake_pymongo.results.clear() + fake_pymongo.scenarios.clear() -- cgit 1.2.3-korg