# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
-"""Define the parent class of Functest TestCase."""
+"""Define the parent class of all Functest TestCases."""
+import logging
import os
-import functest.utils.functest_logger as ft_logger
+import prettytable
+
import functest.utils.functest_utils as ft_utils
+import functest.utils.openstack_clean as os_clean
+import functest.utils.openstack_snapshot as os_snapshot
__author__ = "Cedric Ollivier <cedric.ollivier@orange.com>"
class TestCase(object):
- """Parent class of Functest TestCase."""
+ """Base model for single test case."""
EX_OK = os.EX_OK
- """Status code returned when everything is OK"""
+ """everything is OK"""
EX_RUN_ERROR = os.EX_SOFTWARE
- """Status code returned when run() fails"""
+ """run() failed"""
EX_PUSH_TO_DB_ERROR = os.EX_SOFTWARE - 1
- """Status code returned when push_to_db() fails"""
+ """push_to_db() failed"""
EX_TESTCASE_FAILED = os.EX_SOFTWARE - 2
- """Status code returned when results are false"""
+ """results are false"""
- logger = ft_logger.Logger(__name__).getLogger()
+ __logger = logging.getLogger(__name__)
def __init__(self, **kwargs):
self.details = {}
self.project_name = kwargs.get('project_name', 'functest')
self.case_name = kwargs.get('case_name', '')
- self.criteria = ""
- self.start_time = ""
- self.stop_time = ""
+ self.criteria = kwargs.get('criteria', 100)
+ self.result = 0
+ self.start_time = 0
+ self.stop_time = 0
+
+ def __str__(self):
+ try:
+ assert self.project_name
+ assert self.case_name
+ result = 'PASS' if(self.is_successful(
+ ) == TestCase.EX_OK) else 'FAIL'
+ msg = prettytable.PrettyTable(
+ header_style='upper', padding_width=5,
+ field_names=['test case', 'project', 'duration',
+ 'result'])
+ msg.add_row([self.case_name, self.project_name,
+ self.get_duration(), result])
+ return msg.get_string()
+ except AssertionError:
+ self.__logger.error("We cannot print invalid objects")
+ return super(TestCase, self).__str__()
+
+ def get_duration(self):
+ """Return the duration of the test case.
+
+ Returns:
+ duration if start_time and stop_time are set
+ "XX:XX" otherwise.
+ """
+ try:
+ assert self.start_time
+ assert self.stop_time
+ if self.stop_time < self.start_time:
+ return "XX:XX"
+ return "{0[0]:02.0f}:{0[1]:02.0f}".format(divmod(
+ self.stop_time - self.start_time, 60))
+ except Exception: # pylint: disable=broad-except
+ self.__logger.error("Please run test before getting the duration")
+ return "XX:XX"
- def check_criteria(self):
- """Interpret the results of TestCase.
+ def is_successful(self):
+ """Interpret the result of the test case.
- It allows getting the results of TestCase. It completes run()
+ It allows getting the result of TestCase. It completes run()
which only returns the execution status.
- It can be overriden if checking criteria is not suitable.
+ It can be overriden if checking result is not suitable.
Returns:
- TestCase.EX_OK if criteria is 'PASS'.
+ TestCase.EX_OK if result is 'PASS'.
TestCase.EX_TESTCASE_FAILED otherwise.
"""
try:
assert self.criteria
- if self.criteria == 'PASS':
- return TestCase.EX_OK
+ assert self.result is not None
+ if (not isinstance(self.result, str) and
+ not isinstance(self.criteria, str)):
+ if self.result >= self.criteria:
+ return TestCase.EX_OK
+ else:
+ # Backward compatibility
+ # It must be removed as soon as TestCase subclasses
+ # stop setting result = 'PASS' or 'FAIL'.
+ # In this case criteria is unread.
+ self.__logger.warning(
+ "Please update result which must be an int!")
+ if self.result == 'PASS':
+ return TestCase.EX_OK
except AssertionError:
- self.logger.error("Please run test before checking the results")
+ self.__logger.error("Please run test before checking the results")
return TestCase.EX_TESTCASE_FAILED
def run(self, **kwargs):
- """Run TestCase.
+ """Run the test case.
It allows running TestCase and getting its execution
status.
The subclasses must override the default implementation which
- is false on purpose. The only prerequisite is to set the
- following attributes to push the results to DB:
+ is false on purpose.
- * case_name,
- * criteria,
+ The new implementation must set the following attributes to
+ push the results to DB:
+
+ * result,
* start_time,
* stop_time.
TestCase.EX_RUN_ERROR.
"""
# pylint: disable=unused-argument
- self.logger.error("Run must be implemented")
+ self.__logger.error("Run must be implemented")
return TestCase.EX_RUN_ERROR
def push_to_db(self):
- """Push the results of TestCase to the DB.
+ """Push the results of the test case to the DB.
It allows publishing the results and to check the status.
* project_name,
* case_name,
- * criteria,
+ * result,
* start_time,
* stop_time.
try:
assert self.project_name
assert self.case_name
- assert self.criteria
assert self.start_time
assert self.stop_time
+ pub_result = 'PASS' if self.is_successful(
+ ) == TestCase.EX_OK else 'FAIL'
if ft_utils.push_results_to_db(
self.project_name, self.case_name, self.start_time,
- self.stop_time, self.criteria, self.details):
- self.logger.info("The results were successfully pushed to DB")
+ self.stop_time, pub_result, self.details):
+ self.__logger.info(
+ "The results were successfully pushed to DB")
return TestCase.EX_OK
else:
- self.logger.error("The results cannot be pushed to DB")
+ self.__logger.error("The results cannot be pushed to DB")
return TestCase.EX_PUSH_TO_DB_ERROR
except Exception: # pylint: disable=broad-except
- self.logger.exception("The results cannot be pushed to DB")
+ self.__logger.exception("The results cannot be pushed to DB")
return TestCase.EX_PUSH_TO_DB_ERROR
+
+ def create_snapshot(self): # pylint: disable=no-self-use
+ """Save the testing environment before running test.
+
+ It can be overriden if resources must be listed running the
+ test case.
+
+ Returns:
+ TestCase.EX_OK
+ """
+ return TestCase.EX_OK
+
+ def clean(self):
+ """Clean the resources.
+
+ It can be overriden if resources must be deleted after
+ running the test case.
+ """
+
+
+class OSGCTestCase(TestCase):
+ """Model for single test case which requires an OpenStack Garbage
+ Collector."""
+
+ __logger = logging.getLogger(__name__)
+
+ def create_snapshot(self):
+ """Create a snapshot listing the OpenStack resources.
+
+ Returns:
+ TestCase.EX_OK if os_snapshot.main() returns 0.
+ TestCase.EX_RUN_ERROR otherwise.
+ """
+ try:
+ assert os_snapshot.main() == 0
+ self.__logger.info("OpenStack resources snapshot created")
+ return TestCase.EX_OK
+ except Exception: # pylint: disable=broad-except
+ self.__logger.exception("Cannot create the snapshot")
+ return TestCase.EX_RUN_ERROR
+
+ def clean(self):
+ """Clean the OpenStack resources."""
+ try:
+ assert os_clean.main() == 0
+ self.__logger.info("OpenStack resources cleaned")
+ except Exception: # pylint: disable=broad-except
+ self.__logger.exception("Cannot clean the OpenStack resources")
Code Review / functest.git / blobdiff