functest/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 Functest TestCases."""
  11
  12 import logging
  13 import os
  14
  15 import prettytable
  16
  17 import functest.utils.functest_utils as ft_utils
  18 import functest.utils.openstack_clean as os_clean
  19 import functest.utils.openstack_snapshot as os_snapshot
  20
  21 __author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
  22
  23
  24 class TestCase(object):
  25     """Base model for single test case."""
  26
  27     EX_OK = os.EX_OK
  28     """everything is OK"""
  29
  30     EX_RUN_ERROR = os.EX_SOFTWARE
  31     """run() failed"""
  32
  33     EX_PUSH_TO_DB_ERROR = os.EX_SOFTWARE - 1
  34     """push_to_db() failed"""
  35
  36     EX_TESTCASE_FAILED = os.EX_SOFTWARE - 2
  37     """results are false"""
  38
  39     __logger = logging.getLogger(__name__)
  40
  41     def __init__(self, **kwargs):
  42         self.details = {}
  43         self.project_name = kwargs.get('project_name', 'functest')
  44         self.case_name = kwargs.get('case_name', '')
  45         self.criteria = kwargs.get('criteria', 100)
  46         self.result = 0
  47         self.start_time = 0
  48         self.stop_time = 0
  49
  50     def __str__(self):
  51         try:
  52             assert self.project_name
  53             assert self.case_name
  54             result = 'PASS' if(self.is_successful(
  55                 ) == TestCase.EX_OK) else 'FAIL'
  56             msg = prettytable.PrettyTable(
  57                 header_style='upper', padding_width=5,
  58                 field_names=['test case', 'project', 'duration',
  59                              'result'])
  60             msg.add_row([self.case_name, self.project_name,
  61                          self.get_duration(), result])
  62             return msg.get_string()
  63         except AssertionError:
  64             self.__logger.error("We cannot print invalid objects")
  65             return super(TestCase, self).__str__()
  66
  67     def get_duration(self):
  68         """Return the duration of the test case.
  69
  70         Returns:
  71             duration if start_time and stop_time are set
  72             "XX:XX" otherwise.
  73         """
  74         try:
  75             assert self.start_time
  76             assert self.stop_time
  77             if self.stop_time < self.start_time:
  78                 return "XX:XX"
  79             return "{0[0]:02.0f}:{0[1]:02.0f}".format(divmod(
  80                 self.stop_time - self.start_time, 60))
  81         except Exception:  # pylint: disable=broad-except
  82             self.__logger.error("Please run test before getting the duration")
  83             return "XX:XX"
  84
  85     def is_successful(self):
  86         """Interpret the result of the test case.
  87
  88         It allows getting the result of TestCase. It completes run()
  89         which only returns the execution status.
  90
  91         It can be overriden if checking result is not suitable.
  92
  93         Returns:
  94             TestCase.EX_OK if result is 'PASS'.
  95             TestCase.EX_TESTCASE_FAILED otherwise.
  96         """
  97         try:
  98             assert self.criteria
  99             assert self.result is not None
 100             if (not isinstance(self.result, str) and
 101                     not isinstance(self.criteria, str)):
 102                 if self.result >= self.criteria:
 103                     return TestCase.EX_OK
 104             else:
 105                 # Backward compatibility
 106                 # It must be removed as soon as TestCase subclasses
 107                 # stop setting result = 'PASS' or 'FAIL'.
 108                 # In this case criteria is unread.
 109                 self.__logger.warning(
 110                     "Please update result which must be an int!")
 111                 if self.result == 'PASS':
 112                     return TestCase.EX_OK
 113         except AssertionError:
 114             self.__logger.error("Please run test before checking the results")
 115         return TestCase.EX_TESTCASE_FAILED
 116
 117     def run(self, **kwargs):
 118         """Run the test case.
 119
 120         It allows running TestCase and getting its execution
 121         status.
 122
 123         The subclasses must override the default implementation which
 124         is false on purpose.
 125
 126         The new implementation must set the following attributes to
 127         push the results to DB:
 128
 129             * result,
 130             * start_time,
 131             * stop_time.
 132
 133         Args:
 134             kwargs: Arbitrary keyword arguments.
 135
 136         Returns:
 137             TestCase.EX_RUN_ERROR.
 138         """
 139         # pylint: disable=unused-argument
 140         self.__logger.error("Run must be implemented")
 141         return TestCase.EX_RUN_ERROR
 142
 143     def push_to_db(self):
 144         """Push the results of the test case to the DB.
 145
 146         It allows publishing the results and to check the status.
 147
 148         It could be overriden if the common implementation is not
 149         suitable. The following attributes must be set before pushing
 150         the results to DB:
 151
 152             * project_name,
 153             * case_name,
 154             * result,
 155             * start_time,
 156             * stop_time.
 157
 158         Returns:
 159             TestCase.EX_OK if results were pushed to DB.
 160             TestCase.EX_PUSH_TO_DB_ERROR otherwise.
 161         """
 162         try:
 163             assert self.project_name
 164             assert self.case_name
 165             assert self.start_time
 166             assert self.stop_time
 167             pub_result = 'PASS' if self.is_successful(
 168                 ) == TestCase.EX_OK else 'FAIL'
 169             if ft_utils.push_results_to_db(
 170                     self.project_name, self.case_name, self.start_time,
 171                     self.stop_time, pub_result, self.details):
 172                 self.__logger.info(
 173                     "The results were successfully pushed to DB")
 174                 return TestCase.EX_OK
 175             else:
 176                 self.__logger.error("The results cannot be pushed to DB")
 177                 return TestCase.EX_PUSH_TO_DB_ERROR
 178         except Exception:  # pylint: disable=broad-except
 179             self.__logger.exception("The results cannot be pushed to DB")
 180             return TestCase.EX_PUSH_TO_DB_ERROR
 181
 182     def create_snapshot(self):  # pylint: disable=no-self-use
 183         """Save the testing environment before running test.
 184
 185         It can be overriden if resources must be listed running the
 186         test case.
 187
 188         Returns:
 189             TestCase.EX_OK
 190         """
 191         return TestCase.EX_OK
 192
 193     def clean(self):
 194         """Clean the resources.
 195
 196         It can be overriden if resources must be deleted after
 197         running the test case.
 198         """
 199
 200
 201 class OSGCTestCase(TestCase):
 202     """Model for single test case which requires an OpenStack Garbage
 203     Collector."""
 204
 205     __logger = logging.getLogger(__name__)
 206
 207     def create_snapshot(self):
 208         """Create a snapshot listing the OpenStack resources.
 209
 210         Returns:
 211             TestCase.EX_OK if os_snapshot.main() returns 0.
 212             TestCase.EX_RUN_ERROR otherwise.
 213         """
 214         try:
 215             assert os_snapshot.main() == 0
 216             self.__logger.info("OpenStack resources snapshot created")
 217             return TestCase.EX_OK
 218         except Exception:  # pylint: disable=broad-except
 219             self.__logger.exception("Cannot create the snapshot")
 220             return TestCase.EX_RUN_ERROR
 221
 222     def clean(self):
 223         """Clean the OpenStack resources."""
 224         try:
 225             assert os_clean.main() == 0
 226             self.__logger.info("OpenStack resources cleaned")
 227         except Exception:  # pylint: disable=broad-except
 228             self.__logger.exception("Cannot clean the OpenStack resources")