-
Notifications
You must be signed in to change notification settings - Fork 893
/
util.py
116 lines (89 loc) · 2.87 KB
/
util.py
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
"""Common functions for e2e testing."""
from __future__ import annotations
import os
import re
from contextlib import contextmanager
from time import sleep, time
from typing import TYPE_CHECKING, Any, Callable, Iterator
if TYPE_CHECKING:
from pathlib import Path
@contextmanager
def chdir(path: Path) -> Iterator:
"""Context manager to help execute code in a different directory.
Args:
path: directory to change to.
Yields:
None
"""
old_pwd = os.getcwd()
os.chdir(str(path))
try:
yield
finally:
os.chdir(old_pwd)
class WaitForException(Exception):
pass
def wait_for(
func: Callable,
timeout_: int = 10,
print_error: bool = False,
sleep_for: int = 1,
**kwargs,
) -> Any:
"""Run specified function until it returns expected result until timeout.
Args:
func: Specified function.
timeout_: Time out in seconds. Defaults to 10.
print_error: whether any exceptions raised should be printed.
Defaults to False.
sleep_for: Execute func every specified number of seconds.
Defaults to 1.
**kwargs: Arguments to be passed to func.
Raises:
WaitForException: if func doesn't return expected result within the
specified time.
Returns:
Function return.
"""
end = time() + timeout_
while time() <= end:
try:
result = func(**kwargs)
return result
except Exception as err:
if print_error:
print(err)
sleep(sleep_for)
raise WaitForException(
f"func: {func}, didn't return within specified timeout: {timeout_}"
)
def parse_csv(text: str) -> list[str]:
"""Parse comma separated **double quoted** strings in behave steps
Args:
text: double quoted comma separated string
Returns:
List of string tokens
"""
return re.findall(r"\"(.+?)\"\s*,?", text)
def clean_up_log(stdout: str) -> str:
"""
Cleans up log output by removing duplicate lines, extra whitespaces,
and log levels (INFO, WARNING, ERROR) along with .py filenames.
Args:
stdout (str): The log output to be cleaned.
Returns:
str: Cleaned log output without unnecessary information.
"""
cleaned_lines = []
already_extracted = set()
for line in stdout.split("\n"):
if any(word in line for word in ["WARNING", "INFO", "ERROR"]):
# Remove log levels and .py filenames
cleaned_line = re.sub(r"\b(INFO|WARNING|ERROR)\b|\s+\w+\.py:\d+", "", line)
cleaned_lines.append(cleaned_line.strip())
already_extracted.add(line)
elif line not in already_extracted:
cleaned_lines.append(line)
cleaned_output = "\n".join(cleaned_lines)
cleaned_output = re.sub(r"\s+", " ", cleaned_output)
return cleaned_output.strip()