diff options
author | Andrei Zeliankou <zelenkov@nginx.com> | 2023-05-29 14:23:52 +0100 |
---|---|---|
committer | Andrei Zeliankou <zelenkov@nginx.com> | 2023-05-29 14:23:52 +0100 |
commit | f55818059c01ff9e61bee8107ed1389fe272a787 (patch) | |
tree | 27137343940f9212c5a22599c8d829f8759e84f1 /test/unit/log.py | |
parent | b034bf67034c4f0e966ebd207ba2f407f6f15fa8 (diff) | |
download | unit-f55818059c01ff9e61bee8107ed1389fe272a787.tar.gz unit-f55818059c01ff9e61bee8107ed1389fe272a787.tar.bz2 |
Tests: Log reworked.
All log-related code moved to the log.py.
Diffstat (limited to '')
-rw-r--r-- | test/unit/log.py | 86 |
1 files changed, 79 insertions, 7 deletions
diff --git a/test/unit/log.py b/test/unit/log.py index f984d7a1..c98054d5 100644 --- a/test/unit/log.py +++ b/test/unit/log.py @@ -1,23 +1,95 @@ +import os +import re +import sys + +from unit.option import option + UNIT_LOG = 'unit.log' +def print_log_on_assert(func): + def inner_function(*args, **kwargs): + try: + func(*args, **kwargs) + except AssertionError as exception: + Log.print_log(*args, **kwargs) + raise exception + + return inner_function + + class Log: - temp_dir = None pos = {} + @staticmethod + @print_log_on_assert + def check_alerts(log=None): + if log is None: + log = Log.read(encoding='utf-8') + + found = False + alerts = re.findall(r'.+\[alert\].+', log) + + if alerts: + found = True + + if option.detailed: + print('\nAll alerts/sanitizer errors found in log:') + _ = [print(alert) for alert in alerts] + + if option.skip_alerts: + for skip in option.skip_alerts: + alerts = [al for al in alerts if re.search(skip, al) is None] + + assert not alerts, 'alert(s)' + + if not option.skip_sanitizer: + sanitizer_errors = re.findall('.+Sanitizer.+', log) + + assert not sanitizer_errors, 'sanitizer error(s)' + + if found and option.detailed: + print('skipped.') + + @staticmethod + def get_path(name=UNIT_LOG): + return f'{option.temp_dir}/{name}' + + @staticmethod def open(name=UNIT_LOG, encoding=None): - f = open(Log.get_path(name), 'r', encoding=encoding, errors='ignore') - f.seek(Log.pos.get(name, 0)) + file = open(Log.get_path(name), 'r', encoding=encoding, errors='ignore') + file.seek(Log.pos.get(name, 0)) + + return file + + @staticmethod + def print_log(log=None): + Log.print_path() - return f + if option.print_log: + os.set_blocking(sys.stdout.fileno(), True) + sys.stdout.flush() + if log is None: + log = Log.read(encoding='utf-8') + + sys.stdout.write(log) + + @staticmethod + def print_path(): + print(f'Path to {UNIT_LOG}:\n{Log.get_path()}\n') + + @staticmethod + def read(*args, **kwargs): + with Log.open(*args, **kwargs) as file: + return file.read() + + @staticmethod def set_pos(pos, name=UNIT_LOG): Log.pos[name] = pos + @staticmethod def swap(name): pos = Log.pos.get(UNIT_LOG, 0) Log.pos[UNIT_LOG] = Log.pos.get(name, 0) Log.pos[name] = pos - - def get_path(name=UNIT_LOG): - return f'{Log.temp_dir}/{name}' |