'''
Returns if the node is a controller
'''
- if 'controller' in self.roles:
- return True
- return False
+ return Role.CONTROLLER in self.roles
def is_compute(self):
'''
Returns if the node is a compute
'''
- if 'compute' in self.roles:
- return True
- return False
+ return Role.COMPUTE in self.roles
+
+ def is_odl(self):
+ '''
+ Returns if the node is an opendaylight
+ '''
+ return Role.ODL in self.roles
def get_ovs_info(self):
'''
if timestamp is None:
timestamp = time.strftime("%Y%m%d-%H%M%S")
- for controller_client in controller_clients:
- self.ofctl_dump_flows(controller_client,
- timestamp=timestamp)
- self.vsctl_show(controller_client,
- timestamp=timestamp)
-
- for compute_client in compute_clients:
- self.ofctl_dump_flows(compute_client,
- timestamp=timestamp)
- self.vsctl_show(compute_client,
- timestamp=timestamp)
+ clients = controller_clients + compute_clients
+ for client in clients:
+ self.ofctl_dump_flows(client, timestamp=timestamp)
+ self.vsctl_show(client, timestamp=timestamp)
if related_error is not None:
dumpdir = os.path.join(self.ovs_dir, timestamp)
+ self.__mkdir_p(dumpdir)
with open(os.path.join(dumpdir, 'error'), 'w') as f:
f.write(related_error)
import argparse
-import tornado.ioloop
import motor
+import tornado.ioloop
-from opnfv_testapi.common.config import APIConfig
-from opnfv_testapi.tornado_swagger import swagger
+from opnfv_testapi.common import config
from opnfv_testapi.router import url_mappings
+from opnfv_testapi.tornado_swagger import swagger
# optionally get config file from command line
parser = argparse.ArgumentParser()
parser.add_argument("-c", "--config-file", dest='config_file',
help="Config file location")
args = parser.parse_args()
-CONF = APIConfig().parse(args.config_file)
+CONF = config.APIConfig().parse(args.config_file)
# connecting to MongoDB server, and choosing database
client = motor.MotorClient(CONF.mongo_url)
# http://www.apache.org/licenses/LICENSE-2.0
# feng.xiaowei@zte.com.cn remove prepare_put_request 5-30-2016
##############################################################################
-
-
-from ConfigParser import SafeConfigParser, NoOptionError
+import ConfigParser
class ParseError(Exception):
def _get_parameter(self, section, param):
try:
return self._parser.get(section, param)
- except NoOptionError:
+ except ConfigParser.NoOptionError:
raise ParseError("[%s.%s] parameter not found" % (section, param))
def _get_int_parameter(self, section, param):
if config_location is None:
config_location = obj._default_config_location
- obj._parser = SafeConfigParser()
+ obj._parser = ConfigParser.SafeConfigParser()
obj._parser.read(config_location)
if not obj._parser:
raise ParseError("%s not found" % config_location)
# feng.xiaowei@zte.com.cn remove DashboardHandler 5-30-2016
##############################################################################
-import json
from datetime import datetime
+import json
from tornado import gen
-from tornado.web import RequestHandler, asynchronous, HTTPError
+from tornado import web
-from models import CreateResponse
-from opnfv_testapi.common.constants import DEFAULT_REPRESENTATION, \
- HTTP_BAD_REQUEST, HTTP_NOT_FOUND, HTTP_FORBIDDEN
+import models
+from opnfv_testapi.common import constants
from opnfv_testapi.tornado_swagger import swagger
-class GenericApiHandler(RequestHandler):
+class GenericApiHandler(web.RequestHandler):
def __init__(self, application, request, **kwargs):
super(GenericApiHandler, self).__init__(application, request, **kwargs)
self.db = self.settings["db"]
if self.request.method != "GET" and self.request.method != "DELETE":
if self.request.headers.get("Content-Type") is not None:
if self.request.headers["Content-Type"].startswith(
- DEFAULT_REPRESENTATION):
+ constants.DEFAULT_REPRESENTATION):
try:
self.json_args = json.loads(self.request.body)
except (ValueError, KeyError, TypeError) as error:
- raise HTTPError(HTTP_BAD_REQUEST,
- "Bad Json format [{}]".
- format(error))
+ raise web.HTTPError(constants.HTTP_BAD_REQUEST,
+ "Bad Json format [{}]".
+ format(error))
def finish_request(self, json_object=None):
if json_object:
self.write(json.dumps(json_object))
- self.set_header("Content-Type", DEFAULT_REPRESENTATION)
+ self.set_header("Content-Type", constants.DEFAULT_REPRESENTATION)
self.finish()
def _create_response(self, resource):
href = self.request.full_url() + '/' + str(resource)
- return CreateResponse(href=href).format()
+ return models.CreateResponse(href=href).format()
def format_data(self, data):
cls_data = self.table_cls.from_dict(data)
return cls_data.format_http()
- @asynchronous
+ @web.asynchronous
@gen.coroutine
def _create(self, miss_checks, db_checks, **kwargs):
"""
:param db_checks: [(table, exist, query, error)]
"""
if self.json_args is None:
- raise HTTPError(HTTP_BAD_REQUEST, "no body")
+ raise web.HTTPError(constants.HTTP_BAD_REQUEST, "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 == '':
- raise HTTPError(HTTP_BAD_REQUEST,
- '{} missing'.format(miss))
+ raise web.HTTPError(constants.HTTP_BAD_REQUEST,
+ '{} missing'.format(miss))
for k, v in kwargs.iteritems():
data.__setattr__(k, v)
check = yield self._eval_db_find_one(query(data), table)
if (exist and not check) or (not exist and check):
code, message = error(data)
- raise HTTPError(code, message)
+ raise web.HTTPError(code, message)
if self.table != 'results':
data.creation_date = datetime.now()
resource = _id
self.finish_request(self._create_response(resource))
- @asynchronous
+ @web.asynchronous
@gen.coroutine
def _list(self, query=None, res_op=None, *args, **kwargs):
if query is None:
res = res_op(data, *args)
self.finish_request(res)
- @asynchronous
+ @web.asynchronous
@gen.coroutine
def _get_one(self, query):
data = yield self._eval_db_find_one(query)
if data is None:
- raise HTTPError(HTTP_NOT_FOUND,
- "[{}] not exist in table [{}]"
- .format(query, self.table))
+ raise web.HTTPError(constants.HTTP_NOT_FOUND,
+ "[{}] not exist in table [{}]"
+ .format(query, self.table))
self.finish_request(self.format_data(data))
- @asynchronous
+ @web.asynchronous
@gen.coroutine
def _delete(self, query):
data = yield self._eval_db_find_one(query)
if data is None:
- raise HTTPError(HTTP_NOT_FOUND,
- "[{}] not exit in table [{}]"
- .format(query, self.table))
+ raise web.HTTPError(constants.HTTP_NOT_FOUND,
+ "[{}] not exit in table [{}]"
+ .format(query, self.table))
yield self._eval_db(self.table, 'remove', query)
self.finish_request()
- @asynchronous
+ @web.asynchronous
@gen.coroutine
def _update(self, query, db_keys):
if self.json_args is None:
- raise HTTPError(HTTP_BAD_REQUEST, "No payload")
+ raise web.HTTPError(constants.HTTP_BAD_REQUEST, "No payload")
# check old data exist
from_data = yield self._eval_db_find_one(query)
if from_data is None:
- raise HTTPError(HTTP_NOT_FOUND,
- "{} could not be found in table [{}]"
- .format(query, self.table))
+ raise web.HTTPError(constants.HTTP_NOT_FOUND,
+ "{} could not be found in table [{}]"
+ .format(query, self.table))
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:
- raise HTTPError(HTTP_FORBIDDEN,
- "{} already exists in table [{}]"
- .format(new_query, self.table))
+ raise web.HTTPError(constants.HTTP_FORBIDDEN,
+ "{} already exists in table [{}]"
+ .format(new_query, self.table))
# we merge the whole document """
edit_request = self._update_requests(data)
request = self._update_request(request, k, v,
data.__getattribute__(k))
if not request:
- raise HTTPError(HTTP_FORBIDDEN, "Nothing to update")
+ raise web.HTTPError(constants.HTTP_FORBIDDEN, "Nothing to update")
edit_request = data.format()
edit_request.update(request)
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+import handlers
+from opnfv_testapi.common import constants
from opnfv_testapi.tornado_swagger import swagger
-from handlers import GenericApiHandler
-from pod_models import Pod
-from opnfv_testapi.common.constants import HTTP_FORBIDDEN
+import pod_models
-class GenericPodHandler(GenericApiHandler):
+class GenericPodHandler(handlers.GenericApiHandler):
def __init__(self, application, request, **kwargs):
super(GenericPodHandler, self).__init__(application, request, **kwargs)
self.table = 'pods'
- self.table_cls = Pod
+ self.table_cls = pod_models.Pod
class PodCLHandler(GenericPodHandler):
def error(data):
message = '{} already exists as a pod'.format(data.name)
- return HTTP_FORBIDDEN, message
+ return constants.HTTP_FORBIDDEN, message
miss_checks = ['name']
db_checks = [(self.table, False, query, error)]
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
+import handlers
+from opnfv_testapi.common import constants
from opnfv_testapi.tornado_swagger import swagger
-from handlers import GenericApiHandler
-from opnfv_testapi.common.constants import HTTP_FORBIDDEN
-from project_models import Project
+import project_models
-class GenericProjectHandler(GenericApiHandler):
+class GenericProjectHandler(handlers.GenericApiHandler):
def __init__(self, application, request, **kwargs):
super(GenericProjectHandler, self).__init__(application,
request,
**kwargs)
self.table = 'projects'
- self.table_cls = Project
+ self.table_cls = project_models.Project
class ProjectCLHandler(GenericProjectHandler):
def error(data):
message = '{} already exists as a project'.format(data.name)
- return HTTP_FORBIDDEN, message
+ return constants.HTTP_FORBIDDEN, message
miss_checks = ['name']
db_checks = [(self.table, False, query, error)]
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-from datetime import datetime, timedelta
+from datetime import datetime
+from datetime import timedelta
-from bson.objectid import ObjectId
-from tornado.web import HTTPError
+from bson import objectid
+from tornado import web
-from opnfv_testapi.common.constants import HTTP_BAD_REQUEST, HTTP_NOT_FOUND
-from opnfv_testapi.resources.handlers import GenericApiHandler
-from opnfv_testapi.resources.result_models import TestResult
+from opnfv_testapi.common import constants
+from opnfv_testapi.resources import handlers
+from opnfv_testapi.resources import result_models
from opnfv_testapi.tornado_swagger import swagger
-class GenericResultHandler(GenericApiHandler):
+class GenericResultHandler(handlers.GenericApiHandler):
def __init__(self, application, request, **kwargs):
super(GenericResultHandler, self).__init__(application,
request,
**kwargs)
self.table = self.db_results
- self.table_cls = TestResult
+ self.table_cls = result_models.TestResult
def get_int(self, key, value):
try:
value = int(value)
except:
- raise HTTPError(HTTP_BAD_REQUEST, '{} must be int'.format(key))
+ raise web.HTTPError(constants.HTTP_BAD_REQUEST,
+ '{} must be int'.format(key))
return value
def set_query(self):
def pod_error(data):
message = 'Could not find pod [{}]'.format(data.pod_name)
- return HTTP_NOT_FOUND, message
+ return constants.HTTP_NOT_FOUND, message
def project_query(data):
return {'name': data.project_name}
def project_error(data):
message = 'Could not find project [{}]'.format(data.project_name)
- return HTTP_NOT_FOUND, message
+ return constants.HTTP_NOT_FOUND, message
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 HTTP_NOT_FOUND, message
+ return constants.HTTP_NOT_FOUND, message
miss_checks = ['pod_name', 'project_name', 'case_name']
db_checks = [('pods', True, pod_query, pod_error),
@raise 404: test result not exist
"""
query = dict()
- query["_id"] = ObjectId(result_id)
+ query["_id"] = objectid.ObjectId(result_id)
self._get_one(query)
@swagger.operation(nickname="updateTestResultById")
@raise 404: result not exist
@raise 403: nothing to update
"""
- query = {'_id': ObjectId(result_id)}
+ query = {'_id': objectid.ObjectId(result_id)}
db_keys = []
self._update(query, db_keys)
-from opnfv_testapi.common.constants import HTTP_FORBIDDEN
-from opnfv_testapi.resources.handlers import GenericApiHandler
-from opnfv_testapi.resources.scenario_models import Scenario
+from opnfv_testapi.common import constants
+from opnfv_testapi.resources import handlers
import opnfv_testapi.resources.scenario_models as models
from opnfv_testapi.tornado_swagger import swagger
-class GenericScenarioHandler(GenericApiHandler):
+class GenericScenarioHandler(handlers.GenericApiHandler):
def __init__(self, application, request, **kwargs):
super(GenericScenarioHandler, self).__init__(application,
request,
**kwargs)
self.table = self.db_scenarios
- self.table_cls = Scenario
+ self.table_cls = models.Scenario
class ScenariosCLHandler(GenericScenarioHandler):
def error(data):
message = '{} already exists as a scenario'.format(data.name)
- return HTTP_FORBIDDEN, message
+ return constants.HTTP_FORBIDDEN, message
miss_checks = ['name']
db_checks = [(self.table, False, query, error)]
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-from opnfv_testapi.common.constants import HTTP_FORBIDDEN
-from opnfv_testapi.resources.handlers import GenericApiHandler
-from opnfv_testapi.resources.testcase_models import Testcase
+from opnfv_testapi.common import constants
+from opnfv_testapi.resources import handlers
+from opnfv_testapi.resources import testcase_models
from opnfv_testapi.tornado_swagger import swagger
-class GenericTestcaseHandler(GenericApiHandler):
+class GenericTestcaseHandler(handlers.GenericApiHandler):
def __init__(self, application, request, **kwargs):
super(GenericTestcaseHandler, self).__init__(application,
request,
**kwargs)
self.table = self.db_testcases
- self.table_cls = Testcase
+ self.table_cls = testcase_models.Testcase
class TestcaseCLHandler(GenericTestcaseHandler):
def p_error(data):
message = 'Could not find project [{}]'.format(data.project_name)
- return HTTP_FORBIDDEN, message
+ return constants.HTTP_FORBIDDEN, message
def tc_error(data):
message = '{} already exists as a testcase in project {}'\
.format(data.name, data.project_name)
- return HTTP_FORBIDDEN, message
+ return constants.HTTP_FORBIDDEN, message
miss_checks = ['name']
db_checks = [(self.db_projects, True, p_query, p_error),
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-from opnfv_testapi.resources.handlers import VersionHandler
-from opnfv_testapi.resources.testcase_handlers import TestcaseCLHandler, \
- TestcaseGURHandler
-from opnfv_testapi.resources.pod_handlers import PodCLHandler, PodGURHandler
-from opnfv_testapi.resources.project_handlers import ProjectCLHandler, \
- ProjectGURHandler
-from opnfv_testapi.resources.result_handlers import ResultsCLHandler, \
- ResultsGURHandler
-from opnfv_testapi.resources.scenario_handlers import ScenariosCLHandler
-from opnfv_testapi.resources.scenario_handlers import ScenarioGURHandler
+from opnfv_testapi.resources import handlers
+from opnfv_testapi.resources import pod_handlers
+from opnfv_testapi.resources import project_handlers
+from opnfv_testapi.resources import result_handlers
+from opnfv_testapi.resources import scenario_handlers
+from opnfv_testapi.resources import testcase_handlers
mappings = [
# GET /versions => GET API version
- (r"/versions", VersionHandler),
+ (r"/versions", handlers.VersionHandler),
# few examples:
# GET /api/v1/pods => Get all pods
# GET /api/v1/pods/1 => Get details on POD 1
- (r"/api/v1/pods", PodCLHandler),
- (r"/api/v1/pods/([^/]+)", PodGURHandler),
+ (r"/api/v1/pods", pod_handlers.PodCLHandler),
+ (r"/api/v1/pods/([^/]+)", pod_handlers.PodGURHandler),
# few examples:
# GET /projects
# GET /projects/yardstick
- (r"/api/v1/projects", ProjectCLHandler),
- (r"/api/v1/projects/([^/]+)", ProjectGURHandler),
+ (r"/api/v1/projects", project_handlers.ProjectCLHandler),
+ (r"/api/v1/projects/([^/]+)", project_handlers.ProjectGURHandler),
# few examples
# GET /projects/qtip/cases => Get cases for qtip
- (r"/api/v1/projects/([^/]+)/cases", TestcaseCLHandler),
- (r"/api/v1/projects/([^/]+)/cases/([^/]+)", TestcaseGURHandler),
+ (r"/api/v1/projects/([^/]+)/cases", testcase_handlers.TestcaseCLHandler),
+ (r"/api/v1/projects/([^/]+)/cases/([^/]+)",
+ testcase_handlers.TestcaseGURHandler),
# new path to avoid a long depth
# GET /results?project=functest&case=keystone.catalog&pod=1
# POST /results =>
# Push results with mandatory request payload parameters
# (project, case, and pod)
- (r"/api/v1/results", ResultsCLHandler),
- (r"/api/v1/results/([^/]+)", ResultsGURHandler),
+ (r"/api/v1/results", result_handlers.ResultsCLHandler),
+ (r"/api/v1/results/([^/]+)", result_handlers.ResultsGURHandler),
# scenarios
- (r"/api/v1/scenarios", ScenariosCLHandler),
- (r"/api/v1/scenarios/([^/]+)", ScenarioGURHandler),
+ (r"/api/v1/scenarios", scenario_handlers.ScenariosCLHandler),
+ (r"/api/v1/scenarios/([^/]+)", scenario_handlers.ScenarioGURHandler),
]
##############################################################################
import json
-from tornado.web import Application
-from tornado.testing import AsyncHTTPTestCase
+from tornado import testing
+from tornado import web
-from opnfv_testapi.router import url_mappings
-from opnfv_testapi.resources.models import CreateResponse
import fake_pymongo
+from opnfv_testapi.resources import models
+from opnfv_testapi.router import url_mappings
-class TestBase(AsyncHTTPTestCase):
+class TestBase(testing.AsyncHTTPTestCase):
headers = {'Content-Type': 'application/json; charset=UTF-8'}
def setUp(self):
self.basePath = ''
- self.create_res = CreateResponse
+ self.create_res = models.CreateResponse
self.get_res = None
self.list_res = None
self.update_res = None
super(TestBase, self).setUp()
def get_app(self):
- return Application(
+ return web.Application(
url_mappings.mappings,
db=fake_pymongo,
debug=True,
import unittest
from tornado import gen
-from tornado.testing import AsyncHTTPTestCase, gen_test
-from tornado.web import Application
+from tornado import testing
+from tornado import web
import fake_pymongo
-class MyTest(AsyncHTTPTestCase):
+class MyTest(testing.AsyncHTTPTestCase):
def setUp(self):
super(MyTest, self).setUp()
self.db = fake_pymongo
self.io_loop.run_sync(self.fixture_setup)
def get_app(self):
- return Application()
+ return web.Application()
@gen.coroutine
def fixture_setup(self):
yield self.db.pods.insert({'_id': '1', 'name': 'test1'})
yield self.db.pods.insert({'name': 'test2'})
- @gen_test
+ @testing.gen_test
def test_find_one(self):
user = yield self.db.pods.find_one({'name': 'test1'})
self.assertEqual(user, self.test1)
self.db.pods.remove()
- @gen_test
+ @testing.gen_test
def test_find(self):
cursor = self.db.pods.find()
names = []
names.append(ob.get('name'))
self.assertItemsEqual(names, ['test1', 'test2'])
- @gen_test
+ @testing.gen_test
def test_update(self):
yield self.db.pods.update({'_id': '1'}, {'name': 'new_test1'})
user = yield self.db.pods.find_one({'_id': '1'})
None,
check_keys=False)
- @gen_test
+ @testing.gen_test
def test_remove(self):
yield self.db.pods.remove({'_id': '1'})
user = yield self.db.pods.find_one({'_id': '1'})
def _insert_assert(self, docs, error=None, **kwargs):
self._db_assert('insert', error, docs, **kwargs)
- @gen_test
+ @testing.gen_test
def _db_assert(self, method, error, *args, **kwargs):
name_error = None
try:
##############################################################################
import unittest
-from test_base import TestBase
-from opnfv_testapi.resources.pod_models import PodCreateRequest, Pod, Pods
-from opnfv_testapi.common.constants import HTTP_OK, HTTP_BAD_REQUEST, \
- HTTP_FORBIDDEN, HTTP_NOT_FOUND
+from opnfv_testapi.common import constants
+from opnfv_testapi.resources import pod_models
+import test_base as base
-class TestPodBase(TestBase):
+class TestPodBase(base.TestBase):
def setUp(self):
super(TestPodBase, self).setUp()
- self.req_d = PodCreateRequest('zte-1', 'virtual',
- 'zte pod 1', 'ci-pod')
- self.req_e = PodCreateRequest('zte-2', 'metal', 'zte pod 2')
- self.get_res = Pod
- self.list_res = Pods
+ self.req_d = pod_models.PodCreateRequest('zte-1', 'virtual',
+ 'zte pod 1', 'ci-pod')
+ self.req_e = pod_models.PodCreateRequest('zte-2', 'metal', 'zte pod 2')
+ self.get_res = pod_models.Pod
+ self.list_res = pod_models.Pods
self.basePath = '/api/v1/pods'
def assert_get_body(self, pod, req=None):
class TestPodCreate(TestPodBase):
def test_withoutBody(self):
(code, body) = self.create()
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
def test_emptyName(self):
- req_empty = PodCreateRequest('')
+ req_empty = pod_models.PodCreateRequest('')
(code, body) = self.create(req_empty)
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
self.assertIn('name missing', body)
def test_noneName(self):
- req_none = PodCreateRequest(None)
+ req_none = pod_models.PodCreateRequest(None)
(code, body) = self.create(req_none)
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
self.assertIn('name missing', body)
def test_success(self):
code, body = self.create_d()
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
self.assert_create_body(body)
def test_alreadyExist(self):
self.create_d()
code, body = self.create_d()
- self.assertEqual(code, HTTP_FORBIDDEN)
+ self.assertEqual(code, constants.HTTP_FORBIDDEN)
self.assertIn('already exists', body)
class TestPodGet(TestPodBase):
def test_notExist(self):
code, body = self.get('notExist')
- self.assertEqual(code, HTTP_NOT_FOUND)
+ self.assertEqual(code, constants.HTTP_NOT_FOUND)
def test_getOne(self):
self.create_d()
##############################################################################
import unittest
-from test_base import TestBase
-from opnfv_testapi.resources.project_models import ProjectCreateRequest, \
- Project, Projects, ProjectUpdateRequest
-from opnfv_testapi.common.constants import HTTP_OK, HTTP_BAD_REQUEST, \
- HTTP_FORBIDDEN, HTTP_NOT_FOUND
+from opnfv_testapi.common import constants
+from opnfv_testapi.resources import project_models
+import test_base as base
-class TestProjectBase(TestBase):
+class TestProjectBase(base.TestBase):
def setUp(self):
super(TestProjectBase, self).setUp()
- self.req_d = ProjectCreateRequest('vping', 'vping-ssh test')
- self.req_e = ProjectCreateRequest('doctor', 'doctor test')
- self.get_res = Project
- self.list_res = Projects
- self.update_res = Project
+ self.req_d = project_models.ProjectCreateRequest('vping',
+ 'vping-ssh test')
+ self.req_e = project_models.ProjectCreateRequest('doctor',
+ 'doctor test')
+ self.get_res = project_models.Project
+ self.list_res = project_models.Projects
+ self.update_res = project_models.Project
self.basePath = '/api/v1/projects'
def assert_body(self, project, req=None):
class TestProjectCreate(TestProjectBase):
def test_withoutBody(self):
(code, body) = self.create()
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
def test_emptyName(self):
- req_empty = ProjectCreateRequest('')
+ req_empty = project_models.ProjectCreateRequest('')
(code, body) = self.create(req_empty)
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
self.assertIn('name missing', body)
def test_noneName(self):
- req_none = ProjectCreateRequest(None)
+ req_none = project_models.ProjectCreateRequest(None)
(code, body) = self.create(req_none)
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
self.assertIn('name missing', body)
def test_success(self):
(code, body) = self.create_d()
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
self.assert_create_body(body)
def test_alreadyExist(self):
self.create_d()
(code, body) = self.create_d()
- self.assertEqual(code, HTTP_FORBIDDEN)
+ self.assertEqual(code, constants.HTTP_FORBIDDEN)
self.assertIn('already exists', body)
class TestProjectGet(TestProjectBase):
def test_notExist(self):
code, body = self.get('notExist')
- self.assertEqual(code, HTTP_NOT_FOUND)
+ self.assertEqual(code, constants.HTTP_NOT_FOUND)
def test_getOne(self):
self.create_d()
code, body = self.get(self.req_d.name)
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
self.assert_body(body)
def test_list(self):
class TestProjectUpdate(TestProjectBase):
def test_withoutBody(self):
code, _ = self.update(None, 'noBody')
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
def test_notFound(self):
code, _ = self.update(self.req_e, 'notFound')
- self.assertEqual(code, HTTP_NOT_FOUND)
+ self.assertEqual(code, constants.HTTP_NOT_FOUND)
def test_newNameExist(self):
self.create_d()
self.create_e()
code, body = self.update(self.req_e, self.req_d.name)
- self.assertEqual(code, HTTP_FORBIDDEN)
+ self.assertEqual(code, constants.HTTP_FORBIDDEN)
self.assertIn("already exists", body)
def test_noUpdate(self):
self.create_d()
code, body = self.update(self.req_d, self.req_d.name)
- self.assertEqual(code, HTTP_FORBIDDEN)
+ self.assertEqual(code, constants.HTTP_FORBIDDEN)
self.assertIn("Nothing to update", body)
def test_success(self):
code, body = self.get(self.req_d.name)
_id = body._id
- req = ProjectUpdateRequest('newName', 'new description')
+ req = project_models.ProjectUpdateRequest('newName', 'new description')
code, body = self.update(req, self.req_d.name)
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
self.assertEqual(_id, body._id)
self.assert_body(body, req)
class TestProjectDelete(TestProjectBase):
def test_notFound(self):
code, body = self.delete('notFound')
- self.assertEqual(code, HTTP_NOT_FOUND)
+ self.assertEqual(code, constants.HTTP_NOT_FOUND)
def test_success(self):
self.create_d()
code, body = self.delete(self.req_d.name)
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
self.assertEqual(body, '')
code, body = self.get(self.req_d.name)
- self.assertEqual(code, HTTP_NOT_FOUND)
+ self.assertEqual(code, constants.HTTP_NOT_FOUND)
if __name__ == '__main__':
unittest.main()
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
import copy
-import unittest
from datetime import datetime, timedelta
+import unittest
-from opnfv_testapi.common.constants import HTTP_OK, HTTP_BAD_REQUEST, \
- HTTP_NOT_FOUND
-from opnfv_testapi.resources.pod_models import PodCreateRequest
-from opnfv_testapi.resources.project_models import ProjectCreateRequest
-from opnfv_testapi.resources.result_models import ResultCreateRequest, \
- TestResult, TestResults, ResultUpdateRequest, TI, TIHistory
-from opnfv_testapi.resources.testcase_models import TestcaseCreateRequest
-from test_base import TestBase
+from opnfv_testapi.common import constants
+from opnfv_testapi.resources import pod_models
+from opnfv_testapi.resources import project_models
+from opnfv_testapi.resources import result_models
+from opnfv_testapi.resources import testcase_models
+import test_base as base
class Details(object):
return t
-class TestResultBase(TestBase):
+class TestResultBase(base.TestBase):
def setUp(self):
self.pod = 'zte-pod1'
self.project = 'functest'
self.build_tag = 'v3.0'
self.scenario = 'odl-l2'
self.criteria = 'passed'
- self.trust_indicator = TI(0.7)
+ self.trust_indicator = result_models.TI(0.7)
self.start_date = "2016-05-23 07:16:09.477097"
self.stop_date = "2016-05-23 07:16:19.477097"
self.update_date = "2016-05-24 07:16:19.477097"
self.update_step = -0.05
super(TestResultBase, self).setUp()
self.details = Details(timestart='0', duration='9s', status='OK')
- self.req_d = ResultCreateRequest(pod_name=self.pod,
- project_name=self.project,
- case_name=self.case,
- installer=self.installer,
- version=self.version,
- start_date=self.start_date,
- stop_date=self.stop_date,
- details=self.details.format(),
- build_tag=self.build_tag,
- scenario=self.scenario,
- criteria=self.criteria,
- trust_indicator=self.trust_indicator)
- self.get_res = TestResult
- self.list_res = TestResults
- self.update_res = TestResult
+ self.req_d = result_models.ResultCreateRequest(
+ pod_name=self.pod,
+ project_name=self.project,
+ case_name=self.case,
+ installer=self.installer,
+ version=self.version,
+ start_date=self.start_date,
+ stop_date=self.stop_date,
+ details=self.details.format(),
+ build_tag=self.build_tag,
+ scenario=self.scenario,
+ criteria=self.criteria,
+ trust_indicator=self.trust_indicator)
+ self.get_res = result_models.TestResult
+ self.list_res = result_models.TestResults
+ self.update_res = result_models.TestResult
self.basePath = '/api/v1/results'
- self.req_pod = PodCreateRequest(self.pod, 'metal', 'zte pod 1')
- self.req_project = ProjectCreateRequest(self.project, 'vping test')
- self.req_testcase = TestcaseCreateRequest(self.case,
- '/cases/vping',
- 'vping-ssh test')
+ self.req_pod = pod_models.PodCreateRequest(
+ self.pod,
+ 'metal',
+ 'zte pod 1')
+ self.req_project = project_models.ProjectCreateRequest(
+ self.project,
+ 'vping test')
+ self.req_testcase = testcase_models.TestcaseCreateRequest(
+ self.case,
+ '/cases/vping',
+ 'vping-ssh test')
self.create_help('/api/v1/pods', self.req_pod)
self.create_help('/api/v1/projects', self.req_project)
self.create_help('/api/v1/projects/%s/cases',
self.project)
def assert_res(self, code, result, req=None):
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
if req is None:
req = self.req_d
self.assertEqual(result.pod_name, req.pod_name)
class TestResultCreate(TestResultBase):
def test_nobody(self):
(code, body) = self.create(None)
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
self.assertIn('no body', body)
def test_podNotProvided(self):
req = self.req_d
req.pod_name = None
(code, body) = self.create(req)
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
self.assertIn('pod_name missing', body)
def test_projectNotProvided(self):
req = self.req_d
req.project_name = None
(code, body) = self.create(req)
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
self.assertIn('project_name missing', body)
def test_testcaseNotProvided(self):
req = self.req_d
req.case_name = None
(code, body) = self.create(req)
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
self.assertIn('case_name missing', body)
def test_noPod(self):
req = self.req_d
req.pod_name = 'notExistPod'
(code, body) = self.create(req)
- self.assertEqual(code, HTTP_NOT_FOUND)
+ self.assertEqual(code, constants.HTTP_NOT_FOUND)
self.assertIn('Could not find pod', body)
def test_noProject(self):
req = self.req_d
req.project_name = 'notExistProject'
(code, body) = self.create(req)
- self.assertEqual(code, HTTP_NOT_FOUND)
+ self.assertEqual(code, constants.HTTP_NOT_FOUND)
self.assertIn('Could not find project', body)
def test_noTestcase(self):
req = self.req_d
req.case_name = 'notExistTestcase'
(code, body) = self.create(req)
- self.assertEqual(code, HTTP_NOT_FOUND)
+ self.assertEqual(code, constants.HTTP_NOT_FOUND)
self.assertIn('Could not find testcase', body)
def test_success(self):
(code, body) = self.create_d()
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
self.assert_href(body)
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, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
self.assert_href(body)
def test_no_ti(self):
- req = ResultCreateRequest(pod_name=self.pod,
- project_name=self.project,
- case_name=self.case,
- installer=self.installer,
- version=self.version,
- start_date=self.start_date,
- stop_date=self.stop_date,
- details=self.details.format(),
- build_tag=self.build_tag,
- scenario=self.scenario,
- criteria=self.criteria)
+ req = result_models.ResultCreateRequest(pod_name=self.pod,
+ project_name=self.project,
+ case_name=self.case,
+ installer=self.installer,
+ version=self.version,
+ start_date=self.start_date,
+ stop_date=self.stop_date,
+ details=self.details.format(),
+ build_tag=self.build_tag,
+ scenario=self.scenario,
+ criteria=self.criteria)
(code, res) = self.create(req)
_id = res.href.split('/')[-1]
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
code, body = self.get(_id)
self.assert_res(code, body, req)
def test_queryPeriodNotInt(self):
code, body = self.query(self._set_query('period=a'))
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
self.assertIn('period must be int', body)
def test_queryPeriodFail(self):
def test_queryLastNotInt(self):
code, body = self.query(self._set_query('last=a'))
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
self.assertIn('last must be int', body)
def test_queryLast(self):
req = self._create_changed_date(**kwargs)
code, body = self.query(query)
if not found:
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
self.assertEqual(0, len(body.results))
else:
self.assertEqual(1, len(body.results))
new_ti = copy.deepcopy(self.trust_indicator)
new_ti.current += self.update_step
- new_ti.histories.append(TIHistory(self.update_date, self.update_step))
+ new_ti.histories.append(
+ result_models.TIHistory(self.update_date, self.update_step))
new_data = copy.deepcopy(self.req_d)
new_data.trust_indicator = new_ti
- update = ResultUpdateRequest(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)
from opnfv_testapi.common import constants
import opnfv_testapi.resources.scenario_models as models
-from test_testcase import TestBase
+import test_base as base
-class TestScenarioBase(TestBase):
+class TestScenarioBase(base.TestBase):
def setUp(self):
super(TestScenarioBase, self).setUp()
self.get_res = models.Scenario
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
-import unittest
import copy
+import unittest
-from test_base import TestBase
-from opnfv_testapi.resources.testcase_models import TestcaseCreateRequest, \
- Testcase, Testcases, TestcaseUpdateRequest
-from opnfv_testapi.resources.project_models import ProjectCreateRequest
-from opnfv_testapi.common.constants import HTTP_OK, HTTP_BAD_REQUEST, \
- HTTP_FORBIDDEN, HTTP_NOT_FOUND
+from opnfv_testapi.common import constants
+from opnfv_testapi.resources import project_models
+from opnfv_testapi.resources import testcase_models
+import test_base as base
-class TestCaseBase(TestBase):
+class TestCaseBase(base.TestBase):
def setUp(self):
super(TestCaseBase, self).setUp()
- self.req_d = TestcaseCreateRequest('vping_1',
- '/cases/vping_1',
- 'vping-ssh test')
- self.req_e = TestcaseCreateRequest('doctor_1',
- '/cases/doctor_1',
- 'create doctor')
- self.update_d = TestcaseUpdateRequest('vping_1',
- 'vping-ssh test',
- 'functest')
- self.update_e = TestcaseUpdateRequest('doctor_1',
- 'create doctor',
- 'functest')
- self.get_res = Testcase
- self.list_res = Testcases
- self.update_res = Testcase
+ self.req_d = testcase_models.TestcaseCreateRequest('vping_1',
+ '/cases/vping_1',
+ 'vping-ssh test')
+ self.req_e = testcase_models.TestcaseCreateRequest('doctor_1',
+ '/cases/doctor_1',
+ 'create doctor')
+ self.update_d = testcase_models.TestcaseUpdateRequest('vping_1',
+ 'vping-ssh test',
+ 'functest')
+ self.update_e = testcase_models.TestcaseUpdateRequest('doctor_1',
+ 'create doctor',
+ 'functest')
+ self.get_res = testcase_models.Testcase
+ self.list_res = testcase_models.Testcases
+ self.update_res = testcase_models.Testcase
self.basePath = '/api/v1/projects/%s/cases'
self.create_project()
self.assertIsNotNone(new.creation_date)
def create_project(self):
- req_p = ProjectCreateRequest('functest', 'vping-ssh test')
+ req_p = project_models.ProjectCreateRequest('functest',
+ 'vping-ssh test')
self.create_help('/api/v1/projects', req_p)
self.project = req_p.name
class TestCaseCreate(TestCaseBase):
def test_noBody(self):
(code, body) = self.create(None, 'vping')
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
def test_noProject(self):
code, body = self.create(self.req_d, 'noProject')
- self.assertEqual(code, HTTP_FORBIDDEN)
+ self.assertEqual(code, constants.HTTP_FORBIDDEN)
self.assertIn('Could not find project', body)
def test_emptyName(self):
- req_empty = TestcaseCreateRequest('')
+ req_empty = testcase_models.TestcaseCreateRequest('')
(code, body) = self.create(req_empty, self.project)
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
self.assertIn('name missing', body)
def test_noneName(self):
- req_none = TestcaseCreateRequest(None)
+ req_none = testcase_models.TestcaseCreateRequest(None)
(code, body) = self.create(req_none, self.project)
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
self.assertIn('name missing', body)
def test_success(self):
code, body = self.create_d()
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
self.assert_create_body(body, None, self.project)
def test_alreadyExist(self):
self.create_d()
code, body = self.create_d()
- self.assertEqual(code, HTTP_FORBIDDEN)
+ self.assertEqual(code, constants.HTTP_FORBIDDEN)
self.assertIn('already exists', body)
class TestCaseGet(TestCaseBase):
def test_notExist(self):
code, body = self.get('notExist')
- self.assertEqual(code, HTTP_NOT_FOUND)
+ self.assertEqual(code, constants.HTTP_NOT_FOUND)
def test_getOne(self):
self.create_d()
code, body = self.get(self.req_d.name)
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
self.assert_body(body)
def test_list(self):
class TestCaseUpdate(TestCaseBase):
def test_noBody(self):
code, _ = self.update(case='noBody')
- self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertEqual(code, constants.HTTP_BAD_REQUEST)
def test_notFound(self):
code, _ = self.update(self.update_e, 'notFound')
- self.assertEqual(code, HTTP_NOT_FOUND)
+ self.assertEqual(code, constants.HTTP_NOT_FOUND)
def test_newNameExist(self):
self.create_d()
self.create_e()
code, body = self.update(self.update_e, self.req_d.name)
- self.assertEqual(code, HTTP_FORBIDDEN)
+ self.assertEqual(code, constants.HTTP_FORBIDDEN)
self.assertIn("already exists", body)
def test_noUpdate(self):
self.create_d()
code, body = self.update(self.update_d, self.req_d.name)
- self.assertEqual(code, HTTP_FORBIDDEN)
+ self.assertEqual(code, constants.HTTP_FORBIDDEN)
self.assertIn("Nothing to update", body)
def test_success(self):
_id = body._id
code, body = self.update(self.update_e, self.req_d.name)
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
self.assertEqual(_id, body._id)
self.assert_update_body(self.req_d, body, self.update_e)
update = copy.deepcopy(self.update_d)
update.description = {'2. change': 'dollar change'}
code, body = self.update(update, self.req_d.name)
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
class TestCaseDelete(TestCaseBase):
def test_notFound(self):
code, body = self.delete('notFound')
- self.assertEqual(code, HTTP_NOT_FOUND)
+ self.assertEqual(code, constants.HTTP_NOT_FOUND)
def test_success(self):
self.create_d()
code, body = self.delete(self.req_d.name)
- self.assertEqual(code, HTTP_OK)
+ self.assertEqual(code, constants.HTTP_OK)
self.assertEqual(body, '')
code, body = self.get(self.req_d.name)
- self.assertEqual(code, HTTP_NOT_FOUND)
+ self.assertEqual(code, constants.HTTP_NOT_FOUND)
if __name__ == '__main__':
##############################################################################
import unittest
-from test_base import TestBase
-from opnfv_testapi.resources.models import Versions
+from opnfv_testapi.resources import models
+import test_base as base
-class TestVersionBase(TestBase):
+class TestVersionBase(base.TestBase):
def setUp(self):
super(TestVersionBase, self).setUp()
- self.list_res = Versions
+ self.list_res = models.Versions
self.basePath = '/versions'
Code Review / releng.git / commitdiff