2
0
mirror of https://gitlab.isc.org/isc-projects/bind9 synced 2025-08-31 14:35:26 +00:00

Split up waiting for match to a separate WatchLog method

To allow re-use in upcoming functions, isolate the line matching logic
into a separate function. Use an instance-wide deadline attribute, which
is set by the calling function.
This commit is contained in:
Nicki Křížek
2025-06-23 14:37:09 +02:00
parent 2afb3755b2
commit 365f8b6af6

View File

@@ -98,6 +98,7 @@ class WatchLog(abc.ABC):
if timeout <= 0.0:
raise WatchLogException("timeout must be greater than 0")
self._timeout = timeout
self._deadline = 0.0
def _readline(self) -> Optional[str]:
"""
@@ -250,16 +251,22 @@ class WatchLog(abc.ABC):
"""
regexes = self._prepare_patterns(patterns)
self._wait_function_called = True
self._deadline = time.monotonic() + self._timeout
deadline = time.monotonic() + self._timeout
while time.monotonic() < deadline:
return self._wait_for_match(regexes)
def _wait_for_match(self, regexes: List[Pattern]) -> Match:
while time.monotonic() < self._deadline:
for line in self._readlines():
for regex in regexes:
match = regex.search(line)
if match:
return match
time.sleep(0.1)
raise TimeoutError(f"Timeout reached watching {self._path} for {patterns}")
raise TimeoutError(
f"Timeout reached watching {self._path} for "
f"{' | '.join([regex.pattern for regex in regexes])}"
)
def __enter__(self) -> Any:
self._fd = open(self._path, encoding="utf-8")