# 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
+# feng.xiaowei@zte.com.cn refactor db.pod to db.pods 5-19-2016
+# 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
##############################################################################
import json
from tornado import gen
from datetime import datetime, timedelta
-from models import TestProject, TestCase, TestResult
+from models import TestCase, TestResult, CreateResponse
+from resources.project_models import Project
from resources.pod_models import Pod
from common.constants import DEFAULT_REPRESENTATION, HTTP_BAD_REQUEST, \
HTTP_NOT_FOUND, HTTP_FORBIDDEN
get_dashboard_result
+def format_data(data, cls):
+ cls_data = cls.from_dict(data)
+ return cls_data.format_http()
+
+
class GenericApiHandler(RequestHandler):
"""
The purpose of this class is to take benefit of inheritance and prepare
self.db = self.settings["db"]
def prepare(self):
- if not (self.request.method == "GET"):
+ if not (self.request.method == "GET" or self.request.method == "DELETE"):
if self.request.headers.get("Content-Type") is not None:
if self.request.headers["Content-Type"].startswith(
DEFAULT_REPRESENTATION):
else:
self.json_args = None
- def finish_request(self, json_object):
- self.write(json.dumps(json_object))
+ def finish_request(self, json_object=None):
+ if json_object:
+ self.write(json.dumps(json_object))
self.set_header("Content-Type", DEFAULT_REPRESENTATION)
self.finish()
if pod_name is not None:
get_request["name"] = pod_name
-
- res = []
- cursor = self.db.pod.find(get_request)
- while (yield cursor.fetch_next):
- pod = Pod.pod_from_dict(cursor.next_object())
- res.append(pod.format())
-
- meta = dict()
- meta["total"] = len(res)
- meta["success"] = True if len(res) > 0 else False
-
- answer = dict()
- answer["pods"] = res
- answer["meta"] = meta
+ answer = yield self.db.pods.find_one(get_request)
+ if answer is None:
+ raise HTTPError(HTTP_NOT_FOUND,
+ "{} Not Exist".format(pod_name))
+ else:
+ answer = format_data(answer, Pod)
+ else:
+ res = []
+ cursor = self.db.pods.find(get_request)
+ while (yield cursor.fetch_next):
+ res.append(format_data(cursor.next_object(), Pod))
+ answer = dict()
+ 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.pod.find_one(query)
+ mongo_dict = yield self.db.pods.find_one(query)
if mongo_dict is not None:
raise HTTPError(HTTP_FORBIDDEN,
"{} already exists as a pod".format(
self.json_args.get("name")))
- pod = Pod.pod_from_dict(self.json_args)
+ pod = Pod.from_dict(self.json_args)
pod.creation_date = datetime.now()
- future = self.db.pod.insert(pod.format())
- result = yield future
- pod._id = result
+ yield self.db.pods.insert(pod.format())
- meta = dict()
- meta["success"] = True
- meta["uri"] = "/pods/{}".format(pod.name)
-
- answer = dict()
- answer["pod"] = pod.format_http()
- answer["meta"] = meta
-
- self.finish_request(answer)
+ res = CreateResponse(self.request.full_url() + '/{}'.format(pod.name))
+ self.finish_request(res.format())
@asynchronous
@gen.coroutine
""" Remove a POD
# check for an existing pod to be deleted
- mongo_dict = yield self.db.pod.find_one(
+ mongo_dict = yield self.db.pods.find_one(
{'name': pod_name})
pod = TestProject.pod(mongo_dict)
if pod is None:
.format(pod_name))
# just delete it, or maybe save it elsewhere in a future
- res = yield self.db.test_projects.remove(
+ res = yield self.db.projects.remove(
{'name': pod_name})
meta = dict()
:param project_name:
"""
- if project_name is None:
- project_name = ""
-
get_request = dict()
- if len(project_name) > 0:
+ if project_name is not None:
get_request["name"] = project_name
-
- res = []
- cursor = self.db.test_projects.find(get_request)
- while (yield cursor.fetch_next):
- test_project = TestProject.testproject_from_dict(
- cursor.next_object())
- res.append(test_project.format_http())
-
- meta = dict()
- meta["total"] = len(res)
- meta["success"] = True if len(res) > 0 else False
-
- answer = dict()
- answer["test_projects"] = res
- answer["meta"] = meta
+ answer = yield self.db.projects.find_one(get_request)
+ if answer is None:
+ raise HTTPError(HTTP_NOT_FOUND,
+ "{} Not Exist".format(project_name))
+ else:
+ answer = format_data(answer, Project)
+ else:
+ res = []
+ cursor = self.db.projects.find(get_request)
+ while (yield cursor.fetch_next):
+ res.append(format_data(cursor.next_object(), Project))
+ answer = dict()
+ answer['projects'] = res
self.finish_request(answer)
query = {"name": self.json_args.get("name")}
# check for name in db
- mongo_dict = yield self.db.test_projects.find_one(query)
+ mongo_dict = yield self.db.projects.find_one(query)
if mongo_dict is not None:
raise HTTPError(HTTP_FORBIDDEN,
"{} already exists as a project".format(
self.json_args.get("name")))
- test_project = TestProject.testproject_from_dict(self.json_args)
- test_project.creation_date = datetime.now()
+ project = Project.from_dict(self.json_args)
+ project.creation_date = datetime.now()
- future = self.db.test_projects.insert(test_project.format())
- result = yield future
- test_project._id = result
+ yield self.db.projects.insert(project.format())
+
+ res = CreateResponse(self.request.full_url() + '/{}'.format(project.name))
+ self.finish_request(res.format())
- self.finish_request(test_project.format_http())
@asynchronous
@gen.coroutine
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.test_projects.find_one(query)
- test_project = TestProject.testproject_from_dict(mongo_dict)
- if test_project is None:
+ mongo_dict = yield self.db.projects.find_one(query)
+ project = Project.from_dict(mongo_dict)
+ if project is None:
raise HTTPError(HTTP_NOT_FOUND,
"{} could not be found".format(project_name))
# check for payload name parameter in db
# avoid a request if the project name has not changed in the payload
- if new_name != test_project.name:
- mongo_dict = yield self.db.test_projects.find_one(
+ if new_name != project.name:
+ mongo_dict = yield self.db.projects.find_one(
{"name": new_name})
if mongo_dict is not None:
raise HTTPError(HTTP_FORBIDDEN,
request = prepare_put_request(request,
"name",
new_name,
- test_project.name)
+ project.name)
request = prepare_put_request(request,
"description",
new_description,
- test_project.description)
+ project.description)
""" raise exception if there isn't a change """
if not request:
- raise HTTPError(HTTP_FORBIDDEN,
- "Nothing to update")
+ raise HTTPError(HTTP_FORBIDDEN, "Nothing to update")
""" we merge the whole document """
- edit_request = test_project.format()
+ edit_request = project.format()
edit_request.update(request)
""" Updating the DB """
- res = yield self.db.test_projects.update({'name': project_name},
- edit_request)
- print res
- edit_request["_id"] = str(test_project._id)
+ yield self.db.projects.update({'name': project_name}, edit_request)
+ new_project = yield self.db.projects.find_one({"_id": project._id})
- self.finish_request({"message": "success", "content": edit_request})
+ self.finish_request(format_data(new_project, Project))
@asynchronous
@gen.coroutine
print "DELETE request for : {}".format(project_name)
# check for an existing project to be deleted
- mongo_dict = yield self.db.test_projects.find_one(
+ mongo_dict = yield self.db.projects.find_one(
{'name': project_name})
- test_project = TestProject.testproject_from_dict(mongo_dict)
+ test_project = Project.from_dict(mongo_dict)
if test_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
- res = yield self.db.test_projects.remove(
- {'name': project_name})
- print res
+ yield self.db.projects.remove({'name': project_name})
- self.finish_request({"message": "success"})
+ self.finish_request()
class TestCasesHandler(GenericApiHandler):
"Check your request payload")
# retrieve test project
- mongo_dict = yield self.db.test_projects.find_one(
+ mongo_dict = yield self.db.projects.find_one(
{"name": project_name})
if mongo_dict is None:
raise HTTPError(HTTP_FORBIDDEN,
"Could not find project {}"
.format(project_name))
- # test_project = TestProject.testproject_from_dict(self.json_args)
+ # test_project = TestProject.from_dict(self.json_args)
case = TestCase.test_case_from_dict(self.json_args)
case.project_name = project_name
# check for an existing case to be deleted
mongo_dict = yield self.db.test_cases.find_one(case_request)
- test_project = TestProject.testproject_from_dict(mongo_dict)
+ test_project = Project.from_dict(mongo_dict)
if test_project 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.test_projects.remove(case_request)
+ res = yield self.db.projects.remove(case_request)
print res
self.finish_request({"message": "success"})
# TODO : replace checks with jsonschema
# check for project
- mongo_dict = yield self.db.test_projects.find_one(
+ mongo_dict = yield self.db.projects.find_one(
{"name": self.json_args.get("project_name")})
if mongo_dict is None:
raise HTTPError(HTTP_NOT_FOUND,
.format(self.json_args.get("case_name")))
# check for pod
- mongo_dict = yield self.db.pod.find_one(
+ mongo_dict = yield self.db.pods.find_one(
{"name": self.json_args.get("pod_name")})
if mongo_dict is None:
raise HTTPError(HTTP_NOT_FOUND,
# are made available under the terms of the Apache License, Version 2.0\r
# which accompanies this distribution, and is available at\r
# http://www.apache.org/licenses/LICENSE-2.0\r
-# feng.xiaowei@zte.com.cn mv Pod to pod_models.py 6-18-2016\r
-# feng.xiaowei@zte.com.cn add MetaCreateResponse/MetaGetResponse 6-18-2016\r
+# feng.xiaowei@zte.com.cn mv Pod to pod_models.py 5-18-2016\r
+# feng.xiaowei@zte.com.cn add MetaCreateResponse/MetaGetResponse 5-18-2016\r
+# 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
##############################################################################\r
\r
\r
-class MetaCreateResponse(object):\r
- def __init__(self, success=True, uri=''):\r
- self.success = success\r
- self.uri = uri\r
+class CreateResponse(object):\r
+ def __init__(self, href=''):\r
+ self.href = href\r
\r
@staticmethod\r
- def from_dict(meta_dict):\r
- if meta_dict is None:\r
+ def from_dict(res_dict):\r
+ if res_dict is None:\r
return None\r
\r
- meta = MetaCreateResponse()\r
- meta.success = meta_dict.get('success')\r
- meta.uri = meta_dict.get('uri')\r
- return meta\r
-\r
-\r
-class MetaGetResponse(object):\r
- def __init__(self, success=True, total=0):\r
- self.success = success\r
- self.total = total\r
-\r
- @staticmethod\r
- def from_dict(meta_dict):\r
- if meta_dict is None:\r
- return None\r
-\r
- meta = MetaGetResponse()\r
- meta.success = meta_dict.get('success')\r
- meta.total = meta_dict.get('total')\r
- return meta\r
-\r
-\r
-class TestProject:\r
- """ Describes a test project"""\r
-\r
- def __init__(self):\r
- self._id = None\r
- self.name = None\r
- self.description = None\r
- self.creation_date = None\r
-\r
- @staticmethod\r
- def testproject_from_dict(testproject_dict):\r
-\r
- if testproject_dict is None:\r
- return None\r
-\r
- t = TestProject()\r
- t._id = testproject_dict.get('_id')\r
- t.creation_date = testproject_dict.get('creation_date')\r
- t.name = testproject_dict.get('name')\r
- t.description = testproject_dict.get('description')\r
-\r
- return t\r
+ res = CreateResponse()\r
+ res.href = res_dict.get('href')\r
+ return res\r
\r
def format(self):\r
- return {\r
- "name": self.name,\r
- "description": self.description,\r
- "creation_date": str(self.creation_date)\r
- }\r
-\r
- def format_http(self, test_cases=0):\r
- return {\r
- "_id": str(self._id),\r
- "name": self.name,\r
- "description": self.description,\r
- "creation_date": str(self.creation_date),\r
- }\r
+ return {'href': self.href}\r
\r
\r
class TestCase:\r
-from models import MetaCreateResponse, MetaGetResponse
+__author__ = '__serena__'
class PodCreateRequest(object):
"details": self.details,
}
- @staticmethod
- def from_dict(req_dict):
- if req_dict is None:
- return None
-
- req = PodCreateRequest()
- req.name = req_dict.get('name')
- req.mode = req_dict.get('mode')
- req.details = req_dict.get('details')
- return req
-
class Pod(PodCreateRequest):
""" describes a POD platform """
self.creation_date = create_date
@staticmethod
- def pod_from_dict(pod_dict):
+ def from_dict(pod_dict):
if pod_dict is None:
return None
return f
-class PodCreateResponse(object):
- def __init__(self, pod=None, meta=None):
- self.pod = pod
- self.meta = meta
-
- @staticmethod
- def from_dict(res_dict):
- if res_dict is None:
- return None
-
- res = PodCreateResponse()
- res.pod = Pod.pod_from_dict(res_dict.get('pod'))
- res.meta = MetaCreateResponse.from_dict(res_dict.get('meta'))
- return res
-
-
-class PodGetResponse(PodCreateRequest):
- def __init__(self, name='', mode='', details='', create_date=''):
- self.creation_date = create_date
- super(PodGetResponse, self).__init__(name, mode, details)
-
- @staticmethod
- def from_dict(req_dict):
- if req_dict is None:
- return None
-
- res = PodGetResponse()
- res.creation_date = str(req_dict.get('creation_date'))
- res.name = req_dict.get('name')
- res.mode = req_dict.get('mode')
- res.details = req_dict.get('details')
- return res
-
-
-class PodsGetResponse(object):
- def __init__(self, pods=[], meta=None):
+class Pods(object):
+ def __init__(self, pods=list()):
self.pods = pods
- self.meta = meta
@staticmethod
def from_dict(res_dict):
if res_dict is None:
return None
- res = PodsGetResponse()
+ res = Pods()
for pod in res_dict.get('pods'):
- res.pods.append(PodGetResponse.from_dict(pod))
- res.meta = MetaGetResponse.from_dict(res_dict.get('meta'))
+ res.pods.append(Pod.from_dict(pod))
return res
--- /dev/null
+__author__ = '__serena__'
+
+
+class ProjectCreateRequest(object):
+ def __init__(self, name='', description=''):
+ self.name = name
+ self.description = description
+
+ def format(self):
+ return {
+ "name": self.name,
+ "description": self.description,
+ }
+
+
+class Project:
+ """ Describes a test project"""
+
+ def __init__(self):
+ self._id = None
+ self.name = None
+ self.description = None
+ self.creation_date = None
+
+ @staticmethod
+ def from_dict(res_dict):
+
+ if res_dict is None:
+ return None
+
+ t = Project()
+ t._id = res_dict.get('_id')
+ t.creation_date = res_dict.get('creation_date')
+ t.name = res_dict.get('name')
+ t.description = res_dict.get('description')
+
+ return t
+
+ def format(self):
+ return {
+ "name": self.name,
+ "description": self.description,
+ "creation_date": str(self.creation_date)
+ }
+
+ def format_http(self):
+ return {
+ "_id": str(self._id),
+ "name": self.name,
+ "description": self.description,
+ "creation_date": str(self.creation_date),
+ }
+
+
+class Projects(object):
+ def __init__(self, projects=list()):
+ self.projects = projects
+
+ @staticmethod
+ def from_dict(res_dict):
+ if res_dict is None:
+ return None
+
+ res = Projects()
+ for project in res_dict.get('projects'):
+ res.projects.append(Project.from_dict(project))
+ return res
- json args validation with schemes
- POST/PUT/DELETE for PODs
- POST/PUT/GET/DELETE for installers, platforms (enrich results info)
- - count cases for GET on test_projects
+ - count cases for GET on projects
- count results for GET on cases
- include objects
- swagger documentation
(r"/pods/([^/]+)", PodHandler),
# few examples:
- # GET /test_projects
- # GET /test_projects/yardstick
- (r"/test_projects", TestProjectHandler),
- (r"/test_projects/([^/]+)", TestProjectHandler),
+ # GET /projects
+ # GET /projects/yardstick
+ (r"/projects", TestProjectHandler),
+ (r"/projects/([^/]+)", TestProjectHandler),
# few examples
- # GET /test_projects/qtip/cases => Get cases for qtip
+ # GET /projects/qtip/cases => Get cases for qtip
#
- (r"/test_projects/([^/]+)/cases", TestCasesHandler),
- (r"/test_projects/([^/]+)/cases/([^/]+)", TestCasesHandler),
+ (r"/projects/([^/]+)/cases", TestCasesHandler),
+ (r"/projects/([^/]+)/cases/([^/]+)", TestCasesHandler),
# (r"/test_cases/([^/]+)", TestCasesHandler),
# new path to avoid a long depth
{
"id": "1183979d-1847-0238-4abe-12406c10d3e5",
"headers": "",
- "url": "http://127.0.0.1:8000/test_projects/yardstick/cases",
+ "url": "http://127.0.0.1:8000/projects/yardstick/cases",
"preRequestScript": "",
"pathVariables": {},
"method": "GET",
{
"id": "2ea6a7a9-4d73-e78a-4e9a-21df9d72dc4d",
"headers": "",
- "url": "http://127.0.0.1:8000/test_projects/project_foo/cases",
+ "url": "http://127.0.0.1:8000/projects/project_foo/cases",
"preRequestScript": "",
"pathVariables": {},
"method": "GET",
{
"id": "3077838a-4ea1-f35e-5c05-2b1aba7d5b91",
"headers": "",
- "url": "http://213.77.62.197/test_projects/yardstick/cases",
+ "url": "http://213.77.62.197/projects/yardstick/cases",
"preRequestScript": "",
"pathVariables": {},
"method": "GET",
{
"id": "46347e52-c68c-bae8-f64f-a35de3b99c6f",
"headers": "Content-Type: application/json\n",
- "url": "http://127.0.0.1:8000/test_projects/project_foo/cases",
+ "url": "http://127.0.0.1:8000/projects/project_foo/cases",
"preRequestScript": "",
"pathVariables": {},
"method": "POST",
{
"id": "4c7651ea-4395-de8f-ce09-3b182cff1fd4",
"headers": "Content-Type: application/json\n",
- "url": "http://127.0.0.1:8000/test_projects/project_foo/cases",
+ "url": "http://127.0.0.1:8000/projects/project_foo/cases",
"preRequestScript": "",
"pathVariables": {},
"method": "POST",
{
"id": "4cb44e62-8f36-62c4-2f9e-c51019f97535",
"headers": "",
- "url": "http://213.77.62.197/test_projects/vsperf/cases",
+ "url": "http://213.77.62.197/projects/vsperf/cases",
"preRequestScript": "",
"pathVariables": {},
"method": "GET",
{
"id": "7396c965-4732-7ae4-398e-b0d3bc7fe937",
"headers": "",
- "url": "http://127.0.0.1:8000/test_projects/foo/cases",
+ "url": "http://127.0.0.1:8000/projects/foo/cases",
"preRequestScript": "",
"pathVariables": {},
"method": "GET",
{
"id": "78fc1ca8-6179-ffe0-4d10-c345dad994f6",
"headers": "Content-Type: application/json\n",
- "url": "http://127.0.0.1:8000/test_projects/project_foo",
+ "url": "http://127.0.0.1:8000/projects/project_foo",
"preRequestScript": "",
"pathVariables": {},
"method": "DELETE",
{
"id": "8c925a09-4c8a-b811-d25c-be2bb2592c8f",
"headers": "",
- "url": "http://213.77.62.197/test_projects",
+ "url": "http://213.77.62.197/projects",
"preRequestScript": "",
"pathVariables": {},
"method": "GET",
{
"id": "93f87cf3-4b1f-e3b2-9726-94f8fc259ee1",
"headers": "",
- "url": "http://127.0.0.1:8000/test_projects/project_foo/cases",
+ "url": "http://127.0.0.1:8000/projects/project_foo/cases",
"preRequestScript": "",
"pathVariables": {},
"method": "GET",
{
"id": "99c8fa51-df1c-cae5-09c7-d554a4c8b8a0",
"headers": "",
- "url": "http://213.77.62.197/test_projects/foo/cases",
+ "url": "http://213.77.62.197/projects/foo/cases",
"preRequestScript": "",
"pathVariables": {},
"method": "GET",
{
"id": "acac56bf-af93-bb07-774f-11667ea71a8b",
"headers": "",
- "url": "http://213.77.62.197/test_projects/functest/cases",
+ "url": "http://213.77.62.197/projects/functest/cases",
"preRequestScript": "",
"pathVariables": {},
"method": "GET",
{
"id": "c7f8ccfb-ecdd-6d5a-d9d6-3c097dcce62c",
"headers": "",
- "url": "http://127.0.0.1:8000/test_projects",
+ "url": "http://127.0.0.1:8000/projects",
"preRequestScript": "",
"pathVariables": {},
"method": "GET",
{
"id": "c820f80b-cdc8-602d-d9f3-32df11c861ef",
"headers": "",
- "url": "http://127.0.0.1:8000/test_projects/vsperf/cases",
+ "url": "http://127.0.0.1:8000/projects/vsperf/cases",
"preRequestScript": "",
"pathVariables": {},
"method": "GET",
{
"id": "d058c0d9-6c8f-f480-2cdd-f83a91430114",
"headers": "Content-Type: application/json\n",
- "url": "http://127.0.0.1:8000/test_projects",
+ "url": "http://127.0.0.1:8000/projects",
"preRequestScript": "",
"pathVariables": {},
"method": "POST",
{
"id": "d1c53d34-be0e-8eda-31f5-9c9386e892c3",
"headers": "Content-Type: application/json\n",
- "url": "http://127.0.0.1:8000/test_projects/project_foo",
+ "url": "http://127.0.0.1:8000/projects/project_foo",
"preRequestScript": "",
"pathVariables": {},
"method": "PUT",
def clear(self):
self._remove()
-pod = MemDb()
-test_projects = MemDb()
+pods = MemDb()
+projects = MemDb()
test_cases = MemDb()
test_results = MemDb()
from resources.handlers import VersionHandler, PodHandler, \
TestProjectHandler, TestCasesHandler, TestResultsHandler, DashboardHandler
+from resources.models import CreateResponse
import fake_pymongo
headers = {'Content-Type': 'application/json; charset=UTF-8'}
def setUp(self):
+ self.basePath = ''
+ self.create_res = CreateResponse
+ self.get_res = None
+ self.list_res = None
+ self.update_res = None
+ self.req_d = None
+ self.req_e = None
self.addCleanup(self._clear)
super(TestBase, self).setUp()
(r"/version", VersionHandler),
(r"/pods", PodHandler),
(r"/pods/([^/]+)", PodHandler),
- (r"/test_projects", TestProjectHandler),
- (r"/test_projects/([^/]+)", TestProjectHandler),
- (r"/test_projects/([^/]+)/cases", TestCasesHandler),
- (r"/test_projects/([^/]+)/cases/([^/]+)", TestCasesHandler),
+ (r"/projects", TestProjectHandler),
+ (r"/projects/([^/]+)", TestProjectHandler),
+ (r"/projects/([^/]+)/cases", TestCasesHandler),
+ (r"/projects/([^/]+)/cases/([^/]+)", TestCasesHandler),
(r"/results", TestResultsHandler),
(r"/results([^/]*)", TestResultsHandler),
(r"/results/([^/]*)", TestResultsHandler),
debug=True,
)
- def create(self, uri, body=None):
- return self.fetch(uri,
- method='POST',
- body=json.dumps(body),
- headers=self.headers)
+ def create_d(self):
+ return self.create(self.req_d)
- def get(self, uri):
- return self.fetch(uri,
- method='GET',
- headers=self.headers)
+ def create_e(self):
+ return self.create(self.req_e)
+
+ def create(self, req=None):
+ if req:
+ req = req.format()
+
+ res = self.fetch(self.basePath,
+ 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),
+ method='GET',
+ headers=self.headers)
+
+ def inner():
+ return self.get_res if item else self.list_res
+ return self._get_return(res, inner())
+
+ def update(self, item, new=None):
+ if new:
+ new = new.format()
+ res = self.fetch(self._get_uri(item),
+ 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),
+ method='DELETE',
+ headers=self.headers)
+ return res.code
+
+ def _get_uri(self, item=None):
+ uri = self.basePath
+ if item:
+ uri += '/{}'.format(item)
+ return uri
+
+ def _get_return(self, res, cls):
+ code = res.code
+ body = res.body
+ return code, self._get_return_body(code, body, cls)
+
+ @staticmethod
+ 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)
+ if not req:
+ req = self.req_d
+ self.assertIn(self._get_uri(req.name), body.href)
@staticmethod
def _clear():
- fake_pymongo.pod.clear()
- fake_pymongo.test_projects.clear()
+ fake_pymongo.pods.clear()
+ fake_pymongo.projects.clear()
fake_pymongo.test_cases.clear()
fake_pymongo.test_results.clear()
def fixture_setup(self):
self.test1 = {'_id': '1', 'name': 'test1'}
self.test2 = {'name': 'test2'}
- yield self.db.pod.insert({'_id': '1', 'name': 'test1'})
- yield self.db.pod.insert({'name': 'test2'})
+ yield self.db.pods.insert({'_id': '1', 'name': 'test1'})
+ yield self.db.pods.insert({'name': 'test2'})
@gen_test
def test_find_one(self):
- user = yield self.db.pod.find_one({'name': 'test1'})
+ user = yield self.db.pods.find_one({'name': 'test1'})
self.assertEqual(user, self.test1)
@gen_test
def test_find(self):
- cursor = self.db.pod.find()
+ cursor = self.db.pods.find()
names = []
while (yield cursor.fetch_next):
ob = cursor.next_object()
@gen_test
def test_update(self):
- yield self.db.pod.update({'_id': '1'}, {'name': 'new_test1'})
- user = yield self.db.pod.find_one({'_id': '1'})
+ yield self.db.pods.update({'_id': '1'}, {'name': 'new_test1'})
+ user = yield self.db.pods.find_one({'_id': '1'})
self.assertEqual(user.get('name', None), 'new_test1')
@gen_test
def test_remove(self):
- yield self.db.pod.remove({'_id': '1'})
- user = yield self.db.pod.find_one({'_id': '1'})
+ yield self.db.pods.remove({'_id': '1'})
+ user = yield self.db.pods.find_one({'_id': '1'})
self.assertIsNone(user)
if __name__ == '__main__':
import unittest
-import json
from test_base import TestBase
-from resources.pod_models import PodCreateRequest, \
- PodCreateResponse, PodsGetResponse
-from common.constants import HTTP_OK, HTTP_BAD_REQUEST, HTTP_FORBIDDEN
-
-
-class TestPodCreate(TestBase):
- req = PodCreateRequest(name='zte-1', mode='alive', details='zte pod 1')
-
+from resources.pod_models import PodCreateRequest, Pod, Pods
+from common.constants import HTTP_OK, HTTP_BAD_REQUEST, \
+ HTTP_FORBIDDEN, HTTP_NOT_FOUND
+
+
+class TestPodBase(TestBase):
+ def setUp(self):
+ super(TestPodBase, self).setUp()
+ self.req_d = PodCreateRequest('zte-1', 'fuel', 'zte pod 1')
+ self.req_e = PodCreateRequest('zte-2', 'apex', 'zte pod 2')
+ self.get_res = Pod
+ self.list_res = Pods
+ self.basePath = '/pods'
+
+ def assert_get_body(self, pod, req=None):
+ if not req:
+ req = self.req_d
+ self.assertEqual(pod.name, req.name)
+ self.assertEqual(pod.mode, req.mode)
+ self.assertEqual(pod.details, req.details)
+ self.assertIsNotNone(pod.creation_date)
+ self.assertIsNotNone(pod._id)
+
+
+class TestPodCreate(TestPodBase):
def test_withoutBody(self):
- res = self.create('/pods', body=None)
- self.assertEqual(res.code, HTTP_BAD_REQUEST)
+ (code, body) = self.create()
+ self.assertEqual(code, HTTP_BAD_REQUEST)
def test_success(self):
- res = self.create('/pods', body=self.req.format())
- self.assertEqual(res.code, HTTP_OK)
- res_body = PodCreateResponse.from_dict(json.loads(res.body))
- self._assertMeta(res_body.meta, True)
- self._assertBody(res_body.pod)
+ code, body = self.create_d()
+ self.assertEqual(code, HTTP_OK)
+ self.assert_create_body(body)
def test_alreadyExist(self):
- self.create('/pods', body=self.req.format())
- res = self.create('/pods', body=self.req.format())
- self.assertEqual(res.code, HTTP_FORBIDDEN)
- self.assertIn('already exists', res.body)
+ self.create_d()
+ code, body = self.create_d()
+ self.assertEqual(code, HTTP_FORBIDDEN)
+ self.assertIn('already exists', body)
def _assertMeta(self, meta, success):
self.assertEqual(meta.success, success)
if success:
- self.assertEqual(meta.uri, '/pods/{}'.format(self.req.name))
-
- def _assertBody(self, res):
- self.assertEqual(res.name, self.req.name)
- self.assertEqual(res.mode, self.req.mode)
- self.assertEqual(res.details, self.req.details)
- self.assertIsNotNone(res.creation_date)
- self.assertIsNotNone(res._id)
+ self.assertEqual(meta.uri, '/pods/{}'.format(self.req_d.name))
-class TestPodGet(TestBase):
+class TestPodGet(TestPodBase):
def test_notExist(self):
- res = self.get('/pods/notExist')
- body = PodsGetResponse.from_dict(json.loads(res.body))
- self._assertMeta(body.meta, 0)
+ code, body = self.get('notExist')
+ self.assertEqual(code, HTTP_NOT_FOUND)
def test_getOne(self):
- self.create('/pods', body=TestPodCreate.req.format())
- res = self.get('/pods/{}'.format(TestPodCreate.req.name))
- body = PodsGetResponse.from_dict(json.loads(res.body))
- self._assertMeta(body.meta, 1)
- self._assertBody(TestPodCreate.req, body.pods[0])
+ self.create_d()
+ code, body = self.get(self.req_d.name)
+ self.assert_get_body(body)
def test_list(self):
- req = PodCreateRequest(name='zte-2', mode='alive', details='zte pod 2')
- self.create('/pods', body=TestPodCreate.req.format())
- self.create('/pods', body=req.format())
- res = self.get('/pods')
- body = PodsGetResponse.from_dict(json.loads(res.body))
- self._assertMeta(body.meta, 2)
+ self.create_d()
+ self.create_e()
+ code, body = self.get()
+ self.assertEqual(len(body.pods), 2)
for pod in body.pods:
- if req.name == pod.name:
- self._assertBody(req, pod)
+ if self.req_d.name == pod.name:
+ self.assert_get_body(pod)
else:
- self._assertBody(TestPodCreate.req, pod)
-
- def _assertMeta(self, meta, total):
- def check_success():
- if total is 0:
- return False
- else:
- return True
- self.assertEqual(meta.total, total)
- self.assertEqual(meta.success, check_success())
-
- def _assertBody(self, req, res):
- self.assertEqual(res.name, req.name)
- self.assertEqual(res.mode, req.mode)
- self.assertEqual(res.details, req.details)
- self.assertIsNotNone(res.creation_date)
-
+ self.assert_get_body(pod, self.req_e)
if __name__ == '__main__':
unittest.main()
--- /dev/null
+import unittest
+
+from test_base import TestBase
+from resources.project_models import ProjectCreateRequest, Project, Projects
+from common.constants import HTTP_OK, HTTP_BAD_REQUEST, \
+ HTTP_FORBIDDEN, HTTP_NOT_FOUND
+
+
+class TestProjectBase(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.basePath = '/projects'
+
+ def assert_body(self, project, req=None):
+ if not req:
+ req = self.req_d
+ self.assertEqual(project.name, req.name)
+ self.assertEqual(project.description, req.description)
+ self.assertIsNotNone(project._id)
+ self.assertIsNotNone(project.creation_date)
+
+
+class TestProjectCreate(TestProjectBase):
+ def test_withoutBody(self):
+ (code, body) = self.create()
+ self.assertEqual(code, HTTP_BAD_REQUEST)
+
+ def test_success(self):
+ (code, body) = self.create_d()
+ self.assertEqual(code, 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.assertIn('already exists', body)
+
+
+class TestProjectGet(TestProjectBase):
+ 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 project in body.projects:
+ if self.req_d.name == project.name:
+ self.assert_body(project)
+ else:
+ self.assert_body(project, self.req_e)
+
+
+class TestProjectUpdate(TestProjectBase):
+ def test_withoutBody(self):
+ code, _ = self.update('noBody')
+ self.assertEqual(code, HTTP_BAD_REQUEST)
+
+ def test_notFound(self):
+ code, _ = self.update('notFound', self.req_e)
+ 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)
+ 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)
+ 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
+
+ req = ProjectCreateRequest('newName', 'new description')
+ code, body = self.update(self.req_d.name, req)
+ self.assertEqual(code, HTTP_OK)
+ self.assertEqual(_id, body._id)
+ self.assert_body(body, req)
+
+ _, new_body = self.get(req.name)
+ self.assertEqual(_id, new_body._id)
+ self.assert_body(new_body, req)
+
+
+class TestProjectDelete(TestProjectBase):
+ def test_notFound(self):
+ code = self.delete('notFound')
+ self.assertEqual(code, HTTP_NOT_FOUND)
+
+ def test_success(self):
+ self.create_d()
+ code = self.delete(self.req_d.name)
+ self.assertEqual(code, HTTP_OK)
+
+if __name__ == '__main__':
+ unittest.main()
Code Review / releng.git / commitdiff