from models import CreateResponse
from resources.result_models import TestResult
-from resources.testcase_models import Testcase
from common.constants import DEFAULT_REPRESENTATION, HTTP_BAD_REQUEST, \
HTTP_NOT_FOUND, HTTP_FORBIDDEN
from common.config import prepare_put_request
self.json_args = None
self.table = None
self.table_cls = None
+ self.db_projects = 'projects'
+ self.db_pods = 'pods'
+ self.db_testcases = 'testcases'
+ self.db_results = 'results'
def prepare(self):
if self.request.method != "GET" and self.request.method != "DELETE":
@asynchronous
@gen.coroutine
- def _create(self, error):
+ def _create(self, db_checks, **kwargs):
+ """
+ :param db_checks: [(table, exist, query, (error, message))]
+ :param db_op: (insert/remove)
+ :param res_op: (_create_response/None)
+ :return:
+ """
if self.json_args is None:
- raise HTTPError(HTTP_BAD_REQUEST, 'no body')
+ raise HTTPError(HTTP_BAD_REQUEST, "no body")
data = self.table_cls.from_dict(self.json_args)
name = data.name
raise HTTPError(HTTP_BAD_REQUEST,
'{} name missing'.format(self.table[:-1]))
- exist_data = yield self._eval_db(self.table, 'find_one',
- {"name": name})
- if exist_data is not None:
- raise HTTPError(HTTP_FORBIDDEN,
- error.format(name, self.table[:-1]))
+ for k, v in kwargs.iteritems():
+ data.__setattr__(k, v)
+
+ for table, exist, query, error in db_checks:
+ check = yield self._eval_db(table, 'find_one', query(data))
+ if (exist and not check) or (not exist and check):
+ code, message = error(data)
+ raise HTTPError(code, message)
+
data.creation_date = datetime.now()
yield self._eval_db(self.table, 'insert', data.format())
self.finish_request(self._create_response(name))
yield self._eval_db(self.table, 'remove', query)
self.finish_request()
- def _eval_db(self, table, method, param):
- return eval('self.db.%s.%s(param)' % (table, method))
-
-
-class VersionHandler(GenericApiHandler):
- """ Display a message for the API version """
- def get(self):
- self.finish_request([{'v1': 'basics'}])
-
-
-class TestcaseHandler(GenericApiHandler):
- """
- TestCasesHandler Class
- Handle the requests about the Test cases for test projects
- HTTP Methdods :
- - GET : Get all test cases and details about a specific one
- - POST : Add a test project
- - PUT : Edit test projects information (name and/or description)
- """
-
- def initialize(self):
- """ Prepares the database for the entire class """
- super(TestcaseHandler, self).initialize()
-
- @asynchronous
- @gen.coroutine
- def get(self, project_name, case_name=None):
- """
- Get testcases(s) info
- :param project_name:
- :param case_name:
- """
-
- query = {'project_name': project_name}
-
- if case_name is not None:
- query["name"] = case_name
- answer = yield self.db.testcases.find_one(query)
- if answer is None:
- raise HTTPError(HTTP_NOT_FOUND,
- "{} Not Exist".format(case_name))
- else:
- answer = format_data(answer, Testcase)
- else:
- res = []
- cursor = self.db.testcases.find(query)
- while (yield cursor.fetch_next):
- res.append(format_data(cursor.next_object(), Testcase))
- answer = {'testcases': res}
-
- self.finish_request(answer)
-
@asynchronous
@gen.coroutine
- def post(self, project_name):
- """ Create a test case"""
-
- if self.json_args is None:
- raise HTTPError(HTTP_BAD_REQUEST,
- "Check your request payload")
-
- # retrieve test project
- project = yield self.db.projects.find_one(
- {"name": project_name})
- if project is None:
- raise HTTPError(HTTP_FORBIDDEN,
- "Could not find project {}"
- .format(project_name))
-
- case_name = self.json_args.get('name')
- the_testcase = yield self.db.testcases.find_one(
- {"project_name": project_name, 'name': case_name})
- if the_testcase:
- raise HTTPError(HTTP_FORBIDDEN,
- "{} already exists as a case in project {}"
- .format(case_name, project_name))
-
- testcase = Testcase.from_dict(self.json_args)
- testcase.project_name = project_name
- testcase.creation_date = datetime.now()
-
- yield self.db.testcases.insert(testcase.format())
- self.finish_request(self._create_response(testcase.name))
-
- @asynchronous
- @gen.coroutine
- def put(self, project_name, case_name):
- """
- Updates the name and description of a test case
- :raises HTTPError (HTTP_NOT_FOUND, HTTP_FORBIDDEN)
- """
-
- query = {'project_name': project_name, 'name': case_name}
-
+ def _update(self, query, db_keys):
if self.json_args is None:
raise HTTPError(HTTP_BAD_REQUEST, "No payload")
- # check if there is a case for the project in url parameters
- from_testcase = yield self.db.testcases.find_one(query)
- if from_testcase is None:
+ # check old data exist
+ from_data = yield self._eval_db(self.table, 'find_one', query)
+ if from_data is None:
raise HTTPError(HTTP_NOT_FOUND,
- "{} could not be found as a {} case to be updated"
- .format(case_name, project_name))
-
- testcase = Testcase.from_dict(from_testcase)
- new_name = self.json_args.get("name")
- new_project_name = self.json_args.get("project_name")
- if not new_project_name:
- new_project_name = project_name
- new_description = self.json_args.get("description")
-
- # check if there is not an existing test case
- # with the name provided in the json payload
- if new_name != case_name or new_project_name != project_name:
- to_testcase = yield self.db.testcases.find_one(
- {'project_name': new_project_name, 'name': new_name})
- if to_testcase is not None:
- raise HTTPError(HTTP_FORBIDDEN,
- "{} already exists as a case in project"
- .format(new_name, new_project_name))
+ "{} could not be found in table [{}]"
+ .format(query, self.table))
- # new dict for changes
- request = dict()
- request = prepare_put_request(request,
- "name",
- new_name,
- testcase.name)
- request = prepare_put_request(request,
- "project_name",
- new_project_name,
- testcase.project_name)
- request = prepare_put_request(request,
- "description",
- new_description,
- testcase.description)
-
- # we raise an exception if there isn't a change
- if not request:
- raise HTTPError(HTTP_FORBIDDEN,
- "Nothing to update")
+ 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(self.table, 'find_one', new_query)
+ if to_data is not None:
+ raise HTTPError(HTTP_FORBIDDEN,
+ "{} already exists in table [{}]"
+ .format(new_query, self.table))
# we merge the whole document """
- edit_request = testcase.format()
- edit_request.update(request)
+ edit_request = data.format()
+ edit_request.update(self._update_request(data))
""" Updating the DB """
- yield self.db.testcases.update(query, edit_request)
- new_case = yield self.db.testcases.find_one({"_id": testcase._id})
- self.finish_request(format_data(new_case, Testcase))
+ yield self._eval_db(self.table, 'update', query, edit_request)
+ edit_request['_id'] = str(data._id)
+ self.finish_request(edit_request)
- @asynchronous
- @gen.coroutine
- def delete(self, project_name, case_name):
- """ Remove a test case"""
+ def _update_request(self, data):
+ request = dict()
+ for k, v in self.json_args.iteritems():
+ request = prepare_put_request(request, k, v,
+ data.__getattribute__(k))
+ if not request:
+ raise HTTPError(HTTP_FORBIDDEN, "Nothing to update")
+ return request
- query = {'project_name': project_name, 'name': case_name}
+ def _update_query(self, keys, data):
+ query = dict()
+ equal = True
+ for key in keys:
+ new = self.json_args.get(key)
+ old = data.__getattribute__(key)
+ if new is None:
+ new = old
+ elif new != old:
+ equal = False
+ query[key] = new
+ return equal, query
- # check for an existing case to be deleted
- testcase = yield self.db.testcases.find_one(query)
- if testcase is None:
- raise HTTPError(HTTP_NOT_FOUND,
- "{}/{} could not be found as a case to be deleted"
- .format(project_name, case_name))
+ def _eval_db(self, table, method, *args):
+ return eval('self.db.%s.%s(*args)' % (table, method))
- # just delete it, or maybe save it elsewhere in a future
- yield self.db.testcases.remove(query)
- self.finish_request()
+
+class VersionHandler(GenericApiHandler):
+ """ Display a message for the API version """
+ def get(self):
+ self.finish_request([{'v1': 'basics'}])
class TestResultsHandler(GenericApiHandler):
-from tornado import gen
-from tornado.web import asynchronous
-
from tornado_swagger_ui.tornado_swagger import swagger
from handlers import GenericApiHandler
from pod_models import Pod
+from common.constants import HTTP_FORBIDDEN
class GenericPodHandler(GenericApiHandler):
"""
self._list()
- @gen.coroutine
@swagger.operation(nickname='create')
def post(self):
"""
@raise 403: pod already exists
@raise 400: post without body
"""
- self._create('{} already exists as a {}')
+ def query(data):
+ return {'name': data.name}
+
+ def error(data):
+ message = '{} already exists as a pod'.format(data.name)
+ return HTTP_FORBIDDEN, message
+
+ db_check = [(self.table, False, query, error)]
+ self._create(db_check)
class PodGURHandler(GenericPodHandler):
query['name'] = pod_name
self._get_one(query)
- @asynchronous
- @gen.coroutine
def delete(self, pod_name):
""" Remove a POD
-from tornado import gen
-from tornado.web import HTTPError, asynchronous
-
from tornado_swagger_ui.tornado_swagger import swagger
-from handlers import GenericApiHandler, prepare_put_request, format_data
-from common.constants import HTTP_BAD_REQUEST, HTTP_FORBIDDEN, HTTP_NOT_FOUND
+from handlers import GenericApiHandler
+from common.constants import HTTP_FORBIDDEN
from project_models import Project
@raise 403: project already exists
@raise 400: post without body
"""
- self._create('{} already exists as a {}')
+ def query(data):
+ return {'name': data.name}
+
+ def error(data):
+ message = '{} already exists as a project'.format(data.name)
+ return HTTP_FORBIDDEN, message
+
+ db_check = [(self.table, False, query, error)]
+ self._create(db_check)
class ProjectGURHandler(GenericProjectHandler):
"""
self._get_one({'name': project_name})
- @asynchronous
- @gen.coroutine
@swagger.operation(nickname="update")
def put(self, project_name):
"""
@raise 404: project not exist
@raise 403: new project name already exist or nothing to update
"""
- if self.json_args is None:
- raise HTTPError(HTTP_BAD_REQUEST)
-
query = {'name': project_name}
- from_project = yield self.db.projects.find_one(query)
- if from_project is None:
- raise HTTPError(HTTP_NOT_FOUND,
- "{} could not be found".format(project_name))
-
- project = Project.from_dict(from_project)
- new_name = self.json_args.get("name")
- new_description = self.json_args.get("description")
-
- # check for payload name parameter in db
- # avoid a request if the project name has not changed in the payload
- if new_name != project.name:
- to_project = yield self.db.projects.find_one(
- {"name": new_name})
- if to_project is not None:
- raise HTTPError(HTTP_FORBIDDEN,
- "{} already exists as a project"
- .format(new_name))
-
- # new dict for changes
- request = dict()
- request = prepare_put_request(request,
- "name",
- new_name,
- project.name)
- request = prepare_put_request(request,
- "description",
- new_description,
- project.description)
-
- """ raise exception if there isn't a change """
- if not request:
- raise HTTPError(HTTP_FORBIDDEN, "Nothing to update")
-
- """ we merge the whole document """
- edit_request = project.format()
- edit_request.update(request)
-
- """ Updating the DB """
- yield self.db.projects.update(query, edit_request)
- new_project = yield self.db.projects.find_one({"_id": project._id})
-
- self.finish_request(format_data(new_project, Project))
+ db_keys = ['name']
+ self._update(query, db_keys)
@swagger.operation(nickname='delete')
def delete(self, project_name):
@swagger.model()
-class Project:
+class Project(object):
def __init__(self,
name=None, _id=None, description=None, create_date=None):
self._id = _id
--- /dev/null
+from common.constants import HTTP_FORBIDDEN
+from resources.handlers import GenericApiHandler
+from resources.testcase_models import Testcase
+from tornado_swagger_ui.tornado_swagger import swagger
+
+
+class GenericTestcaseHandler(GenericApiHandler):
+ def __init__(self, application, request, **kwargs):
+ super(GenericTestcaseHandler, self).__init__(application,
+ request,
+ **kwargs)
+ self.table = self.db_testcases
+ self.table_cls = Testcase
+
+
+class TestcaseCLHandler(GenericTestcaseHandler):
+ @swagger.operation(nickname="list-all")
+ def get(self, project_name):
+ """
+ @description: list all testcases of a project by project_name
+ @return 200: return all testcases of this project,
+ empty list is no testcase exist in this project
+ @rtype: L{TestCases}
+ """
+ query = dict()
+ query['project_name'] = project_name
+ self._list(query)
+
+ @swagger.operation(nickname="create")
+ def post(self, project_name):
+ """
+ @description: create a testcase of a project by project_name
+ @param body: testcase to be created
+ @type body: L{TestcaseCreateRequest}
+ @in body: body
+ @rtype: L{Testcase}
+ @return 200: testcase is created in this project.
+ @raise 403: project not exist
+ or testcase already exists in this project
+ @raise 400: post without body
+ """
+ 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):
+ message = 'Could not find project [{}]'.format(data.project_name)
+ return 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
+
+ db_checks = [(self.db_projects, True, p_query, p_error),
+ (self.db_testcases, False, tc_query, tc_error)]
+ self._create(db_checks, project_name=project_name)
+
+
+class TestcaseGURHandler(GenericTestcaseHandler):
+ @swagger.operation(nickname='get-one')
+ def get(self, project_name, case_name):
+ """
+ @description: get a single testcase
+ by case_name and project_name
+ @rtype: L{Testcase}
+ @return 200: testcase exist
+ @raise 404: testcase not exist
+ """
+ query = dict()
+ query['project_name'] = project_name
+ query["name"] = case_name
+ self._get_one(query)
+
+ @swagger.operation(nickname="update")
+ def put(self, project_name, case_name):
+ """
+ @description: update a single testcase
+ by project_name and case_name
+ @param body: testcase to be updated
+ @type body: L{TestcaseUpdateRequest}
+ @in body: body
+ @rtype: L{Project}
+ @return 200: update success
+ @raise 404: testcase or project not exist
+ @raise 403: new testcase name already exist in project
+ or nothing to update
+ """
+ query = {'project_name': project_name, 'name': case_name}
+ db_keys = ['name', 'project_name']
+ self._update(query, db_keys)
+
+ @swagger.operation(nickname='delete')
+ def delete(self, project_name, case_name):
+ """
+ @description: delete a testcase by project_name and case_name
+ @return 200: delete success
+ @raise 404: testcase not exist
+ """
+ query = {'project_name': project_name, 'name': case_name}
+ self._delete(query)
+from tornado_swagger_ui.tornado_swagger import swagger
+
+__author__ = '__serena__'
+
+
+@swagger.model()
class TestcaseCreateRequest(object):
- def __init__(self, url=None, name=None, description=None):
+ def __init__(self, name, url=None, description=None):
self.name = name
self.url = url
self.description = description
}
+@swagger.model()
class TestcaseUpdateRequest(object):
def __init__(self, name=None, description=None, project_name=None):
self.name = name
}
-class Testcase:
+@swagger.model()
+class Testcase(object):
""" Describes a test case"""
def __init__(self):
}
+@swagger.model()
class Testcases(object):
+ """
+ @ptype testcases: C{list} of L{Testcase}
+ """
def __init__(self, testcases=list()):
self.testcases = testcases
import motor
from resources.handlers import VersionHandler, \
- TestcaseHandler, TestResultsHandler, DashboardHandler
+ TestResultsHandler, DashboardHandler
+from resources.testcase_handlers import TestcaseCLHandler, TestcaseGURHandler
from resources.pod_handlers import PodCLHandler, PodGURHandler
from resources.project_handlers import ProjectCLHandler, ProjectGURHandler
from common.config import APIConfig
# few examples
# GET /projects/qtip/cases => Get cases for qtip
- (r"/api/v1/projects/([^/]+)/cases", TestcaseHandler),
- (r"/api/v1/projects/([^/]+)/cases/([^/]+)", TestcaseHandler),
+ (r"/api/v1/projects/([^/]+)/cases", TestcaseCLHandler),
+ (r"/api/v1/projects/([^/]+)/cases/([^/]+)", TestcaseGURHandler),
# new path to avoid a long depth
# GET /results?project=functest&case=keystone.catalog&pod=1
from resources.pod_handlers import PodCLHandler, PodGURHandler
from resources.project_handlers import ProjectCLHandler, ProjectGURHandler
from resources.handlers import VersionHandler, \
- TestcaseHandler, TestResultsHandler, DashboardHandler
+ TestResultsHandler, DashboardHandler
+from resources.testcase_handlers import TestcaseCLHandler, TestcaseGURHandler
from resources.models import CreateResponse
import fake_pymongo
(r"/api/v1/pods/([^/]+)", PodGURHandler),
(r"/api/v1/projects", ProjectCLHandler),
(r"/api/v1/projects/([^/]+)", ProjectGURHandler),
- (r"/api/v1/projects/([^/]+)/cases", TestcaseHandler),
- (r"/api/v1/projects/([^/]+)/cases/([^/]+)", TestcaseHandler),
+ (r"/api/v1/projects/([^/]+)/cases", TestcaseCLHandler),
+ (r"/api/v1/projects/([^/]+)/cases/([^/]+)",
+ TestcaseGURHandler),
(r"/api/v1/results", TestResultsHandler),
(r"/api/v1/results([^/]*)", TestResultsHandler),
(r"/api/v1/results/([^/]*)", TestResultsHandler),
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('/cases/vping',
- self.case,
+ self.req_testcase = 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)
class TestCaseBase(TestBase):
def setUp(self):
super(TestCaseBase, self).setUp()
- self.req_d = TestcaseCreateRequest('/cases/vping_1',
- 'vping_1',
+ self.req_d = TestcaseCreateRequest('vping_1',
+ '/cases/vping_1',
'vping-ssh test')
- self.req_e = TestcaseCreateRequest('/cases/doctor_1',
- 'doctor_1',
+ self.req_e = TestcaseCreateRequest('doctor_1',
+ '/cases/doctor_1',
'create doctor')
self.update_d = TestcaseUpdateRequest('vping_1',
'vping-ssh test',
self.assertEqual(code, HTTP_FORBIDDEN)
self.assertIn('Could not find project', body)
+ def test_emptyName(self):
+ req_empty = TestcaseCreateRequest('')
+ (code, body) = self.create(req_empty, self.project)
+ self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertIn('testcase name missing', body)
+
+ def test_noneName(self):
+ req_none = TestcaseCreateRequest(None)
+ (code, body) = self.create(req_none, self.project)
+ self.assertEqual(code, HTTP_BAD_REQUEST)
+ self.assertIn('testcase name missing', body)
+
def test_success(self):
code, body = self.create_d()
self.assertEqual(code, HTTP_OK)
Code Review / releng.git / commitdiff