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 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 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):
        return self._test.__doc__.split("\n")[0].strip() 

    @property
    def description(self):
        return self._test.__doc__.strip() 
    
    @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 handle_exception(self, exc_info):
        self._status = ERROR
        typ, value, trace = exc_info
        error = TestError("Une erreur inconnue s'est produite, veuillez consulter les fichiers de journalisation.", 
                          {"exc_info": "{}\n{}\n{}".format(typ.__name__, value, ''.join(traceback.format_tb(trace)))})
        self.errors.append(error)

    def grouped_errors(self):
        res = {}
        for err in self.errors:
            if not err.message in res:
                res[err.message] = []
            res[err.message].append(err)
        return res

class BaseChecker():
    
    def __init__(self):
        self._test_running = None
        
    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):
        
        tests_results = []
        
        tests = sorted([m for _, m in inspect.getmembers(self, predicate=inspect.ismethod) if m.__name__[:5] == 'test_'], key=_linenumber)
        
        for test in tests:
            
            result = TestResult(test)
            
            self._test_running = result
            
            self.setUp()
            
            try:
                test()
            except:
                result.handle_exception(sys.exc_info())
            
            self.tearDown()

            tests_results.append(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)