blob: 6c41f58c15ec9b75bd2ac5a58e9390019e643ae4 [file] [log] [blame]
from typing import Any, Callable, Dict, List
from django.core.cache import cache
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
if __name__ == "__main__":
pass