testrail_bot/control/utils.py

import logging
from typing import Any, Callable, Dict, List

from django.core.cache import cache
from django.db.models import Model
from parse import parse


def parse_title(test_name):
    # Sometimes id can be without the closing ] symbol
    if "[" in test_name and "]" not in test_name:
        test_name += "]"
    token_count = test_name.split(".").__len__()

    if test_name.startswith("=="):
        return test_name

    if test_name.startswith(".setUp") or test_name.startswith(".tearDown"):
        fmt = "{test_title}(" + "{}." * (token_count - 2) + "{class_name})"
        r = parse(fmt, test_name)
        return f"{r['class_name']}.{r['test_title']}".strip()
    try:
        fmt = "{}." * (token_count - 2) + "{class_name}.{test_title}[{id}]"
        r = parse(fmt, test_name)
        return f"{r['test_title']}[{r['id']}]"
    except TypeError:
        # return file_name.test_class.test_name in other complicated cases
        return ".".join(test_name.split(".")[:3])


def short_names_for_dict(_dict):
    __dict = {}
    for _k in _dict.keys():
        __k = parse_title(_k)
        # Replace only those keys which are absent in the dict or empty
        # (defined as "No result found")
        if __dict.get(__k) == "No result found" or not __dict.get(__k):
            __dict[__k] = _dict[_k]
    return __dict


def get_dict_diff(
    dict1: dict, dict2: dict, compare_by_key=None
) -> Dict[str, List]:
    all_keys = sorted(set(list(dict1.keys()) + list(dict2.keys())))

    result = dict()
    for k in all_keys:
        if compare_by_key:
            if dict1.get(k, {}).get(compare_by_key) == dict2.get(k, {}).get(
                compare_by_key
            ):
                continue
        else:
            if dict1.get(k) == dict2.get(k):
                continue
        result[k] = [dict1.get(k), dict2.get(k)]
    return result


def replace_all(text: str, olds: str, new: str) -> str:
    r = text
    for _s in olds:
        r = r.replace(_s, new)
    return r


def cached(
    timeout: int = None,
    condition_for_endless_cache: Callable = lambda x: False,
) -> Callable:
    """
    :param timeout: (in seconds) usage accordingly
    https://docs.djangoproject.com/en/4.2/topics/cache/#basic-usage
    :param condition_for_endless_cache: Callable should return boolean.
    Checks a result of function. If Result meets requirements of condition
    then the endless timeout will be set. Or it will use provided timeout
    otherwise

    :return: decorator
    """

    def decorator(func: Callable) -> Callable:
        def wrapper(*args, **kwargs) -> Any:
            cache_key = f"{func.__name__}_{args}_{kwargs}"
            cache_key = replace_all(cache_key, "{}()'\" .,:", "_")
            cached_value = cache.get(cache_key)
            if cached_value is None:
                print(f"{func.__name__} MISS")
                result = func(*args, **kwargs)
                _timeout = (
                    None if condition_for_endless_cache(result) else timeout
                )

                cache.set(cache_key, result, timeout=_timeout)
                return result
            print(f"{func.__name__} hit")

            # # FIXME Assert to test the caching mechanism
            # _result = func(*args, **kwargs)
            # for d in difflib.ndiff(str(result), str(_result)):
            #     if not d.startswith(" "):
            #         print(d)
            # assert result == _result
            # # ENDFIXME

            return cached_value

        return wrapper

    return decorator


class DBHandler(logging.Handler):

    def __init__(self, storage: Model) -> None:
        logging.Handler.__init__(self)
        self.storage = storage

    def emit(self, record: logging.LogRecord) -> None:
        msg = self.format(record)
        color = "black"
        if record.levelno == logging.ERROR:
            color = "red"
        if record.levelno == logging.DEBUG:
            color = "grey"
        self.storage.text += f"<a style='color:{color}'>{msg} </a>\n"
        self.storage.save()
        print(self.storage.text)


def DBlogger(name: str, storage: Model) -> logging.Logger:
    _log = logging.getLogger(name)
    _log.setLevel(logging.DEBUG)
    formatter = logging.Formatter(
        "[%(asctime)s] %(message)s", "%d %b %H:%M:%S"
    )
    db_h = DBHandler(storage)
    db_h.setFormatter(formatter)
    _log.addHandler(db_h)
    return _log


if __name__ == "__main__":
    pass