# feng.xiaowei@zte.com.cn refactor test_project to project 5-19-2016
# feng.xiaowei@zte.com.cn refactor response body 5-19-2016
# feng.xiaowei@zte.com.cn refactor pod/project response info 5-19-2016
+# feng.xiaowei@zte.com.cn refactor case related handler 5-20-2016
##############################################################################
import json
from tornado import gen
from datetime import datetime, timedelta
-from models import TestCase, TestResult, CreateResponse
+from models import TestResult, CreateResponse
+from resources.testcase_models import Testcase
from resources.project_models import Project
from resources.pod_models import Pod
from common.constants import DEFAULT_REPRESENTATION, HTTP_BAD_REQUEST, \
Get all pods or a single pod
:param pod_id:
"""
- get_request = dict()
+ query = dict()
if pod_name is not None:
- get_request["name"] = pod_name
- answer = yield self.db.pods.find_one(get_request)
+ query["name"] = pod_name
+ answer = yield self.db.pods.find_one(query)
if answer is None:
raise HTTPError(HTTP_NOT_FOUND,
"{} Not Exist".format(pod_name))
answer = format_data(answer, Pod)
else:
res = []
- cursor = self.db.pods.find(get_request)
+ cursor = self.db.pods.find(query)
while (yield cursor.fetch_next):
res.append(format_data(cursor.next_object(), Pod))
- answer = dict()
- answer['pods'] = res
+ answer = {'pods': res}
self.finish_request(answer)
query = {"name": self.json_args.get("name")}
# check for existing name in db
- mongo_dict = yield self.db.pods.find_one(query)
- if mongo_dict is not None:
+ the_pod = yield self.db.pods.find_one(query)
+ if the_pod is not None:
raise HTTPError(HTTP_FORBIDDEN,
"{} already exists as a pod".format(
self.json_args.get("name")))
res = yield self.db.projects.remove(
{'name': pod_name})
- meta = dict()
- meta["success"] = True
- meta["deletion-data"] = res
-
- answer = dict()
- answer["meta"] = meta
-
self.finish_request(answer)
"""
pass
-class TestProjectHandler(GenericApiHandler):
+class ProjectHandler(GenericApiHandler):
"""
TestProjectHandler Class
Handle the requests about the Test projects
def initialize(self):
""" Prepares the database for the entire class """
- super(TestProjectHandler, self).initialize()
+ super(ProjectHandler, self).initialize()
@asynchronous
@gen.coroutine
:param project_name:
"""
- get_request = dict()
+ query = dict()
if project_name is not None:
- get_request["name"] = project_name
- answer = yield self.db.projects.find_one(get_request)
+ query["name"] = project_name
+ answer = yield self.db.projects.find_one(query)
if answer is None:
raise HTTPError(HTTP_NOT_FOUND,
"{} Not Exist".format(project_name))
answer = format_data(answer, Project)
else:
res = []
- cursor = self.db.projects.find(get_request)
+ cursor = self.db.projects.find(query)
while (yield cursor.fetch_next):
res.append(format_data(cursor.next_object(), Project))
- answer = dict()
- answer['projects'] = res
+ answer = {'projects': res}
self.finish_request(answer)
query = {"name": self.json_args.get("name")}
# check for name in db
- mongo_dict = yield self.db.projects.find_one(query)
- if mongo_dict is not None:
+ the_project = yield self.db.projects.find_one(query)
+ if the_project is not None:
raise HTTPError(HTTP_FORBIDDEN,
"{} already exists as a project".format(
self.json_args.get("name")))
def put(self, project_name):
""" Updates the name and description of a test project"""
- print "PUT request for : {}".format(project_name)
-
if self.json_args is None:
raise HTTPError(HTTP_BAD_REQUEST)
query = {'name': project_name}
- mongo_dict = yield self.db.projects.find_one(query)
- project = Project.from_dict(mongo_dict)
- if project is None:
+ 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:
- mongo_dict = yield self.db.projects.find_one(
+ to_project = yield self.db.projects.find_one(
{"name": new_name})
- if mongo_dict is not None:
+ if to_project is not None:
raise HTTPError(HTTP_FORBIDDEN,
"{} already exists as a project"
.format(new_name))
@gen.coroutine
def delete(self, project_name):
""" Remove a test project"""
-
- print "DELETE request for : {}".format(project_name)
+ query = {'name': project_name}
# check for an existing project to be deleted
- mongo_dict = yield self.db.projects.find_one(
- {'name': project_name})
- test_project = Project.from_dict(mongo_dict)
- if test_project is None:
+ project = yield self.db.projects.find_one(query)
+ if project is None:
raise HTTPError(HTTP_NOT_FOUND,
"{} could not be found as a project to be deleted"
.format(project_name))
# just delete it, or maybe save it elsewhere in a future
- yield self.db.projects.remove({'name': project_name})
+ yield self.db.projects.remove(query)
self.finish_request()
-class TestCasesHandler(GenericApiHandler):
+class TestcaseHandler(GenericApiHandler):
"""
TestCasesHandler Class
Handle the requests about the Test cases for test projects
def initialize(self):
""" Prepares the database for the entire class """
- super(TestCasesHandler, self).initialize()
+ super(TestcaseHandler, self).initialize()
@asynchronous
@gen.coroutine
:param case_name:
"""
- if case_name is None:
- case_name = ""
-
- get_request = dict()
- get_request["project_name"] = project_name
-
- if len(case_name) > 0:
- get_request["name"] = case_name
-
- res = []
- cursor = self.db.test_cases.find(get_request)
- print get_request
- while (yield cursor.fetch_next):
- test_case = TestCase.test_case_from_dict(cursor.next_object())
- res.append(test_case.format_http())
-
- meta = dict()
- meta["total"] = len(res)
- meta["success"] = True if len(res) > 0 else False
+ query = {'project_name': project_name}
- answer = dict()
- answer["test_cases"] = res
- answer["meta"] = meta
+ 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)
def post(self, project_name):
""" Create a test case"""
- print "POST Request for {}".format(project_name)
-
if self.json_args is None:
raise HTTPError(HTTP_BAD_REQUEST,
"Check your request payload")
# retrieve test project
- mongo_dict = yield self.db.projects.find_one(
+ project = yield self.db.projects.find_one(
{"name": project_name})
- if mongo_dict is None:
+ if project is None:
raise HTTPError(HTTP_FORBIDDEN,
"Could not find project {}"
.format(project_name))
- # test_project = TestProject.from_dict(self.json_args)
+ 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))
- case = TestCase.test_case_from_dict(self.json_args)
- case.project_name = project_name
- case.creation_date = datetime.now()
+ testcase = Testcase.from_dict(self.json_args)
+ testcase.project_name = project_name
+ testcase.creation_date = datetime.now()
- future = self.db.test_cases.insert(case.format())
- result = yield future
- case._id = result
- self.finish_request(case.format_http())
+ yield self.db.testcases.insert(testcase.format())
+ res = CreateResponse(self.request.full_url() + '/{}'.format(testcase.name))
+ self.finish_request(res.format())
@asynchronous
@gen.coroutine
:raises HTTPError (HTTP_NOT_FOUND, HTTP_FORBIDDEN)
"""
- print "PUT request for : {}/{}".format(project_name, case_name)
- case_request = {'project_name': project_name, 'name': case_name}
+ query = {'project_name': project_name, 'name': case_name}
+
+ 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
- mongo_dict = yield self.db.test_cases.find_one(case_request)
- test_case = TestCase.test_case_from_dict(mongo_dict)
- if test_case is None:
+ from_testcase = yield self.db.testcases.find_one(query)
+ if from_testcase 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
- mongo_dict = yield self.db.test_cases.find_one(
- {'project_name': new_project_name, 'name': new_name})
- if mongo_dict is not None:
- raise HTTPError(HTTP_FORBIDDEN,
- "{} already exists as a project"
- .format(new_name))
+ 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))
# new dict for changes
request = dict()
request = prepare_put_request(request,
"name",
new_name,
- test_case.name)
+ testcase.name)
request = prepare_put_request(request,
"project_name",
new_project_name,
- test_case.project_name)
+ testcase.project_name)
request = prepare_put_request(request,
"description",
new_description,
- test_case.description)
+ testcase.description)
# we raise an exception if there isn't a change
if not request:
"Nothing to update")
# we merge the whole document """
- edit_request = test_case.format()
+ edit_request = testcase.format()
edit_request.update(request)
""" Updating the DB """
- res = yield self.db.test_cases.update(case_request, edit_request)
- print res
- edit_request["_id"] = str(test_case._id)
-
- self.finish_request({"message": "success", "content": edit_request})
+ 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))
@asynchronous
@gen.coroutine
def delete(self, project_name, case_name):
""" Remove a test case"""
- print "DELETE request for : {}/{}".format(project_name, case_name)
- case_request = {'project_name': project_name, 'name': case_name}
+ query = {'project_name': project_name, 'name': case_name}
# check for an existing case to be deleted
- mongo_dict = yield self.db.test_cases.find_one(case_request)
- test_project = Project.from_dict(mongo_dict)
- if test_project is None:
+ 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))
# just delete it, or maybe save it elsewhere in a future
- res = yield self.db.projects.remove(case_request)
- print res
-
- self.finish_request({"message": "success"})
+ yield self.db.testcases.remove(query)
+ self.finish_request()
class TestResultsHandler(GenericApiHandler):
.format(self.json_args.get("project_name")))
# check for case
- mongo_dict = yield self.db.test_cases.find_one(
+ mongo_dict = yield self.db.testcases.find_one(
{"name": self.json_args.get("case_name")})
if mongo_dict is None:
raise HTTPError(HTTP_NOT_FOUND,
# feng.xiaowei@zte.com.cn mv TestProject to project_models.py 5-19-2016\r
# feng.xiaowei@zte.com.cn delete meta class 5-19-2016\r
# feng.xiaowei@zte.com.cn add CreateResponse 5-19-2016\r
+# feng.xiaowei@zte.com.cn mv TestCase to testcase_models.py 5-20-2016\r
##############################################################################\r
\r
\r
return {'href': self.href}\r
\r
\r
-class TestCase:\r
- """ Describes a test case"""\r
-\r
- def __init__(self):\r
- self._id = None\r
- self.name = None\r
- self.project_name = None\r
- self.description = None\r
- self.url = None\r
- self.creation_date = None\r
-\r
- @staticmethod\r
- def test_case_from_dict(testcase_dict):\r
-\r
- if testcase_dict is None:\r
- return None\r
-\r
- t = TestCase()\r
- t._id = testcase_dict.get('_id')\r
- t.project_name = testcase_dict.get('project_name')\r
- t.creation_date = testcase_dict.get('creation_date')\r
- t.name = testcase_dict.get('name')\r
- t.description = testcase_dict.get('description')\r
- t.url = testcase_dict.get('url')\r
-\r
- return t\r
-\r
- def format(self):\r
- return {\r
- "name": self.name,\r
- "description": self.description,\r
- "project_name": self.project_name,\r
- "creation_date": str(self.creation_date),\r
- "url": self.url\r
- }\r
-\r
- def format_http(self, test_project=None):\r
- res = {\r
- "_id": str(self._id),\r
- "name": self.name,\r
- "description": self.description,\r
- "creation_date": str(self.creation_date),\r
- "url": self.url,\r
- }\r
- if test_project is not None:\r
- res["test_project"] = test_project\r
-\r
- return res\r
-\r
-\r
class TestResult:\r
""" Describes a test result"""\r
\r
--- /dev/null
+class TestcaseCreateRequest(object):
+ def __init__(self, url=None, name=None, description=None):
+ self.name = name
+ self.url = url
+ self.description = description
+
+ def format(self):
+ return {
+ "name": self.name,
+ "description": self.description,
+ "url": self.url,
+ }
+
+
+class TestcaseUpdateRequest(object):
+ def __init__(self, name=None, description=None, project_name=None):
+ self.name = name
+ self.description = description
+ self.project_name = project_name
+
+ def format(self):
+ return {
+ "name": self.name,
+ "description": self.description,
+ "project_name": self.project_name,
+ }
+
+
+class Testcase:
+ """ Describes a test case"""
+
+ def __init__(self):
+ self._id = None
+ self.name = None
+ self.project_name = None
+ self.description = None
+ self.url = None
+ self.creation_date = None
+
+ @staticmethod
+ def from_dict(a_dict):
+
+ if a_dict is None:
+ return None
+
+ t = Testcase()
+ t._id = a_dict.get('_id')
+ t.project_name = a_dict.get('project_name')
+ t.creation_date = a_dict.get('creation_date')
+ t.name = a_dict.get('name')
+ t.description = a_dict.get('description')
+ t.url = a_dict.get('url')
+
+ return t
+
+ def format(self):
+ return {
+ "name": self.name,
+ "description": self.description,
+ "project_name": self.project_name,
+ "creation_date": str(self.creation_date),
+ "url": self.url
+ }
+
+ def format_http(self):
+ return {
+ "_id": str(self._id),
+ "name": self.name,
+ "project_name": self.project_name,
+ "description": self.description,
+ "creation_date": str(self.creation_date),
+ "url": self.url,
+ }
+
+
+class Testcases(object):
+ def __init__(self, testcases=list()):
+ self.testcases = testcases
+
+ @staticmethod
+ def from_dict(res_dict):
+ if res_dict is None:
+ return None
+
+ res = Testcases()
+ for testcase in res_dict.get('testcases'):
+ res.testcases.append(Testcase.from_dict(testcase))
+ return res
import argparse
from resources.handlers import VersionHandler, PodHandler, \
- TestProjectHandler, TestCasesHandler, TestResultsHandler, DashboardHandler
+ ProjectHandler, TestcaseHandler, TestResultsHandler, DashboardHandler
from common.config import APIConfig
# few examples:
# GET /projects
# GET /projects/yardstick
- (r"/projects", TestProjectHandler),
- (r"/projects/([^/]+)", TestProjectHandler),
+ (r"/projects", ProjectHandler),
+ (r"/projects/([^/]+)", ProjectHandler),
# few examples
# GET /projects/qtip/cases => Get cases for qtip
#
- (r"/projects/([^/]+)/cases", TestCasesHandler),
- (r"/projects/([^/]+)/cases/([^/]+)", TestCasesHandler),
+ (r"/projects/([^/]+)/cases", TestcaseHandler),
+ (r"/projects/([^/]+)/cases/([^/]+)", TestcaseHandler),
# (r"/test_cases/([^/]+)", TestCasesHandler),
# new path to avoid a long depth
pods = MemDb()
projects = MemDb()
-test_cases = MemDb()
+testcases = MemDb()
test_results = MemDb()
from tornado.testing import AsyncHTTPTestCase
from resources.handlers import VersionHandler, PodHandler, \
- TestProjectHandler, TestCasesHandler, TestResultsHandler, DashboardHandler
+ ProjectHandler, TestcaseHandler, TestResultsHandler, DashboardHandler
from resources.models import CreateResponse
import fake_pymongo
(r"/version", VersionHandler),
(r"/pods", PodHandler),
(r"/pods/([^/]+)", PodHandler),
- (r"/projects", TestProjectHandler),
- (r"/projects/([^/]+)", TestProjectHandler),
- (r"/projects/([^/]+)/cases", TestCasesHandler),
- (r"/projects/([^/]+)/cases/([^/]+)", TestCasesHandler),
+ (r"/projects", ProjectHandler),
+ (r"/projects/([^/]+)", ProjectHandler),
+ (r"/projects/([^/]+)/cases", TestcaseHandler),
+ (r"/projects/([^/]+)/cases/([^/]+)", TestcaseHandler),
(r"/results", TestResultsHandler),
(r"/results([^/]*)", TestResultsHandler),
(r"/results/([^/]*)", TestResultsHandler),
debug=True,
)
- def create_d(self):
- return self.create(self.req_d)
+ def create_d(self, *args):
+ return self.create(self.req_d, *args)
- def create_e(self):
- return self.create(self.req_e)
+ def create_e(self, *args):
+ return self.create(self.req_e, *args)
- def create(self, req=None):
+ def create(self, req=None, *args):
if req:
req = req.format()
- res = self.fetch(self.basePath,
+ res = self.fetch(self._get_uri(*args),
method='POST',
body=json.dumps(req),
headers=self.headers)
return self._get_return(res, self.create_res)
- def get(self, item=None):
- res = self.fetch(self._get_uri(item),
+ def create_help(self, uri, req, cls):
+ res = self.fetch(uri,
+ method='POST',
+ body=json.dumps(req.format()),
+ headers=self.headers)
+
+ return self._get_return(res, cls)
+
+ def get(self, *args):
+ res = self.fetch(self._get_uri(*args),
method='GET',
headers=self.headers)
def inner():
- return self.get_res if item else self.list_res
+ new_args, num = self._get_valid_args(*args)
+ return self.get_res if num != self._need_arg_num() else self.list_res
return self._get_return(res, inner())
- def update(self, item, new=None):
+ def update(self, new=None, *args):
if new:
new = new.format()
- res = self.fetch(self._get_uri(item),
+ res = self.fetch(self._get_uri(*args),
method='PUT',
body=json.dumps(new),
headers=self.headers)
return self._get_return(res, self.update_res)
- def delete(self, item):
- res = self.fetch(self._get_uri(item),
+ def delete(self, *args):
+ res = self.fetch(self._get_uri(*args),
method='DELETE',
headers=self.headers)
- return res.code
+ return res.code, res.body
- def _get_uri(self, item=None):
+ @staticmethod
+ def _get_valid_args(*args):
+ new_args = tuple(['%s' % arg for arg in args if arg is not None])
+ return new_args, len(new_args)
+
+ def _need_arg_num(self):
+ return self.basePath.count('%s')
+
+ def _get_uri(self, *args):
+ new_args, num = self._get_valid_args(*args)
uri = self.basePath
- if item:
- uri += '/{}'.format(item)
- return uri
+ if num != self._need_arg_num():
+ uri += '/%s'
+
+ return uri % tuple(['%s' % arg for arg in new_args])
def _get_return(self, res, cls):
code = res.code
def _get_return_body(code, body, cls):
return cls.from_dict(json.loads(body)) if code < 300 else body
- def assert_create_body(self, body, req=None):
- print(body.href)
+ def assert_create_body(self, body, req=None, *args):
if not req:
req = self.req_d
- self.assertIn(self._get_uri(req.name), body.href)
+ new_args = args + tuple([req.name])
+ self.assertIn(self._get_uri(*new_args), body.href)
@staticmethod
def _clear():
fake_pymongo.pods.clear()
fake_pymongo.projects.clear()
- fake_pymongo.test_cases.clear()
+ fake_pymongo.testcases.clear()
fake_pymongo.test_results.clear()
class TestProjectUpdate(TestProjectBase):
def test_withoutBody(self):
- code, _ = self.update('noBody')
+ code, _ = self.update(None, 'noBody')
self.assertEqual(code, HTTP_BAD_REQUEST)
def test_notFound(self):
- code, _ = self.update('notFound', self.req_e)
+ code, _ = self.update(self.req_e, 'notFound')
self.assertEqual(code, HTTP_NOT_FOUND)
def test_newNameExist(self):
self.create_d()
self.create_e()
- code, body = self.update(self.req_d.name, self.req_e)
+ code, body = self.update(self.req_e, self.req_d.name)
self.assertEqual(code, HTTP_FORBIDDEN)
self.assertIn("already exists", body)
def test_noUpdate(self):
self.create_d()
- code, body = self.update(self.req_d.name, self.req_d)
+ code, body = self.update(self.req_d, self.req_d.name)
self.assertEqual(code, HTTP_FORBIDDEN)
self.assertIn("Nothing to update", body)
_id = body._id
req = ProjectCreateRequest('newName', 'new description')
- code, body = self.update(self.req_d.name, req)
+ code, body = self.update(req, self.req_d.name)
self.assertEqual(code, HTTP_OK)
self.assertEqual(_id, body._id)
self.assert_body(body, req)
class TestProjectDelete(TestProjectBase):
def test_notFound(self):
- code = self.delete('notFound')
+ code, body = self.delete('notFound')
self.assertEqual(code, HTTP_NOT_FOUND)
def test_success(self):
self.create_d()
- code = self.delete(self.req_d.name)
+ code, body = self.delete(self.req_d.name)
self.assertEqual(code, HTTP_OK)
+ self.assertEqual(body, '')
+
+ code, body = self.get(self.req_d.name)
+ self.assertEqual(code, HTTP_NOT_FOUND)
if __name__ == '__main__':
unittest.main()
--- /dev/null
+import unittest
+
+from test_base import TestBase
+from resources.testcase_models import TestcaseCreateRequest, \
+ Testcase, Testcases, TestcaseUpdateRequest
+from resources.project_models import ProjectCreateRequest, Project
+from common.constants import HTTP_OK, HTTP_BAD_REQUEST, \
+ HTTP_FORBIDDEN, HTTP_NOT_FOUND
+
+
+__author__ = '__serena__'
+
+
+class TestCaseBase(TestBase):
+ def setUp(self):
+ super(TestCaseBase, self).setUp()
+ self.req_d = TestcaseCreateRequest('/cases/vping_1', 'vping_1', 'vping-ssh test')
+ self.req_e = TestcaseCreateRequest('/cases/doctor_1', '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.basePath = '/projects/%s/cases'
+ self.create_project()
+
+ def assert_body(self, case, req=None):
+ if not req:
+ req = self.req_d
+ self.assertEqual(case.name, req.name)
+ self.assertEqual(case.description, req.description)
+ self.assertEqual(case.url, req.url)
+ self.assertIsNotNone(case._id)
+ self.assertIsNotNone(case.creation_date)
+
+ def assert_update_body(self, old, new, req=None):
+ if not req:
+ req = self.req_d
+ self.assertEqual(new.name, req.name)
+ self.assertEqual(new.description, req.description)
+ self.assertEqual(new.url, old.url)
+ self.assertIsNotNone(new._id)
+ self.assertIsNotNone(new.creation_date)
+
+ def create_project(self):
+ req_p = ProjectCreateRequest('functest', 'vping-ssh test')
+ self.create_help('/projects', req_p, Project)
+ self.project = req_p.name
+
+ def create_d(self):
+ return super(TestCaseBase, self).create_d(self.project)
+
+ def create_e(self):
+ return super(TestCaseBase, self).create_e(self.project)
+
+ def get(self, case=None):
+ return super(TestCaseBase, self).get(self.project, case)
+
+ def update(self, new=None, case=None):
+ return super(TestCaseBase, self).update(new, self.project, case)
+
+ def delete(self, case):
+ return super(TestCaseBase, self).delete(self.project, case)
+
+
+class TestCaseCreate(TestCaseBase):
+ def test_noBody(self):
+ (code, body) = self.create(None, 'vping')
+ self.assertEqual(code, HTTP_BAD_REQUEST)
+
+ def test_noProject(self):
+ code, body = self.create(self.req_d, 'noProject')
+ self.assertEqual(code, HTTP_FORBIDDEN)
+ self.assertIn('Could not find project', body)
+
+ def test_success(self):
+ code, body = self.create_d()
+ self.assertEqual(code, 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.assertIn('already exists', body)
+
+
+class TestCaseGet(TestCaseBase):
+ def test_notExist(self):
+ code, body = self.get('notExist')
+ self.assertEqual(code, HTTP_NOT_FOUND)
+
+ def test_getOne(self):
+ self.create_d()
+ code, body = self.get(self.req_d.name)
+ self.assertEqual(code, HTTP_OK)
+ self.assert_body(body)
+
+ def test_list(self):
+ self.create_d()
+ self.create_e()
+ code, body = self.get()
+ for case in body.testcases:
+ if self.req_d.name == case.name:
+ self.assert_body(case)
+ else:
+ self.assert_body(case, self.req_e)
+
+
+class TestCaseUpdate(TestCaseBase):
+ def test_noBody(self):
+ code, _ = self.update(case='noBody')
+ self.assertEqual(code, HTTP_BAD_REQUEST)
+
+ def test_notFound(self):
+ code, _ = self.update(self.update_e, 'notFound')
+ self.assertEqual(code, 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.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.assertIn("Nothing to update", body)
+
+ def test_success(self):
+ self.create_d()
+ code, body = self.get(self.req_d.name)
+ _id = body._id
+
+ code, body = self.update(self.update_e, self.req_d.name)
+ self.assertEqual(code, HTTP_OK)
+ self.assertEqual(_id, body._id)
+ self.assert_update_body(self.req_d, body, self.update_e)
+
+ _, new_body = self.get(self.req_e.name)
+ self.assertEqual(_id, new_body._id)
+ self.assert_update_body(self.req_d, new_body, self.update_e)
+
+
+class TestCaseDelete(TestCaseBase):
+ def test_notFound(self):
+ code, body = self.delete('notFound')
+ self.assertEqual(code, 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(body, '')
+
+ code, body = self.get(self.req_d.name)
+ self.assertEqual(code, HTTP_NOT_FOUND)
+
+
+if __name__ == '__main__':
+ unittest.main()
Code Review / releng.git / commitdiff