1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
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:
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):
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()
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
|
