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

_result_to_str = {UNKNOWN: 'Inconnu',
                  SUCCESS: 'Succès',
                  FAILURE: 'Echec',
                  ERROR: 'Erreur'}

def _linenumber(m):
    try:
        _, line_no = inspect.findsource(m)
        return line_no
    except AttributeError:
        return -1

class CheckingException(Exception):
    pass

class TestError():
    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():
    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 _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():
    def _started_test(self, test):
        pass
    
    def _ended_test(self, test):
        pass

class BaseChecker():
    
    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)