# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################
import copy
+from datetime import datetime
+from datetime import timedelta
import httplib
+import json
import unittest
-from datetime import datetime, timedelta
from opnfv_testapi.common import message
-from opnfv_testapi.resources import pod_models
from opnfv_testapi.resources import project_models
from opnfv_testapi.resources import result_models
from opnfv_testapi.resources import testcase_models
from opnfv_testapi.tests.unit import executor
+from opnfv_testapi.tests.unit import fake_pymongo
from opnfv_testapi.tests.unit.resources import test_base as base
class TestResultBase(base.TestBase):
def setUp(self):
- self.pod = 'zte-pod1'
+ super(TestResultBase, self).setUp()
+ self.pod = self.pod_d.name
self.project = 'functest'
self.case = 'vPing'
self.installer = 'fuel'
self.scenario = 'odl-l2'
self.criteria = 'passed'
self.trust_indicator = result_models.TI(0.7)
- self.start_date = "2016-05-23 07:16:09.477097"
- self.stop_date = "2016-05-23 07:16:19.477097"
- self.update_date = "2016-05-24 07:16:19.477097"
+ self.start_date = str(datetime.now())
+ self.stop_date = str(datetime.now() + timedelta(minutes=1))
+ self.update_date = str(datetime.now() + timedelta(days=1))
self.update_step = -0.05
- super(TestResultBase, self).setUp()
self.details = Details(timestart='0', duration='9s', status='OK')
self.req_d = result_models.ResultCreateRequest(
pod_name=self.pod,
self.list_res = result_models.TestResults
self.update_res = result_models.TestResult
self.basePath = '/api/v1/results'
- self.req_pod = pod_models.PodCreateRequest(
- self.pod,
- 'metal',
- 'zte pod 1')
self.req_project = project_models.ProjectCreateRequest(
self.project,
'vping test')
self.case,
'/cases/vping',
'vping-ssh test')
- self.create_help('/api/v1/pods', self.req_pod)
+ fake_pymongo.pods.insert(self.pod_d.format())
self.create_help('/api/v1/projects', self.req_project)
self.create_help('/api/v1/projects/%s/cases',
self.req_testcase,
_, 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())
class TestResultGet(TestResultBase):
def setUp(self):
super(TestResultGet, 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)
- self.req_10d_before = self._create_changed_date(days=-10)
@executor.get(httplib.OK, 'assert_res')
def test_getOne(self):
def test_queryPeriodNotInt(self):
return self._set_query('period=a')
- @executor.query(httplib.OK, '_query_last_one', 1)
+ @executor.query(httplib.OK, '_query_period_one', 1)
def test_queryPeriodSuccess(self):
- return self._set_query('period=1')
+ return self._set_query('period=5')
@executor.query(httplib.BAD_REQUEST, message.must_int('last'))
def test_queryLastNotInt(self):
def test_queryLast(self):
return self._set_query('last=1')
- @executor.query(httplib.OK, '_query_last_one', 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',
'project',
'scenario',
'trust_indicator',
'criteria',
- 'period=1')
+ 'period=5')
@executor.query(httplib.OK, '_query_success', 0)
def test_notFound(self):
'criteria',
'period=1')
+ @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.results))
self.assertEqual(number, len(body.results))
self.assert_res(body.results[0], self.req_10d_later)
+ def _query_period_one(self, body, number):
+ self.assertEqual(number, len(body.results))
+ self.assert_res(body.results[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.start_date = datetime.now() + timedelta(**kwargs)
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):
def get_value(arg):
return self.__getattribute__(arg) \
if arg != 'trust_indicator' else self.trust_indicator.current
uri = ''
for arg in args:
- if '=' in arg:
- uri += arg + '&'
- else:
- uri += '{}={}&'.format(arg, get_value(arg))
+ if arg:
+ if '=' in arg:
+ uri += arg + '&'
+ else:
+ uri += '{}={}&'.format(arg, get_value(arg))
return uri[0: -1]