xtesting/core/testcase.py

   1 #!/usr/bin/env python
   2
   3 # Copyright (c) 2016 Orange and others.
   4 #
   5 # All rights reserved. This program and the accompanying materials
   6 # are made available under the terms of the Apache License, Version 2.0
   7 # which accompanies this distribution, and is available at
   8 # http://www.apache.org/licenses/LICENSE-2.0
   9
  10 """Define the parent class of all Xtesting TestCases."""
  11
  12 import abc
  13 from datetime import datetime
  14 import json
  15 import logging
  16 import os
  17 import re
  18 import requests
  19
  20 import prettytable
  21 import six
  22
  23 from xtesting.utils import decorators
  24 from xtesting.utils import env
  25
  26 __author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
  27
  28
  29 @six.add_metaclass(abc.ABCMeta)
  30 class TestCase(object):
  31     # pylint: disable=too-many-instance-attributes
  32     """Base model for single test case."""
  33
  34     EX_OK = os.EX_OK
  35     """everything is OK"""
  36
  37     EX_RUN_ERROR = os.EX_SOFTWARE
  38     """run() failed"""
  39
  40     EX_PUSH_TO_DB_ERROR = os.EX_SOFTWARE - 1
  41     """push_to_db() failed"""
  42
  43     EX_TESTCASE_FAILED = os.EX_SOFTWARE - 2
  44     """results are false"""
  45
  46     EX_TESTCASE_SKIPPED = os.EX_SOFTWARE - 3
  47     """requirements are unmet"""
  48
  49     _job_name_rule = "(dai|week)ly-(.+?)-[0-9]*"
  50     _headers = {'Content-Type': 'application/json'}
  51     __logger = logging.getLogger(__name__)
  52
  53     def __init__(self, **kwargs):
  54         self.details = {}
  55         self.project_name = kwargs.get('project_name', 'xtesting')
  56         self.case_name = kwargs.get('case_name', '')
  57         self.criteria = kwargs.get('criteria', 100)
  58         self.result = 0
  59         self.start_time = 0
  60         self.stop_time = 0
  61         self.is_skipped = False
  62
  63     def __str__(self):
  64         try:
  65             assert self.project_name
  66             assert self.case_name
  67             if self.is_skipped:
  68                 result = 'SKIP'
  69             else:
  70                 result = 'PASS' if(self.is_successful(
  71                     ) == TestCase.EX_OK) else 'FAIL'
  72             msg = prettytable.PrettyTable(
  73                 header_style='upper', padding_width=5,
  74                 field_names=['test case', 'project', 'duration',
  75                              'result'])
  76             msg.add_row([self.case_name, self.project_name,
  77                          self.get_duration(), result])
  78             return msg.get_string()
  79         except AssertionError:
  80             self.__logger.error("We cannot print invalid objects")
  81             return super(TestCase, self).__str__()
  82
  83     def get_duration(self):
  84         """Return the duration of the test case.
  85
  86         Returns:
  87             duration if start_time and stop_time are set
  88             "XX:XX" otherwise.
  89         """
  90         try:
  91             if self.is_skipped:
  92                 return "00:00"
  93             assert self.start_time
  94             assert self.stop_time
  95             if self.stop_time < self.start_time:
  96                 return "XX:XX"
  97             return "{0[0]:02.0f}:{0[1]:02.0f}".format(divmod(
  98                 self.stop_time - self.start_time, 60))
  99         except Exception:  # pylint: disable=broad-except
 100             self.__logger.error("Please run test before getting the duration")
 101             return "XX:XX"
 102
 103     def is_successful(self):
 104         """Interpret the result of the test case.
 105
 106         It allows getting the result of TestCase. It completes run()
 107         which only returns the execution status.
 108
 109         It can be overriden if checking result is not suitable.
 110
 111         Returns:
 112             TestCase.EX_OK if result is 'PASS'.
 113             TestCase.EX_TESTCASE_SKIPPED if test case is skipped.
 114             TestCase.EX_TESTCASE_FAILED otherwise.
 115         """
 116         try:
 117             if self.is_skipped:
 118                 return TestCase.EX_TESTCASE_SKIPPED
 119             assert self.criteria
 120             assert self.result is not None
 121             if (not isinstance(self.result, str) and
 122                     not isinstance(self.criteria, str)):
 123                 if self.result >= self.criteria:
 124                     return TestCase.EX_OK
 125             else:
 126                 # Backward compatibility
 127                 # It must be removed as soon as TestCase subclasses
 128                 # stop setting result = 'PASS' or 'FAIL'.
 129                 # In this case criteria is unread.
 130                 self.__logger.warning(
 131                     "Please update result which must be an int!")
 132                 if self.result == 'PASS':
 133                     return TestCase.EX_OK
 134         except AssertionError:
 135             self.__logger.error("Please run test before checking the results")
 136         return TestCase.EX_TESTCASE_FAILED
 137
 138     def check_requirements(self):  # pylint: disable=no-self-use
 139         """Check the requirements of the test case.
 140
 141         It can be overriden on purpose.
 142         """
 143         self.is_skipped = False
 144
 145     @abc.abstractmethod
 146     def run(self, **kwargs):
 147         """Run the test case.
 148
 149         It allows running TestCase and getting its execution
 150         status.
 151
 152         The subclasses must override the default implementation which
 153         is false on purpose.
 154
 155         The new implementation must set the following attributes to
 156         push the results to DB:
 157
 158             * result,
 159             * start_time,
 160             * stop_time.
 161
 162         Args:
 163             kwargs: Arbitrary keyword arguments.
 164         """
 165
 166     @decorators.can_dump_request_to_file
 167     def push_to_db(self):
 168         """Push the results of the test case to the DB.
 169
 170         It allows publishing the results and checking the status.
 171
 172         It could be overriden if the common implementation is not
 173         suitable.
 174
 175         The following attributes must be set before pushing the results to DB:
 176
 177             * project_name,
 178             * case_name,
 179             * result,
 180             * start_time,
 181             * stop_time.
 182
 183         The next vars must be set in env:
 184
 185             * TEST_DB_URL,
 186             * INSTALLER_TYPE,
 187             * DEPLOY_SCENARIO,
 188             * NODE_NAME,
 189             * BUILD_TAG.
 190
 191         Returns:
 192             TestCase.EX_OK if results were pushed to DB.
 193             TestCase.EX_PUSH_TO_DB_ERROR otherwise.
 194         """
 195         try:
 196             if self.is_skipped:
 197                 return TestCase.EX_PUSH_TO_DB_ERROR
 198             assert self.project_name
 199             assert self.case_name
 200             assert self.start_time
 201             assert self.stop_time
 202             url = env.get('TEST_DB_URL')
 203             data = {"project_name": self.project_name,
 204                     "case_name": self.case_name,
 205                     "details": self.details}
 206             data["installer"] = env.get('INSTALLER_TYPE')
 207             data["scenario"] = env.get('DEPLOY_SCENARIO')
 208             data["pod_name"] = env.get('NODE_NAME')
 209             data["build_tag"] = env.get('BUILD_TAG')
 210             data["criteria"] = 'PASS' if self.is_successful(
 211                 ) == TestCase.EX_OK else 'FAIL'
 212             data["start_date"] = datetime.fromtimestamp(
 213                 self.start_time).strftime('%Y-%m-%d %H:%M:%S')
 214             data["stop_date"] = datetime.fromtimestamp(
 215                 self.stop_time).strftime('%Y-%m-%d %H:%M:%S')
 216             try:
 217                 data["version"] = re.search(
 218                     TestCase._job_name_rule,
 219                     env.get('BUILD_TAG')).group(2)
 220             except Exception:  # pylint: disable=broad-except
 221                 data["version"] = "unknown"
 222             req = requests.post(
 223                 url, data=json.dumps(data, sort_keys=True),
 224                 headers=self._headers)
 225             req.raise_for_status()
 226             self.__logger.info(
 227                 "The results were successfully pushed to DB")
 228         except AssertionError:
 229             self.__logger.exception(
 230                 "Please run test before publishing the results")
 231             return TestCase.EX_PUSH_TO_DB_ERROR
 232         except requests.exceptions.HTTPError:
 233             self.__logger.exception("The HTTP request raises issues")
 234             return TestCase.EX_PUSH_TO_DB_ERROR
 235         except Exception:  # pylint: disable=broad-except
 236             self.__logger.exception("The results cannot be pushed to DB")
 237             return TestCase.EX_PUSH_TO_DB_ERROR
 238         return TestCase.EX_OK
 239
 240     def clean(self):
 241         """Clean the resources.
 242
 243         It can be overriden if resources must be deleted after
 244         running the test case.
 245         """