summaryrefslogtreecommitdiffhomepage
path: root/test/unit/log.py
diff options
context:
space:
mode:
authorAndrei Zeliankou <zelenkov@nginx.com>2023-05-29 14:23:52 +0100
committerAndrei Zeliankou <zelenkov@nginx.com>2023-05-29 14:23:52 +0100
commitf55818059c01ff9e61bee8107ed1389fe272a787 (patch)
tree27137343940f9212c5a22599c8d829f8759e84f1 /test/unit/log.py
parentb034bf67034c4f0e966ebd207ba2f407f6f15fa8 (diff)
downloadunit-f55818059c01ff9e61bee8107ed1389fe272a787.tar.gz
unit-f55818059c01ff9e61bee8107ed1389fe272a787.tar.bz2
Tests: Log reworked.
All log-related code moved to the log.py.
Diffstat (limited to 'test/unit/log.py')
-rw-r--r--test/unit/log.py86
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}'