##############################################################################
from datetime import datetime
from datetime import timedelta
-import json
-import logging
from bson import objectid
-from opnfv_testapi.common import constants
from opnfv_testapi.common import message
from opnfv_testapi.common import raises
from opnfv_testapi.common.config import CONF
query = dict()
date_range = dict()
- query['public'] = {'$not': {'$eq': 'false'}}
for k in self.request.query_arguments.keys():
v = self.get_query_argument(k)
if k == 'project' or k == 'pod' or k == 'case':
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)
- if role:
- del query['public']
- if role != "reviewer":
- query['user'] = username
elif k not in ['last', 'page', 'descend']:
query[k] = v
if date_range:
- criteria : the global criteria status passed or failed
- trust_indicator : evaluate the stability of the test case
to avoid running systematically long and stable test case
- - signed : get logined user result
GET /results/project=functest&case=vPing&version=Arno-R1 \
- &pod=pod_name&period=15&signed
+ &pod=pod_name&period=15
@return 200: all test results consist with query,
empty list if no result is found
@rtype: L{TestResults}
@type trust_indicator: L{float}
@in trust_indicator: query
@required trust_indicator: False
- @param signed: user results or all results
- @type signed: L{string}
- @in signed: query
- @required signed: False
@param descend: true, newest2oldest; false, oldest2newest
@type descend: L{string}
@in descend: query
values_check=values_check)
-class ResultsUploadHandler(ResultsCLHandler):
- @swagger.operation(nickname="uploadTestResult")
- def post(self):
- """
- @description: upload and create a test result
- @param body: result to be created
- @type body: L{ResultCreateRequest}
- @in body: body
- @rtype: L{CreateResponse}
- @return 200: result is created.
- @raise 404: pod/project/testcase not exist
- @raise 400: body/pod_name/project_name/case_name not provided
- """
- logging.info('file upload')
- if not self.request.files:
- raises.NotFound(message.key_error('file'))
- fileinfo = self.request.files['file'][0]
- is_public = self.get_body_argument('public')
- logging.warning('public:%s', is_public)
- logging.info('results is :%s', fileinfo['filename'])
- logging.info('results is :%s', fileinfo['body'])
- self.json_args = json.loads(fileinfo['body']).copy()
- self.json_args['public'] = is_public
-
- openid = self.get_secure_cookie(constants.TESTAPI_ID)
- if openid:
- self.json_args['user'] = openid
-
- super(ResultsUploadHandler, self).post()
-
-
class ResultsGURHandler(GenericResultHandler):
+
@swagger.operation(nickname='getTestResultById')
def get(self, result_id):
"""
from datetime import datetime
from datetime import timedelta
import httplib
-import json
import urllib
from opnfv_testapi.common import message
_, res = self.create_d()
return res.href.split('/')[-1]
- def upload(self, req):
- if req and not isinstance(req, str) and hasattr(req, 'format'):
- req = req.format()
- res = self.fetch(self.basePath + '/upload',
- method='POST',
- body=json.dumps(req),
- headers=self.headers)
-
- return self._get_return(res, self.create_res)
-
-
-class TestResultUpload(TestResultBase):
- @executor.upload(httplib.BAD_REQUEST, message.key_error('file'))
- def test_filenotfind(self):
- return None
-
class TestResultCreate(TestResultBase):
@executor.create(httplib.BAD_REQUEST, message.no_body())
def test_queryLast(self):
return self._set_query(last=1)
- @executor.query(httplib.OK, '_query_success', 4)
- def test_queryPublic(self):
- self._create_public_data()
- return self._set_query()
-
- @executor.query(httplib.OK, '_query_success', 1)
- def test_queryPrivate(self):
- self._create_private_data()
- return self._set_query(public='false')
-
@executor.query(httplib.OK, '_query_period_one', 1)
def test_combination(self):
return self._set_query('pod',
self.create(req)
return req
- def _create_public_data(self, **kwargs):
- req = copy.deepcopy(self.req_d)
- req.public = 'true'
- self.create(req)
- return req
-
- def _create_private_data(self, **kwargs):
- req = copy.deepcopy(self.req_d)
- req.public = 'false'
- self.create(req)
- return req
-
def _set_query(self, *args, **kwargs):
def get_value(arg):
if arg in ['pod', 'project', 'case']: