utils/test/testapi/opnfv_testapi/tests/unit/test_result.py

   1 ##############################################################################
   2 # Copyright (c) 2016 ZTE Corporation
   3 # feng.xiaowei@zte.com.cn
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9 import copy
  10 from datetime import datetime, timedelta
  11 import unittest
  12
  13 from opnfv_testapi.common import constants
  14 from opnfv_testapi.resources import pod_models
  15 from opnfv_testapi.resources import project_models
  16 from opnfv_testapi.resources import result_models
  17 from opnfv_testapi.resources import testcase_models
  18 import test_base as base
  19
  20
  21 class Details(object):
  22     def __init__(self, timestart=None, duration=None, status=None):
  23         self.timestart = timestart
  24         self.duration = duration
  25         self.status = status
  26         self.items = [{'item1': 1}, {'item2': 2}]
  27
  28     def format(self):
  29         return {
  30             "timestart": self.timestart,
  31             "duration": self.duration,
  32             "status": self.status,
  33             'items': [{'item1': 1}, {'item2': 2}]
  34         }
  35
  36     @staticmethod
  37     def from_dict(a_dict):
  38
  39         if a_dict is None:
  40             return None
  41
  42         t = Details()
  43         t.timestart = a_dict.get('timestart')
  44         t.duration = a_dict.get('duration')
  45         t.status = a_dict.get('status')
  46         t.items = a_dict.get('items')
  47         return t
  48
  49
  50 class TestResultBase(base.TestBase):
  51     def setUp(self):
  52         self.pod = 'zte-pod1'
  53         self.project = 'functest'
  54         self.case = 'vPing'
  55         self.installer = 'fuel'
  56         self.version = 'C'
  57         self.build_tag = 'v3.0'
  58         self.scenario = 'odl-l2'
  59         self.criteria = 'passed'
  60         self.trust_indicator = result_models.TI(0.7)
  61         self.start_date = "2016-05-23 07:16:09.477097"
  62         self.stop_date = "2016-05-23 07:16:19.477097"
  63         self.update_date = "2016-05-24 07:16:19.477097"
  64         self.update_step = -0.05
  65         super(TestResultBase, self).setUp()
  66         self.details = Details(timestart='0', duration='9s', status='OK')
  67         self.req_d = result_models.ResultCreateRequest(
  68             pod_name=self.pod,
  69             project_name=self.project,
  70             case_name=self.case,
  71             installer=self.installer,
  72             version=self.version,
  73             start_date=self.start_date,
  74             stop_date=self.stop_date,
  75             details=self.details.format(),
  76             build_tag=self.build_tag,
  77             scenario=self.scenario,
  78             criteria=self.criteria,
  79             trust_indicator=self.trust_indicator)
  80         self.get_res = result_models.TestResult
  81         self.list_res = result_models.TestResults
  82         self.update_res = result_models.TestResult
  83         self.basePath = '/api/v1/results'
  84         self.req_pod = pod_models.PodCreateRequest(
  85             self.pod,
  86             'metal',
  87             'zte pod 1')
  88         self.req_project = project_models.ProjectCreateRequest(
  89             self.project,
  90             'vping test')
  91         self.req_testcase = testcase_models.TestcaseCreateRequest(
  92             self.case,
  93             '/cases/vping',
  94             'vping-ssh test')
  95         self.create_help('/api/v1/pods', self.req_pod)
  96         self.create_help('/api/v1/projects', self.req_project)
  97         self.create_help('/api/v1/projects/%s/cases',
  98                          self.req_testcase,
  99                          self.project)
 100
 101     def assert_res(self, code, result, req=None):
 102         self.assertEqual(code, constants.HTTP_OK)
 103         if req is None:
 104             req = self.req_d
 105         self.assertEqual(result.pod_name, req.pod_name)
 106         self.assertEqual(result.project_name, req.project_name)
 107         self.assertEqual(result.case_name, req.case_name)
 108         self.assertEqual(result.installer, req.installer)
 109         self.assertEqual(result.version, req.version)
 110         details_req = Details.from_dict(req.details)
 111         details_res = Details.from_dict(result.details)
 112         self.assertEqual(details_res.duration, details_req.duration)
 113         self.assertEqual(details_res.timestart, details_req.timestart)
 114         self.assertEqual(details_res.status, details_req.status)
 115         self.assertEqual(details_res.items, details_req.items)
 116         self.assertEqual(result.build_tag, req.build_tag)
 117         self.assertEqual(result.scenario, req.scenario)
 118         self.assertEqual(result.criteria, req.criteria)
 119         self.assertEqual(result.start_date, req.start_date)
 120         self.assertEqual(result.stop_date, req.stop_date)
 121         self.assertIsNotNone(result._id)
 122         ti = result.trust_indicator
 123         self.assertEqual(ti.current, req.trust_indicator.current)
 124         if ti.histories:
 125             history = ti.histories[0]
 126             self.assertEqual(history.date, self.update_date)
 127             self.assertEqual(history.step, self.update_step)
 128
 129     def _create_d(self):
 130         _, res = self.create_d()
 131         return res.href.split('/')[-1]
 132
 133
 134 class TestResultCreate(TestResultBase):
 135     def test_nobody(self):
 136         (code, body) = self.create(None)
 137         self.assertEqual(code, constants.HTTP_BAD_REQUEST)
 138         self.assertIn('no body', body)
 139
 140     def test_podNotProvided(self):
 141         req = self.req_d
 142         req.pod_name = None
 143         (code, body) = self.create(req)
 144         self.assertEqual(code, constants.HTTP_BAD_REQUEST)
 145         self.assertIn('pod_name missing', body)
 146
 147     def test_projectNotProvided(self):
 148         req = self.req_d
 149         req.project_name = None
 150         (code, body) = self.create(req)
 151         self.assertEqual(code, constants.HTTP_BAD_REQUEST)
 152         self.assertIn('project_name missing', body)
 153
 154     def test_testcaseNotProvided(self):
 155         req = self.req_d
 156         req.case_name = None
 157         (code, body) = self.create(req)
 158         self.assertEqual(code, constants.HTTP_BAD_REQUEST)
 159         self.assertIn('case_name missing', body)
 160
 161     def test_noPod(self):
 162         req = self.req_d
 163         req.pod_name = 'notExistPod'
 164         (code, body) = self.create(req)
 165         self.assertEqual(code, constants.HTTP_NOT_FOUND)
 166         self.assertIn('Could not find pod', body)
 167
 168     def test_noProject(self):
 169         req = self.req_d
 170         req.project_name = 'notExistProject'
 171         (code, body) = self.create(req)
 172         self.assertEqual(code, constants.HTTP_NOT_FOUND)
 173         self.assertIn('Could not find project', body)
 174
 175     def test_noTestcase(self):
 176         req = self.req_d
 177         req.case_name = 'notExistTestcase'
 178         (code, body) = self.create(req)
 179         self.assertEqual(code, constants.HTTP_NOT_FOUND)
 180         self.assertIn('Could not find testcase', body)
 181
 182     def test_success(self):
 183         (code, body) = self.create_d()
 184         self.assertEqual(code, constants.HTTP_OK)
 185         self.assert_href(body)
 186
 187     def test_key_with_doc(self):
 188         req = copy.deepcopy(self.req_d)
 189         req.details = {'1.name': 'dot_name'}
 190         (code, body) = self.create(req)
 191         self.assertEqual(code, constants.HTTP_OK)
 192         self.assert_href(body)
 193
 194     def test_no_ti(self):
 195         req = result_models.ResultCreateRequest(pod_name=self.pod,
 196                                                 project_name=self.project,
 197                                                 case_name=self.case,
 198                                                 installer=self.installer,
 199                                                 version=self.version,
 200                                                 start_date=self.start_date,
 201                                                 stop_date=self.stop_date,
 202                                                 details=self.details.format(),
 203                                                 build_tag=self.build_tag,
 204                                                 scenario=self.scenario,
 205                                                 criteria=self.criteria)
 206         (code, res) = self.create(req)
 207         _id = res.href.split('/')[-1]
 208         self.assertEqual(code, constants.HTTP_OK)
 209         code, body = self.get(_id)
 210         self.assert_res(code, body, req)
 211
 212
 213 class TestResultGet(TestResultBase):
 214     def test_getOne(self):
 215         _id = self._create_d()
 216         code, body = self.get(_id)
 217         self.assert_res(code, body)
 218
 219     def test_queryPod(self):
 220         self._query_and_assert(self._set_query('pod'))
 221
 222     def test_queryProject(self):
 223         self._query_and_assert(self._set_query('project'))
 224
 225     def test_queryTestcase(self):
 226         self._query_and_assert(self._set_query('case'))
 227
 228     def test_queryVersion(self):
 229         self._query_and_assert(self._set_query('version'))
 230
 231     def test_queryInstaller(self):
 232         self._query_and_assert(self._set_query('installer'))
 233
 234     def test_queryBuildTag(self):
 235         self._query_and_assert(self._set_query('build_tag'))
 236
 237     def test_queryScenario(self):
 238         self._query_and_assert(self._set_query('scenario'))
 239
 240     def test_queryTrustIndicator(self):
 241         self._query_and_assert(self._set_query('trust_indicator'))
 242
 243     def test_queryCriteria(self):
 244         self._query_and_assert(self._set_query('criteria'))
 245
 246     def test_queryPeriodNotInt(self):
 247         code, body = self.query(self._set_query('period=a'))
 248         self.assertEqual(code, constants.HTTP_BAD_REQUEST)
 249         self.assertIn('period must be int', body)
 250
 251     def test_queryPeriodFail(self):
 252         self._query_and_assert(self._set_query('period=1'),
 253                                found=False, days=-10)
 254
 255     def test_queryPeriodSuccess(self):
 256         self._query_and_assert(self._set_query('period=1'),
 257                                found=True)
 258
 259     def test_queryLastNotInt(self):
 260         code, body = self.query(self._set_query('last=a'))
 261         self.assertEqual(code, constants.HTTP_BAD_REQUEST)
 262         self.assertIn('last must be int', body)
 263
 264     def test_queryLast(self):
 265         self._create_changed_date()
 266         req = self._create_changed_date(minutes=20)
 267         self._create_changed_date(minutes=-20)
 268         self._query_and_assert(self._set_query('last=1'), req=req)
 269
 270     def test_combination(self):
 271         self._query_and_assert(self._set_query('pod',
 272                                                'project',
 273                                                'case',
 274                                                'version',
 275                                                'installer',
 276                                                'build_tag',
 277                                                'scenario',
 278                                                'trust_indicator',
 279                                                'criteria',
 280                                                'period=1'))
 281
 282     def test_notFound(self):
 283         self._query_and_assert(self._set_query('pod=notExistPod',
 284                                                'project',
 285                                                'case',
 286                                                'version',
 287                                                'installer',
 288                                                'build_tag',
 289                                                'scenario',
 290                                                'trust_indicator',
 291                                                'criteria',
 292                                                'period=1'),
 293                                found=False)
 294
 295     def _query_and_assert(self, query, found=True, req=None, **kwargs):
 296         if req is None:
 297             req = self._create_changed_date(**kwargs)
 298         code, body = self.query(query)
 299         if not found:
 300             self.assertEqual(code, constants.HTTP_OK)
 301             self.assertEqual(0, len(body.results))
 302         else:
 303             self.assertEqual(1, len(body.results))
 304             for result in body.results:
 305                 self.assert_res(code, result, req)
 306
 307     def _create_changed_date(self, **kwargs):
 308         req = copy.deepcopy(self.req_d)
 309         req.start_date = datetime.now() + timedelta(**kwargs)
 310         req.stop_date = str(req.start_date + timedelta(minutes=10))
 311         req.start_date = str(req.start_date)
 312         self.create(req)
 313         return req
 314
 315     def _set_query(self, *args):
 316         def get_value(arg):
 317             return self.__getattribute__(arg) \
 318                 if arg != 'trust_indicator' else self.trust_indicator.current
 319         uri = ''
 320         for arg in args:
 321             if '=' in arg:
 322                 uri += arg + '&'
 323             else:
 324                 uri += '{}={}&'.format(arg, get_value(arg))
 325         return uri[0: -1]
 326
 327
 328 class TestResultUpdate(TestResultBase):
 329     def test_success(self):
 330         _id = self._create_d()
 331
 332         new_ti = copy.deepcopy(self.trust_indicator)
 333         new_ti.current += self.update_step
 334         new_ti.histories.append(
 335             result_models.TIHistory(self.update_date, self.update_step))
 336         new_data = copy.deepcopy(self.req_d)
 337         new_data.trust_indicator = new_ti
 338         update = result_models.ResultUpdateRequest(trust_indicator=new_ti)
 339         code, body = self.update(update, _id)
 340         self.assertEqual(_id, body._id)
 341         self.assert_res(code, body, new_data)
 342
 343         code, new_body = self.get(_id)
 344         self.assertEqual(_id, new_body._id)
 345         self.assert_res(code, new_body, new_data)
 346
 347
 348 if __name__ == '__main__':
 349     unittest.main()