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