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