summaryrefslogtreecommitdiffstats
path: root/utils/test/testapi
diff options
context:
space:
mode:
Diffstat (limited to 'utils/test/testapi')
-rw-r--r--utils/test/testapi/opnfv_testapi/common/check.py111
-rw-r--r--utils/test/testapi/opnfv_testapi/resources/handlers.py96
-rw-r--r--utils/test/testapi/opnfv_testapi/resources/pod_handlers.py20
-rw-r--r--utils/test/testapi/opnfv_testapi/resources/project_handlers.py21
-rw-r--r--utils/test/testapi/opnfv_testapi/resources/result_handlers.py43
-rw-r--r--utils/test/testapi/opnfv_testapi/resources/scenario_handlers.py29
-rw-r--r--utils/test/testapi/opnfv_testapi/resources/testcase_handlers.py42
-rw-r--r--utils/test/testapi/opnfv_testapi/tests/unit/test_result.py208
-rw-r--r--utils/test/testapi/opnfv_testapi/tests/unit/test_token.py74
-rw-r--r--utils/test/testapi/opnfv_testapi/tests/unit/test_version.py9
10 files changed, 335 insertions, 318 deletions
diff --git a/utils/test/testapi/opnfv_testapi/common/check.py b/utils/test/testapi/opnfv_testapi/common/check.py
new file mode 100644
index 000000000..be4b1df12
--- /dev/null
+++ b/utils/test/testapi/opnfv_testapi/common/check.py
@@ -0,0 +1,111 @@
+##############################################################################
+# Copyright (c) 2017 ZTE Corp
+# 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 functools
+
+from tornado import web, gen
+
+from opnfv_testapi.common import raises, message
+
+
+def authenticate(method):
+ @web.asynchronous
+ @gen.coroutine
+ @functools.wraps(method)
+ def wrapper(self, *args, **kwargs):
+ if self.auth:
+ try:
+ token = self.request.headers['X-Auth-Token']
+ except KeyError:
+ raises.Unauthorized(message.unauthorized())
+ query = {'access_token': token}
+ check = yield self._eval_db_find_one(query, 'tokens')
+ if not check:
+ raises.Forbidden(message.invalid_token())
+ ret = yield gen.coroutine(method)(self, *args, **kwargs)
+ raise gen.Return(ret)
+ return wrapper
+
+
+def not_exist(xstep):
+ @functools.wraps(xstep)
+ def wrap(self, *args, **kwargs):
+ query = kwargs.get('query')
+ data = yield self._eval_db_find_one(query)
+ if not data:
+ raises.NotFound(message.not_found(self.table, query))
+ ret = yield gen.coroutine(xstep)(self, data, *args, **kwargs)
+ raise gen.Return(ret)
+
+ return wrap
+
+
+def no_body(xstep):
+ @functools.wraps(xstep)
+ def wrap(self, *args, **kwargs):
+ if self.json_args is None:
+ raises.BadRequest(message.no_body())
+ ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
+ raise gen.Return(ret)
+
+ return wrap
+
+
+def miss_fields(xstep):
+ @functools.wraps(xstep)
+ def wrap(self, *args, **kwargs):
+ fields = kwargs.get('miss_fields')
+ if fields:
+ for miss in fields:
+ miss_data = self.json_args.get(miss)
+ if miss_data is None or miss_data == '':
+ raises.BadRequest(message.missing(miss))
+ ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
+ raise gen.Return(ret)
+ return wrap
+
+
+def carriers_exist(xstep):
+ @functools.wraps(xstep)
+ def wrap(self, *args, **kwargs):
+ carriers = kwargs.get('carriers')
+ if carriers:
+ for table, query in carriers:
+ exist = yield self._eval_db_find_one(query(), table)
+ if not exist:
+ raises.Forbidden(message.not_found(table, query()))
+ ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
+ raise gen.Return(ret)
+ return wrap
+
+
+def new_not_exists(xstep):
+ @functools.wraps(xstep)
+ def wrap(self, *args, **kwargs):
+ query = kwargs.get('query')
+ if query:
+ to_data = yield self._eval_db_find_one(query())
+ if to_data:
+ raises.Forbidden(message.exist(self.table, query()))
+ ret = yield gen.coroutine(xstep)(self, *args, **kwargs)
+ raise gen.Return(ret)
+ return wrap
+
+
+def updated_one_not_exist(xstep):
+ @functools.wraps(xstep)
+ def wrap(self, data, *args, **kwargs):
+ db_keys = kwargs.get('db_keys')
+ query = self._update_query(db_keys, data)
+ if query:
+ to_data = yield self._eval_db_find_one(query)
+ if to_data:
+ raises.Forbidden(message.exist(self.table, query))
+ ret = yield gen.coroutine(xstep)(self, data, *args, **kwargs)
+ raise gen.Return(ret)
+ return wrap
diff --git a/utils/test/testapi/opnfv_testapi/resources/handlers.py b/utils/test/testapi/opnfv_testapi/resources/handlers.py
index 522bbe7f5..955fbbef7 100644
--- a/utils/test/testapi/opnfv_testapi/resources/handlers.py
+++ b/utils/test/testapi/opnfv_testapi/resources/handlers.py
@@ -21,13 +21,13 @@
##############################################################################
from datetime import datetime
-import functools
import json
from tornado import gen
from tornado import web
import models
+from opnfv_testapi.common import check
from opnfv_testapi.common import message
from opnfv_testapi.common import raises
from opnfv_testapi.tornado_swagger import swagger
@@ -73,48 +73,20 @@ class GenericApiHandler(web.RequestHandler):
cls_data = self.table_cls.from_dict(data)
return cls_data.format_http()
- def authenticate(method):
- @web.asynchronous
- @gen.coroutine
- @functools.wraps(method)
- def wrapper(self, *args, **kwargs):
- if self.auth:
- try:
- token = self.request.headers['X-Auth-Token']
- except KeyError:
- raises.Unauthorized(message.unauthorized())
- query = {'access_token': token}
- check = yield self._eval_db_find_one(query, 'tokens')
- if not check:
- raises.Forbidden(message.invalid_token())
- ret = yield gen.coroutine(method)(self, *args, **kwargs)
- raise gen.Return(ret)
- return wrapper
-
- @authenticate
- def _create(self, miss_checks, db_checks, **kwargs):
+ @check.authenticate
+ @check.no_body
+ @check.miss_fields
+ @check.carriers_exist
+ @check.new_not_exists
+ def _create(self, **kwargs):
"""
:param miss_checks: [miss1, miss2]
:param db_checks: [(table, exist, query, error)]
"""
- if self.json_args is None:
- raises.BadRequest(message.no_body())
-
data = self.table_cls.from_dict(self.json_args)
- for miss in miss_checks:
- miss_data = data.__getattribute__(miss)
- if miss_data is None or miss_data == '':
- raises.BadRequest(message.missing(miss))
-
for k, v in kwargs.iteritems():
data.__setattr__(k, v)
- for table, exist, query, error in db_checks:
- check = yield self._eval_db_find_one(query(data), table)
- if (exist and not check) or (not exist and check):
- code, msg = error(data)
- raises.CodeTBD(code, msg)
-
if self.table != 'results':
data.creation_date = datetime.now()
_id = yield self._eval_db(self.table, 'insert', data.format(),
@@ -146,47 +118,27 @@ class GenericApiHandler(web.RequestHandler):
@web.asynchronous
@gen.coroutine
- def _get_one(self, query):
- data = yield self._eval_db_find_one(query)
- if data is None:
- raises.NotFound(message.not_found(self.table, query))
+ @check.not_exist
+ def _get_one(self, data, query=None):
self.finish_request(self.format_data(data))
- @authenticate
- def _delete(self, query):
- data = yield self._eval_db_find_one(query)
- if data is None:
- raises.NotFound(message.not_found(self.table, query))
-
+ @check.authenticate
+ @check.not_exist
+ def _delete(self, data, query=None):
yield self._eval_db(self.table, 'remove', query)
self.finish_request()
- @authenticate
- def _update(self, query, db_keys):
- if self.json_args is None:
- raises.BadRequest(message.no_body())
-
- # check old data exist
- from_data = yield self._eval_db_find_one(query)
- if from_data is None:
- raises.NotFound(message.not_found(self.table, query))
-
- data = self.table_cls.from_dict(from_data)
- # check new data exist
- equal, new_query = self._update_query(db_keys, data)
- if not equal:
- to_data = yield self._eval_db_find_one(new_query)
- if to_data is not None:
- raises.Forbidden(message.exist(self.table, new_query))
-
- # we merge the whole document """
- edit_request = self._update_requests(data)
-
- """ Updating the DB """
- yield self._eval_db(self.table, 'update', query, edit_request,
+ @check.authenticate
+ @check.no_body
+ @check.not_exist
+ @check.updated_one_not_exist
+ def _update(self, data, query=None, **kwargs):
+ data = self.table_cls.from_dict(data)
+ update_req = self._update_requests(data)
+ yield self._eval_db(self.table, 'update', query, update_req,
check_keys=False)
- edit_request['_id'] = str(data._id)
- self.finish_request(edit_request)
+ update_req['_id'] = str(data._id)
+ self.finish_request(update_req)
def _update_requests(self, data):
request = dict()
@@ -219,13 +171,13 @@ class GenericApiHandler(web.RequestHandler):
equal = True
for key in keys:
new = self.json_args.get(key)
- old = data.__getattribute__(key)
+ old = data.get(key)
if new is None:
new = old
elif new != old:
equal = False
query[key] = new
- return equal, query
+ return query if not equal else dict()
def _eval_db(self, table, method, *args, **kwargs):
exec_collection = self.db.__getattr__(table)
diff --git a/utils/test/testapi/opnfv_testapi/resources/pod_handlers.py b/utils/test/testapi/opnfv_testapi/resources/pod_handlers.py
index 2c303c934..e21841d33 100644
--- a/utils/test/testapi/opnfv_testapi/resources/pod_handlers.py
+++ b/utils/test/testapi/opnfv_testapi/resources/pod_handlers.py
@@ -6,10 +6,7 @@
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-import httplib
-
import handlers
-from opnfv_testapi.common import message
from opnfv_testapi.tornado_swagger import swagger
import pod_models
@@ -43,15 +40,10 @@ class PodCLHandler(GenericPodHandler):
@raise 403: pod already exists
@raise 400: body or name not provided
"""
- def query(data):
- return {'name': data.name}
-
- def error(data):
- return httplib.FORBIDDEN, message.exist('pod', data.name)
-
- miss_checks = ['name']
- db_checks = [(self.table, False, query, error)]
- self._create(miss_checks, db_checks)
+ def query():
+ return {'name': self.json_args.get('name')}
+ miss_fields = ['name']
+ self._create(miss_fields=miss_fields, query=query)
class PodGURHandler(GenericPodHandler):
@@ -63,9 +55,7 @@ class PodGURHandler(GenericPodHandler):
@return 200: pod exist
@raise 404: pod not exist
"""
- query = dict()
- query['name'] = pod_name
- self._get_one(query)
+ self._get_one(query={'name': pod_name})
def delete(self, pod_name):
""" Remove a POD
diff --git a/utils/test/testapi/opnfv_testapi/resources/project_handlers.py b/utils/test/testapi/opnfv_testapi/resources/project_handlers.py
index 59e0b88e5..d79cd3b61 100644
--- a/utils/test/testapi/opnfv_testapi/resources/project_handlers.py
+++ b/utils/test/testapi/opnfv_testapi/resources/project_handlers.py
@@ -6,10 +6,8 @@
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-import httplib
import handlers
-from opnfv_testapi.common import message
from opnfv_testapi.tornado_swagger import swagger
import project_models
@@ -45,15 +43,10 @@ class ProjectCLHandler(GenericProjectHandler):
@raise 403: project already exists
@raise 400: body or name not provided
"""
- def query(data):
- return {'name': data.name}
-
- def error(data):
- return httplib.FORBIDDEN, message.exist('project', data.name)
-
- miss_checks = ['name']
- db_checks = [(self.table, False, query, error)]
- self._create(miss_checks, db_checks)
+ def query():
+ return {'name': self.json_args.get('name')}
+ miss_fields = ['name']
+ self._create(miss_fields=miss_fields, query=query)
class ProjectGURHandler(GenericProjectHandler):
@@ -65,7 +58,7 @@ class ProjectGURHandler(GenericProjectHandler):
@return 200: project exist
@raise 404: project not exist
"""
- self._get_one({'name': project_name})
+ self._get_one(query={'name': project_name})
@swagger.operation(nickname="updateProjectByName")
def put(self, project_name):
@@ -81,7 +74,7 @@ class ProjectGURHandler(GenericProjectHandler):
"""
query = {'name': project_name}
db_keys = ['name']
- self._update(query, db_keys)
+ self._update(query=query, db_keys=db_keys)
@swagger.operation(nickname='deleteProjectByName')
def delete(self, project_name):
@@ -90,4 +83,4 @@ class ProjectGURHandler(GenericProjectHandler):
@return 200: delete success
@raise 404: project not exist
"""
- self._delete({'name': project_name})
+ self._delete(query={'name': project_name})
diff --git a/utils/test/testapi/opnfv_testapi/resources/result_handlers.py b/utils/test/testapi/opnfv_testapi/resources/result_handlers.py
index fb5ed9ec7..214706f5f 100644
--- a/utils/test/testapi/opnfv_testapi/resources/result_handlers.py
+++ b/utils/test/testapi/opnfv_testapi/resources/result_handlers.py
@@ -8,7 +8,6 @@
##############################################################################
from datetime import datetime
from datetime import timedelta
-import httplib
from bson import objectid
@@ -127,7 +126,9 @@ class ResultsCLHandler(GenericResultHandler):
if last is not None:
last = self.get_int('last', last)
- self._list(self.set_query(), sort=[('start_date', -1)], last=last)
+ self._list(query=self.set_query(),
+ sort=[('start_date', -1)],
+ last=last)
@swagger.operation(nickname="createTestResult")
def post(self):
@@ -141,31 +142,21 @@ class ResultsCLHandler(GenericResultHandler):
@raise 404: pod/project/testcase not exist
@raise 400: body/pod_name/project_name/case_name not provided
"""
- def pod_query(data):
- return {'name': data.pod_name}
+ def pod_query():
+ return {'name': self.json_args.get('pod_name')}
- def pod_error(data):
- return httplib.FORBIDDEN, message.not_found('pod', data.pod_name)
+ def project_query():
+ return {'name': self.json_args.get('project_name')}
- def project_query(data):
- return {'name': data.project_name}
+ def testcase_query():
+ return {'project_name': self.json_args.get('project_name'),
+ 'name': self.json_args.get('case_name')}
- def project_error(data):
- return httplib.FORBIDDEN, message.not_found('project',
- data.project_name)
-
- def testcase_query(data):
- return {'project_name': data.project_name, 'name': data.case_name}
-
- def testcase_error(data):
- return httplib.FORBIDDEN, message.not_found('testcase',
- data.case_name)
-
- miss_checks = ['pod_name', 'project_name', 'case_name']
- db_checks = [('pods', True, pod_query, pod_error),
- ('projects', True, project_query, project_error),
- ('testcases', True, testcase_query, testcase_error)]
- self._create(miss_checks, db_checks)
+ miss_fields = ['pod_name', 'project_name', 'case_name']
+ carriers = [('pods', pod_query),
+ ('projects', project_query),
+ ('testcases', testcase_query)]
+ self._create(miss_fields=miss_fields, carriers=carriers)
class ResultsGURHandler(GenericResultHandler):
@@ -179,7 +170,7 @@ class ResultsGURHandler(GenericResultHandler):
"""
query = dict()
query["_id"] = objectid.ObjectId(result_id)
- self._get_one(query)
+ self._get_one(query=query)
@swagger.operation(nickname="updateTestResultById")
def put(self, result_id):
@@ -195,4 +186,4 @@ class ResultsGURHandler(GenericResultHandler):
"""
query = {'_id': objectid.ObjectId(result_id)}
db_keys = []
- self._update(query, db_keys)
+ self._update(query=query, db_keys=db_keys)
diff --git a/utils/test/testapi/opnfv_testapi/resources/scenario_handlers.py b/utils/test/testapi/opnfv_testapi/resources/scenario_handlers.py
index bad79fdc6..5d420a56e 100644
--- a/utils/test/testapi/opnfv_testapi/resources/scenario_handlers.py
+++ b/utils/test/testapi/opnfv_testapi/resources/scenario_handlers.py
@@ -1,5 +1,4 @@
import functools
-import httplib
from opnfv_testapi.common import message
from opnfv_testapi.common import raises
@@ -65,7 +64,7 @@ class ScenariosCLHandler(GenericScenarioHandler):
query['installers'] = {'$elemMatch': elem_query}
return query
- self._list(_set_query())
+ self._list(query=_set_query())
@swagger.operation(nickname="createScenario")
def post(self):
@@ -79,15 +78,10 @@ class ScenariosCLHandler(GenericScenarioHandler):
@raise 403: scenario already exists
@raise 400: body or name not provided
"""
- def query(data):
- return {'name': data.name}
-
- def error(data):
- return httplib.FORBIDDEN, message.exist('scenario', data.name)
-
- miss_checks = ['name']
- db_checks = [(self.table, False, query, error)]
- self._create(miss_checks=miss_checks, db_checks=db_checks)
+ def query():
+ return {'name': self.json_args.get('name')}
+ miss_fields = ['name']
+ self._create(miss_fields=miss_fields, query=query)
class ScenarioGURHandler(GenericScenarioHandler):
@@ -99,7 +93,7 @@ class ScenarioGURHandler(GenericScenarioHandler):
@return 200: scenario exist
@raise 404: scenario not exist
"""
- self._get_one({'name': name})
+ self._get_one(query={'name': name})
pass
@swagger.operation(nickname="updateScenarioByName")
@@ -116,7 +110,7 @@ class ScenarioGURHandler(GenericScenarioHandler):
"""
query = {'name': name}
db_keys = ['name']
- self._update(query, db_keys)
+ self._update(query=query, db_keys=db_keys)
@swagger.operation(nickname="deleteScenarioByName")
def delete(self, name):
@@ -126,19 +120,16 @@ class ScenarioGURHandler(GenericScenarioHandler):
@raise 404: scenario not exist:
"""
- query = {'name': name}
- self._delete(query)
+ self._delete(query={'name': name})
def _update_query(self, keys, data):
query = dict()
- equal = True
if self._is_rename():
new = self._term.get('name')
- if data.name != new:
- equal = False
+ if data.get('name') != new:
query['name'] = new
- return equal, query
+ return query
def _update_requests(self, data):
updates = {
diff --git a/utils/test/testapi/opnfv_testapi/resources/testcase_handlers.py b/utils/test/testapi/opnfv_testapi/resources/testcase_handlers.py
index bc22b74e2..9399326f0 100644
--- a/utils/test/testapi/opnfv_testapi/resources/testcase_handlers.py
+++ b/utils/test/testapi/opnfv_testapi/resources/testcase_handlers.py
@@ -6,9 +6,7 @@
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-import httplib
-from opnfv_testapi.common import message
from opnfv_testapi.resources import handlers
from opnfv_testapi.resources import testcase_models
from opnfv_testapi.tornado_swagger import swagger
@@ -32,9 +30,7 @@ class TestcaseCLHandler(GenericTestcaseHandler):
empty list is no testcase exist in this project
@rtype: L{TestCases}
"""
- query = dict()
- query['project_name'] = project_name
- self._list(query)
+ self._list(query={'project_name': project_name})
@swagger.operation(nickname="createTestCase")
def post(self, project_name):
@@ -49,26 +45,18 @@ class TestcaseCLHandler(GenericTestcaseHandler):
or testcase already exists in this project
@raise 400: body or name not provided
"""
- def p_query(data):
- return {'name': data.project_name}
-
- def tc_query(data):
- return {
- 'project_name': data.project_name,
- 'name': data.name
- }
-
- def p_error(data):
- return httplib.FORBIDDEN, message.not_found('project',
- data.project_name)
-
- def tc_error(data):
- return httplib.FORBIDDEN, message.exist('testcase', data.name)
+ def project_query():
+ return {'name': project_name}
- miss_checks = ['name']
- db_checks = [(self.db_projects, True, p_query, p_error),
- (self.db_testcases, False, tc_query, tc_error)]
- self._create(miss_checks, db_checks, project_name=project_name)
+ def testcase_query():
+ return {'project_name': project_name,
+ 'name': self.json_args.get('name')}
+ miss_fields = ['name']
+ carriers = [(self.db_projects, project_query)]
+ self._create(miss_fields=miss_fields,
+ carriers=carriers,
+ query=testcase_query,
+ project_name=project_name)
class TestcaseGURHandler(GenericTestcaseHandler):
@@ -84,7 +72,7 @@ class TestcaseGURHandler(GenericTestcaseHandler):
query = dict()
query['project_name'] = project_name
query["name"] = case_name
- self._get_one(query)
+ self._get_one(query=query)
@swagger.operation(nickname="updateTestCaseByName")
def put(self, project_name, case_name):
@@ -102,7 +90,7 @@ class TestcaseGURHandler(GenericTestcaseHandler):
"""
query = {'project_name': project_name, 'name': case_name}
db_keys = ['name', 'project_name']
- self._update(query, db_keys)
+ self._update(query=query, db_keys=db_keys)
@swagger.operation(nickname='deleteTestCaseByName')
def delete(self, project_name, case_name):
@@ -112,4 +100,4 @@ class TestcaseGURHandler(GenericTestcaseHandler):
@raise 404: testcase not exist
"""
query = {'project_name': project_name, 'name': case_name}
- self._delete(query)
+ self._delete(query=query)
diff --git a/utils/test/testapi/opnfv_testapi/tests/unit/test_result.py b/utils/test/testapi/opnfv_testapi/tests/unit/test_result.py
index 940279cd4..ef2ce307e 100644
--- a/utils/test/testapi/opnfv_testapi/tests/unit/test_result.py
+++ b/utils/test/testapi/opnfv_testapi/tests/unit/test_result.py
@@ -17,6 +17,7 @@ from opnfv_testapi.resources import project_models
from opnfv_testapi.resources import result_models
from opnfv_testapi.resources import testcase_models
from opnfv_testapi.tests.unit import test_base as base
+from opnfv_testapi.tests.unit import executor
class Details(object):
@@ -99,8 +100,7 @@ class TestResultBase(base.TestBase):
self.req_testcase,
self.project)
- def assert_res(self, code, result, req=None):
- self.assertEqual(code, httplib.OK)
+ def assert_res(self, result, req=None):
if req is None:
req = self.req_d
self.assertEqual(result.pod_name, req.pod_name)
@@ -133,65 +133,57 @@ class TestResultBase(base.TestBase):
class TestResultCreate(TestResultBase):
+ @executor.create(httplib.BAD_REQUEST, message.no_body())
def test_nobody(self):
- (code, body) = self.create(None)
- self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn(message.no_body(), body)
+ return None
+ @executor.create(httplib.BAD_REQUEST, message.missing('pod_name'))
def test_podNotProvided(self):
req = self.req_d
req.pod_name = None
- (code, body) = self.create(req)
- self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn(message.missing('pod_name'), body)
+ return req
+ @executor.create(httplib.BAD_REQUEST, message.missing('project_name'))
def test_projectNotProvided(self):
req = self.req_d
req.project_name = None
- (code, body) = self.create(req)
- self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn(message.missing('project_name'), body)
+ return req
+ @executor.create(httplib.BAD_REQUEST, message.missing('case_name'))
def test_testcaseNotProvided(self):
req = self.req_d
req.case_name = None
- (code, body) = self.create(req)
- self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn(message.missing('case_name'), body)
+ return req
+ @executor.create(httplib.FORBIDDEN, message.not_found_base)
def test_noPod(self):
req = self.req_d
req.pod_name = 'notExistPod'
- (code, body) = self.create(req)
- self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn(message.not_found_base, body)
+ return req
+ @executor.create(httplib.FORBIDDEN, message.not_found_base)
def test_noProject(self):
req = self.req_d
req.project_name = 'notExistProject'
- (code, body) = self.create(req)
- self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn(message.not_found_base, body)
+ return req
+ @executor.create(httplib.FORBIDDEN, message.not_found_base)
def test_noTestcase(self):
req = self.req_d
req.case_name = 'notExistTestcase'
- (code, body) = self.create(req)
- self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn(message.not_found_base, body)
+ return req
+ @executor.create(httplib.OK, 'assert_href')
def test_success(self):
- (code, body) = self.create_d()
- self.assertEqual(code, httplib.OK)
- self.assert_href(body)
+ return self.req_d
+ @executor.create(httplib.OK, 'assert_href')
def test_key_with_doc(self):
req = copy.deepcopy(self.req_d)
req.details = {'1.name': 'dot_name'}
- (code, body) = self.create(req)
- self.assertEqual(code, httplib.OK)
- self.assert_href(body)
+ return req
+ @executor.create(httplib.OK, '_assert_no_ti')
def test_no_ti(self):
req = result_models.ResultCreateRequest(pod_name=self.pod,
project_name=self.project,
@@ -204,106 +196,110 @@ class TestResultCreate(TestResultBase):
build_tag=self.build_tag,
scenario=self.scenario,
criteria=self.criteria)
- (code, res) = self.create(req)
- _id = res.href.split('/')[-1]
- self.assertEqual(code, httplib.OK)
+ self.actual_req = req
+ return req
+
+ def _assert_no_ti(self, body):
+ _id = body.href.split('/')[-1]
code, body = self.get(_id)
- self.assert_res(code, body, req)
+ self.assert_res(body, self.actual_req)
class TestResultGet(TestResultBase):
+ def setUp(self):
+ super(TestResultGet, self).setUp()
+ self.req_d_id = self._create_d()
+ self.req_10d_later = self._create_changed_date(days=10)
+ self.req_10d_before = self._create_changed_date(days=-10)
+
+ @executor.get(httplib.OK, 'assert_res')
def test_getOne(self):
- _id = self._create_d()
- code, body = self.get(_id)
- self.assert_res(code, body)
+ return self.req_d_id
+ @executor.query(httplib.OK, '_query_success', 3)
def test_queryPod(self):
- self._query_and_assert(self._set_query('pod'))
+ return self._set_query('pod')
+ @executor.query(httplib.OK, '_query_success', 3)
def test_queryProject(self):
- self._query_and_assert(self._set_query('project'))
+ return self._set_query('project')
+ @executor.query(httplib.OK, '_query_success', 3)
def test_queryTestcase(self):
- self._query_and_assert(self._set_query('case'))
+ return self._set_query('case')
+ @executor.query(httplib.OK, '_query_success', 3)
def test_queryVersion(self):
- self._query_and_assert(self._set_query('version'))
+ return self._set_query('version')
+ @executor.query(httplib.OK, '_query_success', 3)
def test_queryInstaller(self):
- self._query_and_assert(self._set_query('installer'))
+ return self._set_query('installer')
+ @executor.query(httplib.OK, '_query_success', 3)
def test_queryBuildTag(self):
- self._query_and_assert(self._set_query('build_tag'))
+ return self._set_query('build_tag')
+ @executor.query(httplib.OK, '_query_success', 3)
def test_queryScenario(self):
- self._query_and_assert(self._set_query('scenario'))
+ return self._set_query('scenario')
+ @executor.query(httplib.OK, '_query_success', 3)
def test_queryTrustIndicator(self):
- self._query_and_assert(self._set_query('trust_indicator'))
+ return self._set_query('trust_indicator')
+ @executor.query(httplib.OK, '_query_success', 3)
def test_queryCriteria(self):
- self._query_and_assert(self._set_query('criteria'))
+ return self._set_query('criteria')
+ @executor.query(httplib.BAD_REQUEST, message.must_int('period'))
def test_queryPeriodNotInt(self):
- code, body = self.query(self._set_query('period=a'))
- self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn('period must be int', body)
-
- def test_queryPeriodFail(self):
- self._query_and_assert(self._set_query('period=1'),
- found=False, days=-10)
+ return self._set_query('period=a')
+ @executor.query(httplib.OK, '_query_last_one', 1)
def test_queryPeriodSuccess(self):
- self._query_and_assert(self._set_query('period=1'),
- found=True)
+ return self._set_query('period=1')
+ @executor.query(httplib.BAD_REQUEST, message.must_int('last'))
def test_queryLastNotInt(self):
- code, body = self.query(self._set_query('last=a'))
- self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn('last must be int', body)
+ return self._set_query('last=a')
+ @executor.query(httplib.OK, '_query_last_one', 1)
def test_queryLast(self):
- self._create_changed_date()
- req = self._create_changed_date(minutes=20)
- self._create_changed_date(minutes=-20)
- self._query_and_assert(self._set_query('last=1'), req=req)
+ return self._set_query('last=1')
+ @executor.query(httplib.OK, '_query_last_one', 1)
def test_combination(self):
- self._query_and_assert(self._set_query('pod',
- 'project',
- 'case',
- 'version',
- 'installer',
- 'build_tag',
- 'scenario',
- 'trust_indicator',
- 'criteria',
- 'period=1'))
-
+ return self._set_query('pod',
+ 'project',
+ 'case',
+ 'version',
+ 'installer',
+ 'build_tag',
+ 'scenario',
+ 'trust_indicator',
+ 'criteria',
+ 'period=1')
+
+ @executor.query(httplib.OK, '_query_success', 0)
def test_notFound(self):
- self._query_and_assert(self._set_query('pod=notExistPod',
- 'project',
- 'case',
- 'version',
- 'installer',
- 'build_tag',
- 'scenario',
- 'trust_indicator',
- 'criteria',
- 'period=1'),
- found=False)
-
- def _query_and_assert(self, query, found=True, req=None, **kwargs):
- if req is None:
- req = self._create_changed_date(**kwargs)
- code, body = self.query(query)
- if not found:
- self.assertEqual(code, httplib.OK)
- self.assertEqual(0, len(body.results))
- else:
- self.assertEqual(1, len(body.results))
- for result in body.results:
- self.assert_res(code, result, req)
+ return self._set_query('pod=notExistPod',
+ 'project',
+ 'case',
+ 'version',
+ 'installer',
+ 'build_tag',
+ 'scenario',
+ 'trust_indicator',
+ 'criteria',
+ 'period=1')
+
+ def _query_success(self, body, number):
+ self.assertEqual(number, len(body.results))
+
+ def _query_last_one(self, body, number):
+ self.assertEqual(number, len(body.results))
+ self.assert_res(body.results[0], self.req_10d_later)
def _create_changed_date(self, **kwargs):
req = copy.deepcopy(self.req_d)
@@ -327,9 +323,12 @@ class TestResultGet(TestResultBase):
class TestResultUpdate(TestResultBase):
- def test_success(self):
- _id = self._create_d()
+ def setUp(self):
+ super(TestResultUpdate, self).setUp()
+ self.req_d_id = self._create_d()
+ @executor.update(httplib.OK, '_assert_update_ti')
+ def test_success(self):
new_ti = copy.deepcopy(self.trust_indicator)
new_ti.current += self.update_step
new_ti.histories.append(
@@ -337,13 +336,16 @@ class TestResultUpdate(TestResultBase):
new_data = copy.deepcopy(self.req_d)
new_data.trust_indicator = new_ti
update = result_models.ResultUpdateRequest(trust_indicator=new_ti)
- code, body = self.update(update, _id)
- self.assertEqual(_id, body._id)
- self.assert_res(code, body, new_data)
+ self.update_req = new_data
+ return update, self.req_d_id
- code, new_body = self.get(_id)
- self.assertEqual(_id, new_body._id)
- self.assert_res(code, new_body, new_data)
+ def _assert_update_ti(self, request, body):
+ ti = body.trust_indicator
+ self.assertEqual(ti.current, request.trust_indicator.current)
+ if ti.histories:
+ history = ti.histories[0]
+ self.assertEqual(history.date, self.update_date)
+ self.assertEqual(history.step, self.update_step)
if __name__ == '__main__':
diff --git a/utils/test/testapi/opnfv_testapi/tests/unit/test_token.py b/utils/test/testapi/opnfv_testapi/tests/unit/test_token.py
index 6e3d4b3ea..ca247a3b7 100644
--- a/utils/test/testapi/opnfv_testapi/tests/unit/test_token.py
+++ b/utils/test/testapi/opnfv_testapi/tests/unit/test_token.py
@@ -11,6 +11,7 @@ from tornado import web
from opnfv_testapi.common import message
from opnfv_testapi.resources import project_models
from opnfv_testapi.router import url_mappings
+from opnfv_testapi.tests.unit import executor
from opnfv_testapi.tests.unit import fake_pymongo
from opnfv_testapi.tests.unit import test_base as base
@@ -32,22 +33,24 @@ class TestTokenCreateProject(TestToken):
fake_pymongo.tokens.insert({"access_token": "12345"})
self.basePath = '/api/v1/projects'
+ @executor.create(httplib.FORBIDDEN, message.invalid_token())
def test_projectCreateTokenInvalid(self):
self.headers['X-Auth-Token'] = '1234'
- code, body = self.create_d()
- self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn(message.invalid_token(), body)
+ return self.req_d
+ @executor.create(httplib.UNAUTHORIZED, message.unauthorized())
def test_projectCreateTokenUnauthorized(self):
- self.headers.pop('X-Auth-Token')
- code, body = self.create_d()
- self.assertEqual(code, httplib.UNAUTHORIZED)
- self.assertIn(message.unauthorized(), body)
+ if 'X-Auth-Token' in self.headers:
+ self.headers.pop('X-Auth-Token')
+ return self.req_d
+ @executor.create(httplib.OK, '_create_success')
def test_projectCreateTokenSuccess(self):
self.headers['X-Auth-Token'] = '12345'
- code, body = self.create_d()
- self.assertEqual(code, httplib.OK)
+ return self.req_d
+
+ def _create_success(self, body):
+ self.assertIn('CreateResponse', str(type(body)))
class TestTokenDeleteProject(TestToken):
@@ -56,28 +59,25 @@ class TestTokenDeleteProject(TestToken):
self.req_d = project_models.ProjectCreateRequest('vping')
fake_pymongo.tokens.insert({"access_token": "12345"})
self.basePath = '/api/v1/projects'
-
- def test_projectDeleteTokenIvalid(self):
self.headers['X-Auth-Token'] = '12345'
self.create_d()
+
+ @executor.delete(httplib.FORBIDDEN, message.invalid_token())
+ def test_projectDeleteTokenIvalid(self):
self.headers['X-Auth-Token'] = '1234'
- code, body = self.delete(self.req_d.name)
- self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn(message.invalid_token(), body)
+ return self.req_d.name
+ @executor.delete(httplib.UNAUTHORIZED, message.unauthorized())
def test_projectDeleteTokenUnauthorized(self):
- self.headers['X-Auth-Token'] = '12345'
- self.create_d()
self.headers.pop('X-Auth-Token')
- code, body = self.delete(self.req_d.name)
- self.assertEqual(code, httplib.UNAUTHORIZED)
- self.assertIn(message.unauthorized(), body)
+ return self.req_d.name
+ @executor.delete(httplib.OK, '_delete_success')
def test_projectDeleteTokenSuccess(self):
- self.headers['X-Auth-Token'] = '12345'
- self.create_d()
- code, body = self.delete(self.req_d.name)
- self.assertEqual(code, httplib.OK)
+ return self.req_d.name
+
+ def _delete_success(self, body):
+ self.assertEqual('', body)
class TestTokenUpdateProject(TestToken):
@@ -86,34 +86,28 @@ class TestTokenUpdateProject(TestToken):
self.req_d = project_models.ProjectCreateRequest('vping')
fake_pymongo.tokens.insert({"access_token": "12345"})
self.basePath = '/api/v1/projects'
-
- def test_projectUpdateTokenIvalid(self):
self.headers['X-Auth-Token'] = '12345'
self.create_d()
- code, body = self.get(self.req_d.name)
+
+ @executor.update(httplib.FORBIDDEN, message.invalid_token())
+ def test_projectUpdateTokenIvalid(self):
self.headers['X-Auth-Token'] = '1234'
req = project_models.ProjectUpdateRequest('newName', 'new description')
- code, body = self.update(req, self.req_d.name)
- self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn(message.invalid_token(), body)
+ return req, self.req_d.name
+ @executor.update(httplib.UNAUTHORIZED, message.unauthorized())
def test_projectUpdateTokenUnauthorized(self):
- self.headers['X-Auth-Token'] = '12345'
- self.create_d()
- code, body = self.get(self.req_d.name)
self.headers.pop('X-Auth-Token')
req = project_models.ProjectUpdateRequest('newName', 'new description')
- code, body = self.update(req, self.req_d.name)
- self.assertEqual(code, httplib.UNAUTHORIZED)
- self.assertIn(message.unauthorized(), body)
+ return req, self.req_d.name
+ @executor.update(httplib.OK, '_update_success')
def test_projectUpdateTokenSuccess(self):
- self.headers['X-Auth-Token'] = '12345'
- self.create_d()
- code, body = self.get(self.req_d.name)
req = project_models.ProjectUpdateRequest('newName', 'new description')
- code, body = self.update(req, self.req_d.name)
- self.assertEqual(code, httplib.OK)
+ return req, self.req_d.name
+
+ def _update_success(self, request, body):
+ self.assertIn(request.name, body)
if __name__ == '__main__':
unittest.main()
diff --git a/utils/test/testapi/opnfv_testapi/tests/unit/test_version.py b/utils/test/testapi/opnfv_testapi/tests/unit/test_version.py
index 5cd83fda5..fff802ac8 100644
--- a/utils/test/testapi/opnfv_testapi/tests/unit/test_version.py
+++ b/utils/test/testapi/opnfv_testapi/tests/unit/test_version.py
@@ -6,9 +6,11 @@
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+import httplib
import unittest
from opnfv_testapi.resources import models
+from opnfv_testapi.tests.unit import executor
from opnfv_testapi.tests.unit import test_base as base
@@ -20,12 +22,15 @@ class TestVersionBase(base.TestBase):
class TestVersion(TestVersionBase):
+ @executor.get(httplib.OK, '_get_success')
def test_success(self):
- code, body = self.get()
- self.assertEqual(200, code)
+ return None
+
+ def _get_success(self, body):
self.assertEqual(len(body.versions), 1)
self.assertEqual(body.versions[0].version, 'v1.0')
self.assertEqual(body.versions[0].description, 'basics')
+
if __name__ == '__main__':
unittest.main()