--- /dev/null
+##############################################################################
+# 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
+##############################################################################
+not_found_base = 'Could Not Found'
+exist_base = 'Already Exists'
+
+
+def no_body():
+ return 'No Body'
+
+
+def not_found(key, value):
+ return '{} {} [{}]'.format(not_found_base, key, value)
+
+
+def missing(name):
+ return '{} Missing'.format(name)
+
+
+def exist(key, value):
+ return '{} [{}] {}'.format(key, value, exist_base)
+
+
+def bad_format(error):
+ return 'Bad Format [{}]'.format(error)
+
+
+def unauthorized():
+ return 'No Authentication Header'
+
+
+def invalid_token():
+ return 'Invalid Token'
+
+
+def no_update():
+ return 'Nothing to update'
+
+
+def must_int(name):
+ return '{} must be int'.format(name)
+##############################################################################
+# 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 httplib
from tornado import web
from tornado import web
import models
+from opnfv_testapi.common import message
from opnfv_testapi.common import raises
from opnfv_testapi.tornado_swagger import swagger
try:
self.json_args = json.loads(self.request.body)
except (ValueError, KeyError, TypeError) as error:
- raises.BadRequest("Bad Json format [{}]".format(error))
+ raises.BadRequest(message.bad_format(str(error)))
def finish_request(self, json_object=None):
if json_object:
try:
token = self.request.headers['X-Auth-Token']
except KeyError:
- raises.Unauthorized("No Authentication Header.")
+ raises.Unauthorized(message.unauthorized())
query = {'access_token': token}
check = yield self._eval_db_find_one(query, 'tokens')
if not check:
- raises.Forbidden("Invalid Token.")
+ raises.Forbidden(message.invalid_token())
ret = yield gen.coroutine(method)(self, *args, **kwargs)
raise gen.Return(ret)
return wrapper
:param db_checks: [(table, exist, query, error)]
"""
if self.json_args is None:
- raises.BadRequest('no body')
+ 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('{} missing'.format(miss))
+ 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, message = error(data)
- raises.CodeTBD(code, message)
+ code, msg = error(data)
+ raises.CodeTBD(code, msg)
if self.table != 'results':
data.creation_date = datetime.now()
def _get_one(self, query):
data = yield self._eval_db_find_one(query)
if data is None:
- raises.NotFound("[{}] not exist in table [{}]"
- .format(query, self.table))
+ raises.NotFound(message.not_found(self.table, query))
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("[{}] not exit in table [{}]"
- .format(query, self.table))
+ raises.NotFound(message.not_found(self.table, query))
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("No payload")
+ 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("{} could not be found in table [{}]"
- .format(query, self.table))
+ raises.NotFound(message.not_found(self.table, query))
data = self.table_cls.from_dict(from_data)
# check new data exist
if not equal:
to_data = yield self._eval_db_find_one(new_query)
if to_data is not None:
- raises.Forbidden("{} already exists in table [{}]"
- .format(new_query, self.table))
+ raises.Forbidden(message.exist(self.table, new_query))
# we merge the whole document """
edit_request = self._update_requests(data)
request = self._update_request(request, k, v,
data.__getattribute__(k))
if not request:
- raises.Forbidden("Nothing to update")
+ raises.Forbidden(message.no_update())
edit_request = data.format()
edit_request.update(request)
import httplib
import handlers
+from opnfv_testapi.common import message
from opnfv_testapi.tornado_swagger import swagger
import pod_models
return {'name': data.name}
def error(data):
- message = '{} already exists as a pod'.format(data.name)
- return httplib.FORBIDDEN, message
+ return httplib.FORBIDDEN, message.exist('pod', data.name)
miss_checks = ['name']
db_checks = [(self.table, False, query, error)]
import httplib
import handlers
+from opnfv_testapi.common import message
from opnfv_testapi.tornado_swagger import swagger
import project_models
return {'name': data.name}
def error(data):
- message = '{} already exists as a project'.format(data.name)
- return httplib.FORBIDDEN, message
+ return httplib.FORBIDDEN, message.exist('project', data.name)
miss_checks = ['name']
db_checks = [(self.table, False, query, error)]
from bson import objectid
+from opnfv_testapi.common import message
from opnfv_testapi.common import raises
from opnfv_testapi.resources import handlers
from opnfv_testapi.resources import result_models
try:
value = int(value)
except:
- raises.BadRequest('{} must be int'.format(key))
+ raises.BadRequest(message.must_int(key))
return value
def set_query(self):
return {'name': data.pod_name}
def pod_error(data):
- message = 'Could not find pod [{}]'.format(data.pod_name)
- return httplib.NOT_FOUND, message
+ return httplib.NOT_FOUND, message.not_found('pod', data.pod_name)
def project_query(data):
return {'name': data.project_name}
def project_error(data):
- message = 'Could not find project [{}]'.format(data.project_name)
- return httplib.NOT_FOUND, message
+ return httplib.NOT_FOUND, 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):
- message = 'Could not find testcase [{}] in project [{}]'\
- .format(data.case_name, data.project_name)
- return httplib.NOT_FOUND, message
+ return httplib.NOT_FOUND, message.not_found('testcase',
+ data.case_name)
miss_checks = ['pod_name', 'project_name', 'case_name']
db_checks = [('pods', True, pod_query, pod_error),
import functools
import httplib
+from opnfv_testapi.common import message
from opnfv_testapi.common import raises
from opnfv_testapi.resources import handlers
import opnfv_testapi.resources.scenario_models as models
return {'name': data.name}
def error(data):
- message = '{} already exists as a scenario'.format(data.name)
- return httplib.FORBIDDEN, message
+ return httplib.FORBIDDEN, message.exist('scenario', data.name)
miss_checks = ['name']
db_checks = [(self.table, False, query, error)]
def _update_requests_rename(self, data):
data.name = self._term.get('name')
if not data.name:
- raises.BadRequest("new scenario name is not provided")
+ raises.BadRequest(message.missing('name'))
def _update_requests_add_installer(self, data):
data.installers.append(models.ScenarioInstaller.from_dict(self._term))
##############################################################################
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
}
def p_error(data):
- message = 'Could not find project [{}]'.format(data.project_name)
- return httplib.FORBIDDEN, message
+ return httplib.FORBIDDEN, message.not_found('project',
+ data.project_name)
def tc_error(data):
- message = '{} already exists as a testcase in project {}'\
- .format(data.name, data.project_name)
- return httplib.FORBIDDEN, message
+ return httplib.FORBIDDEN, message.exist('testcase', data.name)
miss_checks = ['name']
db_checks = [(self.db_projects, True, p_query, p_error),
import httplib
import unittest
+from opnfv_testapi.common import message
from opnfv_testapi.resources import pod_models
import test_base as base
req_empty = pod_models.PodCreateRequest('')
(code, body) = self.create(req_empty)
self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn('name missing', body)
+ self.assertIn(message.missing('name'), body)
def test_noneName(self):
req_none = pod_models.PodCreateRequest(None)
(code, body) = self.create(req_none)
self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn('name missing', body)
+ self.assertIn(message.missing('name'), body)
def test_success(self):
code, body = self.create_d()
self.create_d()
code, body = self.create_d()
self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn('already exists', body)
+ self.assertIn(message.exist_base, body)
class TestPodGet(TestPodBase):
import httplib
import unittest
+from opnfv_testapi.common import message
from opnfv_testapi.resources import project_models
import test_base as base
req_empty = project_models.ProjectCreateRequest('')
(code, body) = self.create(req_empty)
self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn('name missing', body)
+ self.assertIn(message.missing('name'), body)
def test_noneName(self):
req_none = project_models.ProjectCreateRequest(None)
(code, body) = self.create(req_none)
self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn('name missing', body)
+ self.assertIn(message.missing('name'), body)
def test_success(self):
(code, body) = self.create_d()
self.create_d()
(code, body) = self.create_d()
self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn('already exists', body)
+ self.assertIn(message.exist_base, body)
class TestProjectGet(TestProjectBase):
self.create_e()
code, body = self.update(self.req_e, self.req_d.name)
self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn("already exists", body)
+ self.assertIn(message.exist_base, body)
def test_noUpdate(self):
self.create_d()
code, body = self.update(self.req_d, self.req_d.name)
self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn("Nothing to update", body)
+ self.assertIn(message.no_update(), body)
def test_success(self):
self.create_d()
import httplib
import unittest
+from opnfv_testapi.common import message
from opnfv_testapi.resources import pod_models
from opnfv_testapi.resources import project_models
from opnfv_testapi.resources import result_models
def test_nobody(self):
(code, body) = self.create(None)
self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn('no body', body)
+ self.assertIn(message.no_body(), body)
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('pod_name missing', body)
+ self.assertIn(message.missing('pod_name'), body)
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('project_name missing', body)
+ self.assertIn(message.missing('project_name'), body)
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('case_name missing', body)
+ self.assertIn(message.missing('case_name'), body)
def test_noPod(self):
req = self.req_d
req.pod_name = 'notExistPod'
(code, body) = self.create(req)
self.assertEqual(code, httplib.NOT_FOUND)
- self.assertIn('Could not find pod', body)
+ self.assertIn(message.not_found_base, body)
def test_noProject(self):
req = self.req_d
req.project_name = 'notExistProject'
(code, body) = self.create(req)
self.assertEqual(code, httplib.NOT_FOUND)
- self.assertIn('Could not find project', body)
+ self.assertIn(message.not_found_base, body)
def test_noTestcase(self):
req = self.req_d
req.case_name = 'notExistTestcase'
(code, body) = self.create(req)
self.assertEqual(code, httplib.NOT_FOUND)
- self.assertIn('Could not find testcase', body)
+ self.assertIn(message.not_found_base, body)
def test_success(self):
(code, body) = self.create_d()
import json
import os
+from opnfv_testapi.common import message
import opnfv_testapi.resources.scenario_models as models
import test_base as base
req_empty = models.ScenarioCreateRequest('')
(code, body) = self.create(req_empty)
self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn('name missing', body)
+ self.assertIn(message.missing('name'), body)
def test_noneName(self):
req_none = models.ScenarioCreateRequest(None)
(code, body) = self.create(req_none)
self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn('name missing', body)
+ self.assertIn(message.missing('name'), body)
def test_success(self):
(code, body) = self.create_d()
self.create_d()
(code, body) = self.create_d()
self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn('already exists', body)
+ self.assertIn(message.exist_base, body)
class TestScenarioGet(TestScenarioBase):
import httplib
import unittest
+from opnfv_testapi.common import message
from opnfv_testapi.resources import project_models
from opnfv_testapi.resources import testcase_models
import test_base as base
def test_noProject(self):
code, body = self.create(self.req_d, 'noProject')
self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn('Could not find project', body)
+ self.assertIn(message.not_found_base, body)
def test_emptyName(self):
req_empty = testcase_models.TestcaseCreateRequest('')
(code, body) = self.create(req_empty, self.project)
self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn('name missing', body)
+ self.assertIn(message.missing('name'), body)
def test_noneName(self):
req_none = testcase_models.TestcaseCreateRequest(None)
(code, body) = self.create(req_none, self.project)
self.assertEqual(code, httplib.BAD_REQUEST)
- self.assertIn('name missing', body)
+ self.assertIn(message.missing('name'), body)
def test_success(self):
code, body = self.create_d()
self.create_d()
code, body = self.create_d()
self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn('already exists', body)
+ self.assertIn(message.exist_base, body)
class TestCaseGet(TestCaseBase):
self.create_e()
code, body = self.update(self.update_e, self.req_d.name)
self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn("already exists", body)
+ self.assertIn(message.exist_base, body)
def test_noUpdate(self):
self.create_d()
code, body = self.update(self.update_d, self.req_d.name)
self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn("Nothing to update", body)
+ self.assertIn(message.no_update(), body)
def test_success(self):
self.create_d()
from tornado import web
import fake_pymongo
+from opnfv_testapi.common import message
from opnfv_testapi.resources import project_models
from opnfv_testapi.router import url_mappings
import test_base as base
self.headers['X-Auth-Token'] = '1234'
code, body = self.create_d()
self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn('Invalid Token.', body)
+ self.assertIn(message.invalid_token(), body)
def test_projectCreateTokenUnauthorized(self):
self.headers.pop('X-Auth-Token')
code, body = self.create_d()
self.assertEqual(code, httplib.UNAUTHORIZED)
- self.assertIn('No Authentication Header.', body)
+ self.assertIn(message.unauthorized(), body)
def test_projectCreateTokenSuccess(self):
self.headers['X-Auth-Token'] = '12345'
self.headers['X-Auth-Token'] = '1234'
code, body = self.delete(self.req_d.name)
self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn('Invalid Token.', body)
+ self.assertIn(message.invalid_token(), body)
def test_projectDeleteTokenUnauthorized(self):
self.headers['X-Auth-Token'] = '12345'
self.headers.pop('X-Auth-Token')
code, body = self.delete(self.req_d.name)
self.assertEqual(code, httplib.UNAUTHORIZED)
- self.assertIn('No Authentication Header.', body)
+ self.assertIn(message.unauthorized(), body)
def test_projectDeleteTokenSuccess(self):
self.headers['X-Auth-Token'] = '12345'
req = project_models.ProjectUpdateRequest('newName', 'new description')
code, body = self.update(req, self.req_d.name)
self.assertEqual(code, httplib.FORBIDDEN)
- self.assertIn('Invalid Token.', body)
+ self.assertIn(message.invalid_token(), body)
def test_projectUpdateTokenUnauthorized(self):
self.headers['X-Auth-Token'] = '12345'
req = project_models.ProjectUpdateRequest('newName', 'new description')
code, body = self.update(req, self.req_d.name)
self.assertEqual(code, httplib.UNAUTHORIZED)
- self.assertIn('No Authentication Header.', body)
+ self.assertIn(message.unauthorized(), body)
def test_projectUpdateTokenSuccess(self):
self.headers['X-Auth-Token'] = '12345'