self.db_testcases = 'testcases'
self.db_results = 'results'
self.db_scenarios = 'scenarios'
+ self.db_deployresults = 'deployresults'
self.auth = self.settings["auth"]
def prepare(self):
if k != 'query':
data.__setattr__(k, v)
- if self.table != 'results':
+ if 'results' not in self.table:
data.creation_date = datetime.now()
_id = yield dbapi.db_save(self.table, data.format())
if 'name' in self.json_args:
--- /dev/null
+from opnfv_testapi.handlers import result_handlers
+from opnfv_testapi.models import deploy_result_models
+from opnfv_testapi.tornado_swagger import swagger
+
+
+class GenericDeployResultHandler(result_handlers.GenericResultHandler):
+ def __init__(self, application, request, **kwargs):
+ super(GenericDeployResultHandler, self).__init__(application,
+ request,
+ **kwargs)
+ self.table = self.db_deployresults
+ self.table_cls = deploy_result_models.DeployResult
+
+
+class DeployResultsHandler(GenericDeployResultHandler):
+ @swagger.operation(nickname="queryDeployResults")
+ def get(self):
+ """
+ @description: Retrieve deployment result(s).
+ @notes: Retrieve deployment result(s).
+ Available filters for this request are :
+ - installer : fuel/apex/compass/joid/daisy
+ - version : platform version (Arno-R1, ...)
+ - pod_name : pod name
+ - job_name : jenkins job name
+ - build_id : Jenkins build id
+ - scenario : the test scenario
+ - period : x last days, incompatible with from/to
+ - from : starting time in 2016-01-01 or 2016-01-01 00:01:23
+ - to : ending time in 2016-01-01 or 2016-01-01 00:01:23
+ - criteria : the global criteria status passed or failed
+ - page : which page to list, default to 1
+ - descend : true, newest2oldest; false, oldest2newest
+
+ GET /deployresults/installer=daisy&version=master \
+ &pod_name=pod_name&period=15
+ @return 200: all deployment results consist with query,
+ empty list if no result is found
+ @rtype: L{DeployResults}
+ @param installer: installer name
+ @type installer: L{string}
+ @in installer: query
+ @required installer: False
+ @param version: version name
+ @type version: L{string}
+ @in version: query
+ @required version: False
+ @param pod_name: pod name
+ @type pod_name: L{string}
+ @in pod_name: query
+ @required pod_name: False
+ @param job_name: jenkins job name
+ @type job_name: L{string}
+ @in job_name: query
+ @required job_name: False
+ @param build_id: jenkins build_id
+ @type build_id: L{string}
+ @in build_id: query
+ @required build_id: False
+ @param scenario: i.e. odl
+ @type scenario: L{string}
+ @in scenario: query
+ @required scenario: False
+ @param criteria: i.e. PASS/FAIL
+ @type criteria: L{string}
+ @in criteria: query
+ @required criteria: False
+ @param period: last days
+ @type period: L{string}
+ @in period: query
+ @required period: False
+ @param from: i.e. 2016-01-01 or 2016-01-01 00:01:23
+ @type from: L{string}
+ @in from: query
+ @required from: False
+ @param to: i.e. 2016-01-01 or 2016-01-01 00:01:23
+ @type to: L{string}
+ @in to: query
+ @required to: False
+ @param page: which page to list, default to 1
+ @type page: L{int}
+ @in page: query
+ @required page: False
+ @param descend: true, newest2oldest; false, oldest2newest
+ @type descend: L{string}
+ @in descend: query
+ @required descend: False
+ """
+ super(DeployResultsHandler, self).get()
+
+ @swagger.operation(nickname="createDeployResult")
+ def post(self):
+ """
+ @description: create a deployment result
+ @param body: deployment result to be created
+ @type body: L{DeployResultCreateRequest}
+ @in body: body
+ @rtype: L{CreateResponse}
+ @return 200: deploy result is created.
+ @raise 404: pod not exist
+ @raise 400: body or some field is not provided
+ """
+ def pod_query():
+ return {'name': self.json_args.get('pod_name')}
+
+ def options_check(field, options):
+ return self.json_args.get(field).upper() in options
+
+ miss_fields = ['pod_name', 'installer', 'scenario']
+ carriers = [('pods', pod_query)]
+ values_check = [('criteria', options_check, ['PASS', 'FAIL'])]
+
+ self._create(miss_fields=miss_fields,
+ carriers=carriers,
+ values_check=values_check)
date_range.update({'$gte': str(v)})
elif k == 'to':
date_range.update({'$lt': str(v)})
+ elif 'build_id' in k:
+ query[k] = self.get_int(k, v)
elif k == 'signed':
username = self.get_secure_cookie(constants.TESTAPI_ID)
role = self.get_secure_cookie(constants.ROLE)
return query
+ def get(self):
+ def descend_limit():
+ descend = self.get_query_argument('descend', 'true')
+ return -1 if descend.lower() == 'true' else 1
+
+ def last_limit():
+ return self.get_int('last', self.get_query_argument('last', 0))
+
+ def page_limit():
+ return self.get_int('page', self.get_query_argument('page', 1))
+
+ limitations = {
+ 'sort': {'_id': descend_limit()},
+ 'last': last_limit(),
+ 'page': page_limit(),
+ 'per_page': CONF.api_results_per_page
+ }
+
+ self._list(query=self.set_query(), **limitations)
+
class ResultsCLHandler(GenericResultHandler):
@swagger.operation(nickname="queryTestResults")
@in descend: query
@required descend: False
"""
- def descend_limit():
- descend = self.get_query_argument('descend', 'true')
- return -1 if descend.lower() == 'true' else 1
-
- def last_limit():
- return self.get_int('last', self.get_query_argument('last', 0))
-
- def page_limit():
- return self.get_int('page', self.get_query_argument('page', 1))
-
- limitations = {
- 'sort': {'_id': descend_limit()},
- 'last': last_limit(),
- 'page': page_limit(),
- 'per_page': CONF.api_results_per_page
- }
-
- self._list(query=self.set_query(), **limitations)
+ super(ResultsCLHandler, self).get()
@swagger.operation(nickname="createTestResult")
def post(self):
@raise 404: pod/project/testcase not exist
@raise 400: body/pod_name/project_name/case_name not provided
"""
- self._post()
-
- def _post(self):
def pod_query():
return {'name': self.json_args.get('pod_name')}
if openid:
self.json_args['user'] = openid
- super(ResultsUploadHandler, self)._post()
+ super(ResultsUploadHandler, self).post()
class ResultsGURHandler(GenericResultHandler):
class ModelBase(object):
+ def __eq__(self, other):
+ res = all(getattr(self, k) == getattr(other, k)
+ for k in self.format().keys() if k != '_id')
+ return res
+
def format(self):
return self._format(['_id'])
--- /dev/null
+from opnfv_testapi.models import base_models
+from opnfv_testapi.tornado_swagger import swagger
+
+
+@swagger.model()
+class DeployResultCreateRequest(base_models.ModelBase):
+ def __init__(self,
+ installer=None,
+ version=None,
+ pod_name=None,
+ job_name=None,
+ build_id=None,
+ scenario=None,
+ upstream_job_name=None,
+ upstream_build_id=None,
+ criteria=None,
+ start_date=None,
+ stop_date=None,
+ details=None):
+ self.installer = installer
+ self.version = version
+ self.pod_name = pod_name
+ self.job_name = job_name
+ self.build_id = build_id
+ self.scenario = scenario
+ self.upstream_job_name = upstream_job_name
+ self.upstream_build_id = upstream_build_id
+ self.criteria = criteria
+ self.start_date = start_date
+ self.stop_date = stop_date
+ self.details = details
+
+
+@swagger.model()
+class DeployResult(DeployResultCreateRequest):
+ def __init__(self,
+ _id=None, **kwargs):
+ self._id = _id
+ super(DeployResult, self).__init__(**kwargs)
+
+
+@swagger.model()
+class DeployResults(base_models.ModelBase):
+ """
+ @property deployresults:
+ @ptype deployresults: C{list} of L{DeployResult}
+ """
+ def __init__(self):
+ self.deployresults = list()
+
+ @staticmethod
+ def attr_parser():
+ return {'deployresults': DeployResult}
from opnfv_testapi.common.config import CONF
from opnfv_testapi.handlers import base_handlers
+from opnfv_testapi.handlers import deploy_result_handlers as deploy_handlers
from opnfv_testapi.handlers import pod_handlers
from opnfv_testapi.handlers import project_handlers
from opnfv_testapi.handlers import result_handlers
(r"/api/v1/results", result_handlers.ResultsCLHandler),
(r'/api/v1/results/upload', result_handlers.ResultsUploadHandler),
(r"/api/v1/results/([^/]+)", result_handlers.ResultsGURHandler),
+ (r"/api/v1/deployresults", deploy_handlers.DeployResultsHandler),
# scenarios
(r"/api/v1/scenarios", scenario_handlers.ScenariosCLHandler),
projects = MemDb('projects')
testcases = MemDb('testcases')
results = MemDb('results')
+deployresults = MemDb('deployresults')
scenarios = MemDb('scenarios')
tokens = MemDb('tokens')
users = MemDb('users')
self.config_patcher.start()
self.db_patcher.start()
+ @staticmethod
+ def load_json(json_file):
+ abs_file = path.join(path.dirname(__file__),
+ '../templates',
+ json_file + '.json')
+ with open(abs_file, 'r') as f:
+ loader = json.load(f)
+ f.close()
+ return loader
+
def get_app(self):
from opnfv_testapi.cmd import server
return server.make_app()
fake_pymongo.projects.clear()
fake_pymongo.testcases.clear()
fake_pymongo.results.clear()
+ fake_pymongo.deployresults.clear()
fake_pymongo.scenarios.clear()
--- /dev/null
+##############################################################################
+# Copyright (c) 2016 ZTE Corporation
+# feng.xiaowei@zte.com.cn
+# All rights reserved. This program and the accompanying materials
+# are made available under the terms of the Apache License, Version 2.0
+# which accompanies this distribution, and is available at
+# http://www.apache.org/licenses/LICENSE-2.0
+##############################################################################
+import copy
+from datetime import datetime
+from datetime import timedelta
+import httplib
+import urllib
+
+from opnfv_testapi.common import message
+from opnfv_testapi.models import deploy_result_models as drm
+from opnfv_testapi.tests.unit import executor
+from opnfv_testapi.tests.unit import fake_pymongo
+from opnfv_testapi.tests.unit.handlers import test_base as base
+
+
+class DeployResultBase(base.TestBase):
+ @executor.mock_valid_lfid()
+ def setUp(self):
+ super(DeployResultBase, self).setUp()
+ self.req_d = drm.DeployResultCreateRequest.from_dict(
+ self.load_json('deploy_result'))
+ self.req_d.start_date = str(datetime.now())
+ self.req_d.stop_date = str(datetime.now() + timedelta(minutes=1))
+ self.get_res = drm.DeployResult
+ self.list_res = drm.DeployResults
+ self.basePath = '/api/v1/deployresults'
+ fake_pymongo.pods.insert(self.pod_d.format())
+
+ def assert_res(self, deploy_result, req=None):
+ if req is None:
+ req = self.req_d
+ print req.format()
+ self.assertEqual(deploy_result, req)
+ self.assertIsNotNone(deploy_result._id)
+
+ def _create_d(self):
+ _, res = self.create_d()
+ return res.href.split('/')[-1]
+
+
+class DeployResultCreate(DeployResultBase):
+ @executor.create(httplib.BAD_REQUEST, message.no_body())
+ def test_nobody(self):
+ return None
+
+ @executor.create(httplib.BAD_REQUEST, message.missing('pod_name'))
+ def test_podNotProvided(self):
+ req = self.req_d
+ req.pod_name = None
+ return req
+
+ @executor.create(httplib.BAD_REQUEST, message.missing('installer'))
+ def test_installerNotProvided(self):
+ req = self.req_d
+ req.installer = None
+ return req
+
+ @executor.create(httplib.BAD_REQUEST, message.missing('scenario'))
+ def test_testcaseNotProvided(self):
+ req = self.req_d
+ req.scenario = None
+ return req
+
+ @executor.create(httplib.BAD_REQUEST,
+ message.invalid_value('criteria', ['PASS', 'FAIL']))
+ def test_invalid_criteria(self):
+ req = self.req_d
+ req.criteria = 'invalid'
+ return req
+
+ @executor.create(httplib.FORBIDDEN, message.not_found_base)
+ def test_noPod(self):
+ req = self.req_d
+ req.pod_name = 'notExistPod'
+ return req
+
+ @executor.create(httplib.OK, 'assert_href')
+ def test_success(self):
+ return self.req_d
+
+
+class DeployResultGet(DeployResultBase):
+ def setUp(self):
+ super(DeployResultGet, self).setUp()
+ self.req_10d_before = self._create_changed_date(days=-10)
+ self.req_d_id = self._create_d()
+ self.req_10d_later = self._create_changed_date(days=10)
+
+ @executor.query(httplib.OK, '_query_success', 3)
+ def test_queryInstaller(self):
+ return self._set_query('installer')
+
+ @executor.query(httplib.OK, '_query_success', 3)
+ def test_queryVersion(self):
+ return self._set_query('version')
+
+ @executor.query(httplib.OK, '_query_success', 3)
+ def test_queryPod(self):
+ return self._set_query('pod_name')
+
+ @executor.query(httplib.OK, '_query_success', 3)
+ def test_queryJob(self):
+ return self._set_query('job_name')
+
+ @executor.query(httplib.OK, '_query_success', 1)
+ def test_queryBuildId(self):
+ return self._set_query('build_id')
+
+ @executor.query(httplib.OK, '_query_success', 3)
+ def test_queryScenario(self):
+ return self._set_query('scenario')
+
+ @executor.query(httplib.OK, '_query_success', 3)
+ def test_queryUpstreamJobName(self):
+ return self._set_query('upstream_job_name')
+
+ @executor.query(httplib.OK, '_query_success', 1)
+ def test_queryUpstreamBuildId(self):
+ return self._set_query('upstream_build_id')
+
+ @executor.query(httplib.OK, '_query_success', 3)
+ def test_queryCriteria(self):
+ return self._set_query('criteria')
+
+ @executor.query(httplib.BAD_REQUEST, message.must_int('period'))
+ def test_queryPeriodNotInt(self):
+ return self._set_query(period='a')
+
+ @executor.query(httplib.OK, '_query_period_one', 1)
+ def test_queryPeriodSuccess(self):
+ return self._set_query(period=5)
+
+ @executor.query(httplib.BAD_REQUEST, message.must_int('last'))
+ def test_queryLastNotInt(self):
+ return self._set_query(last='a')
+
+ @executor.query(httplib.OK, '_query_last_one', 1)
+ def test_queryLast(self):
+ return self._set_query(last=1)
+
+ @executor.query(httplib.OK, '_query_period_one', 1)
+ def test_combination(self):
+ return self._set_query('installer',
+ 'version',
+ 'pod_name',
+ 'job_name',
+ 'build_id',
+ 'scenario',
+ 'upstream_job_name',
+ 'upstream_build_id',
+ 'criteria',
+ period=5)
+
+ @executor.query(httplib.OK, '_query_success', 1)
+ def test_filterErrorStartdate(self):
+ self._create_error_start_date(None)
+ self._create_error_start_date('None')
+ self._create_error_start_date('null')
+ self._create_error_start_date('')
+ return self._set_query(period=5)
+
+ def _query_success(self, body, number):
+ self.assertEqual(number, len(body.deployresults))
+
+ def _query_last_one(self, body, number):
+ self.assertEqual(number, len(body.deployresults))
+ self.assert_res(body.deployresults[0], self.req_10d_later)
+
+ def _query_period_one(self, body, number):
+ self.assertEqual(number, len(body.deployresults))
+ self.assert_res(body.deployresults[0], self.req_d)
+
+ def _create_error_start_date(self, start_date):
+ req = copy.deepcopy(self.req_d)
+ req.start_date = start_date
+ self.create(req)
+ return req
+
+ def _create_changed_date(self, **kwargs):
+ req = copy.deepcopy(self.req_d)
+ req.build_id = req.build_id + kwargs.get('days')
+ req.upstream_build_id = req.upstream_build_id + kwargs.get('days')
+ req.start_date = datetime.now() + timedelta(**kwargs)
+ req.stop_date = str(req.start_date + timedelta(minutes=10))
+ req.start_date = str(req.start_date)
+ self.create(req)
+ return req
+
+ def _set_query(self, *args, **kwargs):
+ query = []
+ for arg in args:
+ query.append((arg, getattr(self.req_d, arg)))
+ for k, v in kwargs.iteritems():
+ query.append((k, v))
+ return urllib.urlencode(query)
--- /dev/null
+{
+ "build_id": 100,
+ "scenario": "os-nosdn-nofeature-ha",
+ "stop_date": "",
+ "start_date": "",
+ "upstream_job_name": "daisy-job-master",
+ "version": "master",
+ "pod_name": "zte-pod1",
+ "criteria": "PASS",
+ "installer": "daisy",
+ "upstream_build_id": 100,
+ "job_name": "daisy-deploy-job-master",
+ "details": ""
+}
\ No newline at end of file
Code Review / releng-testresults.git / commitdiff