mncheck/core/checking.py

'''

    A simplified version of the unittest module, 
    adapted to run unit tests on data sets instead of code.

@author: olivier.massot, 2018
'''
import inspect
import re
import sys
import traceback

UNKNOWN = 0
SUCCESS = 1
FAILURE = 2
ERROR = 3


def _linenumber(obj_):
    """ uses the inspect module to find the line number of an object in the module """
    try:
        _, line_no = inspect.findsource(obj_)
        return line_no
    except AttributeError:
        return -1


class CheckingException(Exception):
    """ Base class for checking exceptions """
    pass

class TestError():
    """ Error logged during a test """
    def __init__(self, message, info={}, critical=False):
        self.message = message
        self.info = info
        self.critical = critical

    def __repr__(self):
        return f"TestError[message='{self.message}'; info={self.info}; critical={self.critical}]"


class TestResult():
    """ Result of a test """
    _result_to_str = {UNKNOWN: 'Inconnu',
                  SUCCESS: 'Succès',
                  FAILURE: 'Echec',
                  ERROR: 'Erreur'}

    def __init__(self, test):
        self._test = test
        self._name = ""
        self._status = SUCCESS
        self.errors = []
        self._exc_info = None

    @property
    def name(self):
        return self._name or self._test.__name__[5:]

    @property
    def title(self):
        try:
            return self._test.__doc__.split("\n")[0].strip()
        except AttributeError:
            return self.name

    @property
    def description(self):
        try:
            return re.sub(" +", " ",
                          self._test.__doc__.strip(),
                          flags=re.MULTILINE)  # @UndefinedVariable
        except AttributeError:
            return ""

    @property
    def status(self):
        return self._status

    @property
    def status_str(self):
        return TestResult._result_to_str[self._status]

    def __repr__(self):
        return f"TestResult[title='{self.title}'; " \
                "status={self.status}; " \
                "name={self.name}; " \
                "method={self._test.__name__}; " \
                "errors_count={len(self.errors)}]"

    def log_error(self, message, info={}, critical=False):
        self._status = FAILURE
        error = TestError(message, info, critical)
        self.errors.append(error)

    def log_exception(self, message, info={}):
        self._status = ERROR
        error = TestError(message, info)
        self.errors.append(error)

    def handle_exception(self, exc_info):
        typ, value, trace = exc_info
        self.log_exception("Une erreur s'est produite: {}".format(typ.__name__),
                           {"exc_info": "{}\n{}\n{}".format(typ.__name__, value, 
                                                            ''.join(traceback.format_tb(trace)))})


class Comlink():
    """ Class used to provide pointers to the checker events """
    def _started_test(self, test):
        pass

    def _ended_test(self, test):
        pass


class BaseChecker():
    """ Base class for checkers """
    def __init__(self):
        self._test_running = None
        self.comlink = Comlink()
        self.tests = sorted([m for _, m \
                             in inspect.getmembers(self, predicate=inspect.ismethod) \
                             if m.__name__[:5] == 'test_'], key=_linenumber)

    def setUp(self):
        pass

    def tearDown(self):
        pass

    def log_error(self, message, **info):
        self._test_running.log_error(message, info)

    def log_critical(self, message, **info):
        self._test_running.log_error(f"[CRITIQUE] {message}", info, critical=True)

    def run(self, rxfilter="", dry=False):

        # 'rxfilter' allow to filter the tests to run with a regex
        # if 'dry' is set to True, setUp and tearDown are skipped
        # both are used for testing purpose

        tests_results = []

        for test in self.tests:

            if rxfilter and not re.fullmatch(rxfilter, 
                                             test.__name__, 
                                             re.IGNORECASE):  # @UndefinedVariable
                continue

            result = TestResult(test)

            self._test_running = result
            self.comlink._started_test(result)

            try:
                if not dry:
                    self.setUp()

                test()

                if not dry:
                    self.tearDown()

            except CheckingException as e:
                result.log_exception(str(e))
            except:
                result.handle_exception(sys.exc_info())

            tests_results.append(result)

            self.comlink._ended_test(result)

            if any(err.critical for err in result.errors):
                break

        return tests_results


if __name__ == '__main__':

    class ExampleChecker(BaseChecker):

        def test_c(self):
            """ Test 1 """
            for i in range(10):
                self.log_error(f"error-{i}", i=i)

        def test_b(self):
            """ Test 2
            some longer description """
            raise Exception("bla bla")

    ch = ExampleChecker()
    results = ch.run()

    for r in results:
        print(r)
        for e in r.errors:
            print(e)