unitgrade2.py

"""
git add . && git commit -m "Options" && git push &&  pip install git+ssh://git@gitlab.compute.dtu.dk/tuhe/unitgrade.git --upgrade
"""
import numpy as np
import sys
import re
import threading
import tqdm
import pickle
import os
from io import StringIO
import io
from unittest.runner import _WritelnDecorator
from typing import Any
import inspect
import textwrap
import colorama
from colorama import Fore
from functools import _make_key, RLock
from collections import namedtuple
import unittest
import time

_CacheInfo = namedtuple("CacheInfo", ["hits", "misses", "maxsize", "currsize"])

colorama.init(autoreset=True)  # auto resets your settings after every output

def gprint(s):
    print(f"{Fore.GREEN}{s}")

myround = lambda x: np.round(x)  # required.
msum = lambda x: sum(x)
mfloor = lambda x: np.floor(x)


def setup_dir_by_class(C, base_dir):
    name = C.__class__.__name__
    return base_dir, name


class Logger(object):
    def __init__(self, buffer):
        assert False
        self.terminal = sys.stdout
        self.log = buffer

    def write(self, message):
        self.terminal.write(message)
        self.log.write(message)

    def flush(self):
        # this flush method is needed for python 3 compatibility.
        pass


class Capturing(list):
    def __init__(self, *args, stdout=None, unmute=False, **kwargs):
        self._stdout = stdout
        self.unmute = unmute
        super().__init__(*args, **kwargs)

    def __enter__(self, capture_errors=True):  # don't put arguments here.
        self._stdout = sys.stdout if self._stdout == None else self._stdout
        self._stringio = StringIO()
        if self.unmute:
            sys.stdout = Logger(self._stringio)
        else:
            sys.stdout = self._stringio

        if capture_errors:
            self._sterr = sys.stderr
            sys.sterr = StringIO()  # memory hole it
        self.capture_errors = capture_errors
        return self

    def __exit__(self, *args):
        self.extend(self._stringio.getvalue().splitlines())
        del self._stringio  # free up some memory
        sys.stdout = self._stdout
        if self.capture_errors:
            sys.sterr = self._sterr


class Capturing2(Capturing):
    def __exit__(self, *args):
        lines = self._stringio.getvalue().splitlines()
        txt = "\n".join(lines)
        numbers = extract_numbers(txt)
        self.extend(lines)
        del self._stringio  # free up some memory
        sys.stdout = self._stdout
        if self.capture_errors:
            sys.sterr = self._sterr

        self.output = txt
        self.numbers = numbers


# @classmethod
# class OrderedClassMembers(type):
#     def __prepare__(self, name, bases):
#         assert False
#         return collections.OrderedDict()
#
#     def __new__(self, name, bases, classdict):
#         ks = list(classdict.keys())
#         for b in bases:
#             ks += b.__ordered__
#         classdict['__ordered__'] = [key for key in ks if key not in ('__module__', '__qualname__')]
#         return type.__new__(self, name, bases, classdict)


class Report:
    title = "report title"
    version = None
    questions = []
    pack_imports = []
    individual_imports = []
    nL = 120  # Maximum line width

    @classmethod
    def reset(cls):
        for (q, _) in cls.questions:
            if hasattr(q, 'reset'):
                q.reset()

    @classmethod
    def mfile(clc):
        return inspect.getfile(clc)

    def _file(self):
        return inspect.getfile(type(self))

    def _import_base_relative(self):
        if hasattr(self.pack_imports[0], '__path__'):
            root_dir = self.pack_imports[0].__path__._path[0]
        else:
            root_dir = self.pack_imports[0].__file__

        root_dir = os.path.dirname(root_dir)
        relative_path = os.path.relpath(self._file(), root_dir)
        modules = os.path.normpath(relative_path[:-3]).split(os.sep)
        return root_dir, relative_path, modules

    def __init__(self, strict=False, payload=None):
        working_directory = os.path.abspath(os.path.dirname(self._file()))
        self.wdir, self.name = setup_dir_by_class(self, working_directory)
        # self.computed_answers_file = os.path.join(self.wdir, self.name + "_resources_do_not_hand_in.dat")
        for (q, _) in self.questions:
            q.nL = self.nL  # Set maximum line length.

        if payload is not None:
            self.set_payload(payload, strict=strict)

    def main(self, verbosity=1):
        # Run all tests using standard unittest (nothing fancy).
        loader = unittest.TestLoader()
        for q, _ in self.questions:
            start = time.time()  # A good proxy for setup time is to
            suite = loader.loadTestsFromTestCase(q)
            unittest.TextTestRunner(verbosity=verbosity).run(suite)
            total = time.time() - start
            q.time = total

    def _setup_answers(self, with_coverage=False):
        if with_coverage:
            for q, _ in self.questions:
                q._with_coverage = True
                q._report = self

        self.main()  # Run all tests in class just to get that out of the way...
        report_cache = {}
        for q, _ in self.questions:
            # print(self.questions)
            if hasattr(q, '_save_cache'):
                q()._save_cache()
                print("q is", q())
                q()._cache_put('time', q.time) # = q.time
                report_cache[q.__qualname__] = q._cache2
            else:
                report_cache[q.__qualname__] = {'no cache see _setup_answers in unitgrade2.py': True}
        if with_coverage:
            for q, _ in self.questions:
                q._with_coverage = False
        return report_cache

    def set_payload(self, payloads, strict=False):
        for q, _ in self.questions:
            q._cache = payloads[q.__qualname__]


def rm_progress_bar(txt):
    # More robust version. Apparently length of bar can depend on various factors, so check for order of symbols.
    nlines = []
    for l in txt.splitlines():
        pct = l.find("%")
        ql = False
        if pct > 0:
            i = l.find("|", pct + 1)
            if i > 0 and l.find("|", i + 1) > 0:
                ql = True
        if not ql:
            nlines.append(l)
    return "\n".join(nlines)


def extract_numbers(txt):
    # txt = rm_progress_bar(txt)
    numeric_const_pattern = r'[-+]? (?: (?: \d* \. \d+ ) | (?: \d+ \.? ) )(?: [Ee] [+-]? \d+ ) ?'
    rx = re.compile(numeric_const_pattern, re.VERBOSE)
    all = rx.findall(txt)
    all = [float(a) if ('.' in a or "e" in a) else int(a) for a in all]
    if len(all) > 500:
        print(txt)
        raise Exception("unitgrade.unitgrade.py: Warning, too many numbers!", len(all))
    return all


class ActiveProgress():
    def __init__(self, t, start=True, title="my progress bar", show_progress_bar=True, file=None):
        if file == None:
            file = sys.stdout
        self.file = file
        self.t = t
        self._running = False
        self.title = title
        self.dt = 0.01
        self.n = int(np.round(self.t / self.dt))
        self.show_progress_bar = show_progress_bar
        self.pbar = None

        if start:
            self.start()

    def start(self):
        self._running = True
        if self.show_progress_bar:
            self.thread = threading.Thread(target=self.run)
            self.thread.start()
        self.time_started = time.time()

    def terminate(self):
        if not self._running:
            raise Exception("Stopping a stopped progress bar. ")
        self._running = False
        if self.show_progress_bar:
            self.thread.join()
        if self.pbar is not None:
            self.pbar.update(1)
            self.pbar.close()
            self.pbar = None

        self.file.flush()
        return time.time() - self.time_started

    def run(self):
        self.pbar = tqdm.tqdm(total=self.n, file=self.file, position=0, leave=False, desc=self.title, ncols=100,
                              bar_format='{l_bar}{bar}| [{elapsed}<{remaining}]')

        for _ in range(self.n - 1):  # Don't terminate completely; leave bar at 99% done until terminate.
            if not self._running:
                self.pbar.close()
                self.pbar = None
                break

            time.sleep(self.dt)
            self.pbar.update(1)

def dprint(first, last, nL, extra = "", file=None, dotsym='.', color='white'):
    if file == None:
        file = sys.stdout

    # ss = self.item_title_print
    # state = "PASS" if success else "FAILED"
    dot_parts = (dotsym * max(0, nL - len(last) - len(first)))
    # if self.show_progress_bar or True:
    print(first + dot_parts, end="", file=file)
    # else:
    # print(dot_parts, end="", file=self.cc.file)
    last += extra
    # if tsecs >= 0.5:
    #     state += " (" + str(tsecs) + " seconds)"
    print(last, file=file)


class UTextResult(unittest.TextTestResult):
    nL = 80
    number = -1  # HAcky way to set question number.
    show_progress_bar = True
    cc = None

    def __init__(self, stream, descriptions, verbosity):
        super().__init__(stream, descriptions, verbosity)
        self.successes = []

    def printErrors(self) -> None:
        self.printErrorList('ERROR', self.errors)
        self.printErrorList('FAIL', self.failures)

    def addError(self, test, err):
        super(unittest.TextTestResult, self).addFailure(test, err)
        self.cc_terminate(success=False)

    def addFailure(self, test, err):
        super(unittest.TextTestResult, self).addFailure(test, err)
        self.cc_terminate(success=False)

    def addSuccess(self, test: unittest.case.TestCase) -> None:
        self.successes.append(test)
        self.cc_terminate()

    def cc_terminate(self, success=True):
        if self.show_progress_bar or True:
            tsecs = np.round(self.cc.terminate(), 2)
            self.cc.file.flush()
            ss = self.item_title_print

            state = "PASS" if success else "FAILED"

            dot_parts = ('.' * max(0, self.nL - len(state) - len(ss)))
            if self.show_progress_bar or True:
                print(self.item_title_print + dot_parts, end="", file=self.cc.file)
            else:
                print(dot_parts, end="", file=self.cc.file)

            if tsecs >= 0.5:
                state += " (" + str(tsecs) + " seconds)"
            print(state, file=self.cc.file)

    def startTest(self, test):
        # j =self.testsRun
        self.testsRun += 1
        # item_title = self.getDescription(test)
        item_title = test.shortDescription()  # Better for printing (get from cache).
        if item_title == None:
            # For unittest framework where getDescription may return None.
            item_title = self.getDescription(test)
        self.item_title_print = " * q%i.%i) %s" % (UTextResult.number + 1, self.testsRun, item_title)
        estimated_time = 10
        if self.show_progress_bar or True:
            self.cc = ActiveProgress(t=estimated_time, title=self.item_title_print, show_progress_bar=self.show_progress_bar, file=sys.stdout)
        else:
            print(self.item_title_print + ('.' * max(0, self.nL - 4 - len(self.item_title_print))), end="")

        self._test = test
        self._stdout = sys.stdout
        sys.stdout = io.StringIO()

    def stopTest(self, test):
        sys.stdout = self._stdout
        super().stopTest(test)

    def _setupStdout(self):
        if self._previousTestClass == None:
            total_estimated_time = 1
            if hasattr(self.__class__, 'q_title_print'):
                q_title_print = self.__class__.q_title_print
            else:
                q_title_print = "<unnamed test. See unitgrade.py>"

            cc = ActiveProgress(t=total_estimated_time, title=q_title_print, show_progress_bar=self.show_progress_bar)
            self.cc = cc

    def _restoreStdout(self):  # Used when setting up the test.
        if self._previousTestClass is None:
            q_time = self.cc.terminate()
            q_time = np.round(q_time, 2)
            sys.stdout.flush()
            if self.show_progress_bar:
                print(self.cc.title, end="")
            print(" " * max(0, self.nL - len(self.cc.title)) + (" (" + str(q_time) + " seconds)" if q_time >= 0.5 else ""))


class UTextTestRunner(unittest.TextTestRunner):
    def __init__(self, *args, **kwargs):
        stream = io.StringIO()
        super().__init__(*args, stream=stream, **kwargs)

    def _makeResult(self):
        # stream = self.stream # not you!
        stream = sys.stdout
        stream = _WritelnDecorator(stream)
        return self.resultclass(stream, self.descriptions, self.verbosity)


def cache(foo, typed=False):
    """ Magic cache wrapper
    https://github.com/python/cpython/blob/main/Lib/functools.py
    """
    maxsize = None
    def wrapper(self, *args, **kwargs):
        key = (self.cache_id(), ("@cache", foo.__name__, _make_key(args, kwargs, typed)))
        if not self._cache_contains(key):
            value = foo(self, *args, **kwargs)
            self._cache_put(key, value)
        else:
            value = self._cache_get(key)
        return value

    return wrapper


def get_hints(ss):
    if ss == None:
        return None
    try:
        ss = textwrap.dedent(ss)
        ss = ss.replace('''"""''', "").strip()
        hints = ["hints:", ]
        j = np.argmax([ss.lower().find(h) for h in hints])
        h = hints[j]
        ss = ss[ss.find(h) + len(h) + 1:]
        ss = "\n".join([l for l in ss.split("\n") if not l.strip().startswith(":")])
        ss = textwrap.dedent(ss)
        ss = ss.strip()
        return ss
    except Exception as e:
        print("bad hints", ss, e)


class UTestCase(unittest.TestCase):
    _outcome = None  # A dictionary which stores the user-computed outcomes of all the tests. This differs from the cache.
    _cache = None  # Read-only cache. Ensures method always produce same result.
    _cache2 = None  # User-written cache.
    _with_coverage = False
    _report = None  # The report used. This is very, very hacky and should always be None. Don't rely on it!

    def capture(self):
        if hasattr(self, '_stdout') and self._stdout is not None:
            file = self._stdout
        else:
            # self._stdout = sys.stdout
            # sys._stdout = io.StringIO()
            file = sys.stdout
        return Capturing2(stdout=file)

    @classmethod
    def question_title(cls):
        """ Return the question title """
        return cls.__doc__.strip().splitlines()[0].strip() if cls.__doc__ is not None else cls.__qualname__

    @classmethod
    def reset(cls):
        print("Warning, I am not sure UTestCase.reset() is needed anymore and it seems very hacky.")
        cls._outcome = None
        cls._cache = None
        cls._cache2 = None

    def _callSetUp(self):
        if self._with_coverage:
            if not hasattr(self._report, 'covcache'):
                self._report.covcache = {}
            import coverage
            self.cov = coverage.Coverage()
            self.cov.start()
        self.setUp()

    def _callTearDown(self):
        self.tearDown()
        if self._with_coverage:
            from pathlib import Path
            from snipper import snipper
            self.cov.stop()
            data = self.cov.get_data()
            base, _, _ = self._report._import_base_relative()
            for file in data.measured_files():
                file = os.path.normpath(file)
                root = Path(base)
                child = Path(file)
                if root in child.parents:
                    with open(child, 'r') as f:
                        s = f.read()
                    lines = s.splitlines()
                    garb = 'GARBAGE'

                    lines2 = snipper.censor_code(lines, keep=True)
                    assert len(lines) == len(lines2)

                    for l in data.contexts_by_lineno(file):
                        if lines2[l].strip() == garb:
                            if self.cache_id() not in self._report.covcache:
                                self._report.covcache[self.cache_id()] = {}

                            rel = os.path.relpath(child, root)
                            cc = self._report.covcache[self.cache_id()]
                            j = 0
                            for j in range(l, -1, -1):
                                if "def" in lines2[j] or "class" in lines2[j]:
                                    break
                            from snipper.snipper import gcoms
                            fun = lines2[j]
                            comments, _ = gcoms("\n".join(lines2[j:l]))
                            if rel not in cc:
                                cc[rel] = {}
                            cc[rel][fun] = (l, "\n".join(comments))
                            self._cache_put((self.cache_id(), 'coverage'), self._report.covcache)

    def shortDescriptionStandard(self):
        sd = super().shortDescription()
        if sd is None:
            sd = self._testMethodName
        return sd

    def shortDescription(self):
        sd = self.shortDescriptionStandard()
        title = self._cache_get((self.cache_id(), 'title'), sd)
        return title if title is not None else sd

    @property
    def title(self):
        return self.shortDescription()

    @title.setter
    def title(self, value):
        self._cache_put((self.cache_id(), 'title'), value)

    def _get_outcome(self):
        if not (self.__class__, '_outcome') or self.__class__._outcome is None:
            self.__class__._outcome = {}
        return self.__class__._outcome

    def _callTestMethod(self, testMethod):
        t = time.time()
        self._ensure_cache_exists()  # Make sure cache is there.
        if self._testMethodDoc is not None:
            self._cache_put((self.cache_id(), 'title'), self.shortDescriptionStandard())

        self._cache2[(self.cache_id(), 'assert')] = {}
        res = testMethod()
        elapsed = time.time() - t
        self._get_outcome()[self.cache_id()] = res
        self._cache_put((self.cache_id(), "time"), elapsed)

    def cache_id(self):
        c = self.__class__.__qualname__
        m = self._testMethodName
        return c, m

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._load_cache()
        self._assert_cache_index = 0

    def _ensure_cache_exists(self):
        if not hasattr(self.__class__, '_cache') or self.__class__._cache == None:
            self.__class__._cache = dict()
        if not hasattr(self.__class__, '_cache2') or self.__class__._cache2 == None:
            self.__class__._cache2 = dict()

    def _cache_get(self, key, default=None):
        self._ensure_cache_exists()
        return self.__class__._cache.get(key, default)

    def _cache_put(self, key, value):
        self._ensure_cache_exists()
        self.__class__._cache2[key] = value

    def _cache_contains(self, key):
        self._ensure_cache_exists()
        return key in self.__class__._cache

    def wrap_assert(self, assert_fun, first, *args, **kwargs):
        # sys.stdout = self._stdout
        key = (self.cache_id(), 'assert')
        if not self._cache_contains(key):
            print("Warning, framework missing", key)
            self.__class__._cache[
                key] = {}  # A new dict. We manually insert it because we have to use that the dict is mutable.
        cache = self._cache_get(key)
        id = self._assert_cache_index
        if not id in cache:
            print("Warning, framework missing cache index", key, "id =", id)
        _expected = cache.get(id, f"Key {id} not found in cache; framework files missing. Please run deploy()")

        # The order of these calls is important. If the method assert fails, we should still store the correct result in cache.
        cache[id] = first
        self._cache_put(key, cache)
        self._assert_cache_index += 1
        assert_fun(first, _expected, *args, **kwargs)

    def assertEqualC(self, first: Any, msg: Any = ...) -> None:
        self.wrap_assert(self.assertEqual, first, msg)

    def _cache_file(self):
        return os.path.dirname(inspect.getfile(self.__class__)) + "/unitgrade/" + self.__class__.__name__ + ".pkl"

    def _save_cache(self):
        # get the class name (i.e. what to save to).
        cfile = self._cache_file()
        if not os.path.isdir(os.path.dirname(cfile)):
            os.makedirs(os.path.dirname(cfile))

        if hasattr(self.__class__, '_cache2'):
            with open(cfile, 'wb') as f:
                pickle.dump(self.__class__._cache2, f)

    # But you can also set cache explicitly.
    def _load_cache(self):
        if self._cache is not None:  # Cache already loaded. We will not load it twice.
            return
            # raise Exception("Loaded cache which was already set. What is going on?!")
        cfile = self._cache_file()
        if os.path.exists(cfile):
            try:
                with open(cfile, 'rb') as f:
                    data = pickle.load(f)
                self.__class__._cache = data
            except Exception as e:
                print("Bad cache", cfile)
                print(e)
        else:
            print("Warning! data file not found", cfile)

    def _feedErrorsToResult(self, result, errors):
        """ Use this to show hints on test failure. """
        if not isinstance(result, UTextResult):
            er = [e for e, v in errors if v != None]

            if len(er) > 0:
                hints = []
                key = (self.cache_id(), 'coverage')
                if self._cache_contains(key):
                    CC = self._cache_get(key)
                    for id in CC:
                        if id == self.cache_id():
                            cl, m = id
                            gprint(f"> An error occured while solving: {cl}.{m}. The files/methods you need to edit are:")  # For the test {id} in {file} you should edit:")
                            for file in CC[id]:
                                rec = CC[id][file]
                                gprint(f">   * {file}")
                                for l in rec:
                                    _, comments = CC[id][file][l]
                                    hint = get_hints(comments)

                                    if hint != None:
                                        hints.append(hint)
                                    gprint(f">      - {l}")

                er = er[0]
                doc = er._testMethodDoc
                if doc is not None:
                    hint = get_hints(er._testMethodDoc)
                    if hint is not None:
                        hints = [hint] + hints
                if len(hints) > 0:
                    gprint("> Hints:")
                    gprint(textwrap.indent("\n".join(hints), ">   "))

        super()._feedErrorsToResult(result, errors)

    def startTestRun(self):
        # print("asdfsdaf 11", file=sys.stderr)
        super().startTestRun()
        # print("asdfsdaf")

    def _callTestMethod(self, method):
        # print("asdfsdaf")
        super()._callTestMethod(method)


def hide(func):
    return func


def makeRegisteringDecorator(foreignDecorator):
    """
        Returns a copy of foreignDecorator, which is identical in every
        way(*), except also appends a .decorator property to the callable it
        spits out.
    """

    def newDecorator(func):
        # Call to newDecorator(method)
        # Exactly like old decorator, but output keeps track of what decorated it
        R = foreignDecorator(func)  # apply foreignDecorator, like call to foreignDecorator(method) would have done
        R.decorator = newDecorator  # keep track of decorator
        # R.original = func         # might as well keep track of everything!
        return R

    newDecorator.__name__ = foreignDecorator.__name__
    newDecorator.__doc__ = foreignDecorator.__doc__
    return newDecorator

hide = makeRegisteringDecorator(hide)

def methodsWithDecorator(cls, decorator):
    """
        Returns all methods in CLS with DECORATOR as the
        outermost decorator.

        DECORATOR must be a "registering decorator"; one
        can make any decorator "registering" via the
        makeRegisteringDecorator function.

        import inspect
        ls = list(methodsWithDecorator(GeneratorQuestion, deco))
        for f in ls:
            print(inspect.getsourcelines(f) ) # How to get all hidden questions.
    """
    for maybeDecorated in cls.__dict__.values():
        if hasattr(maybeDecorated, 'decorator'):
            if maybeDecorated.decorator == decorator:
                print(maybeDecorated)
                yield maybeDecorated
# 817