From: Nicki Křížek Date: Mon, 16 Jun 2025 16:39:56 +0000 (+0200) Subject: Unify the WatchLog.wait_for_line/s() API X-Git-Tag: v9.21.11~40^2~8 X-Git-Url: http://git.ipfire.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=5840908ead3a1f1775bda6690c820bbcbfd9d849;p=thirdparty%2Fbind9.git Unify the WatchLog.wait_for_line/s() API Rather than using two distinct functions for matching either one pattern (wait_for_line()), or any of multiple patterns (wait_for_lines()), use a single function that handles both in the same way. Extend the wait_for_line() API: 1. To allow for usage of one or more FlexPatterns, i.e. either plain strings to be matched verbatim, or regular expressions. Both can be used interchangeably to provide the caller to write simple and readable test code, while allowing for increased complexity to allow special cases. 2. Always return the regex match, which allows the caller to identify which line was matched, as well as to extract any additional information, such as individual regex groups. --- diff --git a/bin/tests/system/isctest/log/watchlog.py b/bin/tests/system/isctest/log/watchlog.py index 8173c92c564..9bdcecd8abc 100644 --- a/bin/tests/system/isctest/log/watchlog.py +++ b/bin/tests/system/isctest/log/watchlog.py @@ -9,13 +9,17 @@ # See the COPYRIGHT file distributed with this work for additional # information regarding copyright ownership. -from typing import Iterator, Optional, TextIO, Dict, Any, Union, Pattern +from typing import Iterator, Optional, TextIO, Any, List, Union, Pattern, Match import abc import os +import re import time +FlexPattern = Union[str, Pattern] + + class WatchLogException(Exception): pass @@ -137,14 +141,44 @@ class WatchLog(abc.ABC): return yield line - def wait_for_line(self, string: str) -> None: + def _prepare_patterns( + self, strings: Union[FlexPattern, List[FlexPattern]] + ) -> List[Pattern]: + """ + Convert a mix of string(s) and/or pattern(s) into a list of patterns. + + Any strings are converted into regular expression patterns that match + the string verbatim. """ - Block execution until a line containing the provided `string` appears - in the log file. Return `None` once the line is found or raise a - `TimeoutError` after timeout if `string` does not appear in the log - file (strings and regular expressions are supported). (Catching this - exception is discouraged as it indicates that the test code did not - behave as expected.) + patterns = [] + if not isinstance(strings, list): + strings = [strings] + for string in strings: + if isinstance(string, Pattern): + patterns.append(string) + elif isinstance(string, str): + pattern = re.compile(re.escape(string)) + patterns.append(pattern) + else: + raise WatchLogException( + "only string and re.Pattern allowed for matching" + ) + return patterns + + def wait_for_line(self, patterns: Union[FlexPattern, List[FlexPattern]]) -> Match: + """ + Block execution until any line of interest appears in the log file. + + `patterns` accepts one value or a list of values, with each value being + either a regular expression pattern, or a string which should be + matched verbatim (without interpreting it as a regular expression). + + If any of the patterns is found anywhere within a line in the log file, + return the match, allowing access to the matched line, the regex + groups, and the regex which matched. See re.Match for more. + + A `TimeoutError` is raised if the function fails to find any of the + `patterns` in the allotted time. Recommended use: @@ -152,13 +186,27 @@ class WatchLog(abc.ABC): import isctest def test_foo(servers): + with servers["ns1"].watch_log_from_start() as watcher: + watcher.wait_for_line("all zones loaded") + + pattern = re.compile(r"next key event in ([0-9]+) seconds") with servers["ns1"].watch_log_from_here() as watcher: # ... do stuff here ... - watcher.wait_for_line("foo bar") + match = watcher.wait_for_line(pattern) + seconds = int(match.groups(1)) + + strings = [ + "freezing zone", + "thawing zone", + ] + with servers["ns1"].watch_log_from_here() as watcher: + # ... do stuff here ... + match = watcher.wait_for_line(strings) + line = match.string ``` - One of `wait_for_line()` or `wait_for_lines()` must be called exactly - once for every `WatchLogFrom*` instance. + `wait_for_line()` must be called exactly once for every `WatchLog` + instance. >>> # For `WatchLogFromStart`, `wait_for_line()` returns without >>> # raising an exception as soon as the line being looked for appears @@ -166,101 +214,55 @@ class WatchLog(abc.ABC): >>> # after the `with` statement is reached. >>> import tempfile >>> with tempfile.NamedTemporaryFile("w") as file: - ... print("foo", file=file, flush=True) + ... print("foo bar baz", file=file, flush=True) ... with WatchLogFromStart(file.name) as watcher: - ... retval = watcher.wait_for_line("foo") - >>> print(retval) - None + ... match = watcher.wait_for_line("bar") + >>> print(match.string.strip()) + foo bar baz >>> with tempfile.NamedTemporaryFile("w") as file: ... with WatchLogFromStart(file.name) as watcher: - ... print("foo", file=file, flush=True) - ... retval = watcher.wait_for_line("foo") - >>> print(retval) - None + ... print("foo bar baz", file=file, flush=True) + ... match = watcher.wait_for_line("bar") + >>> print(match.group(0)) + bar >>> # For `WatchLogFromHere`, `wait_for_line()` only returns without >>> # raising an exception if the string being looked for appears in >>> # the log file after the `with` statement is reached. >>> import tempfile >>> with tempfile.NamedTemporaryFile("w") as file: - ... print("foo", file=file, flush=True) + ... print("foo bar baz", file=file, flush=True) ... with WatchLogFromHere(file.name, timeout=0.1) as watcher: - ... watcher.wait_for_line("foo") #doctest: +ELLIPSIS + ... watcher.wait_for_line("bar") #doctest: +ELLIPSIS Traceback (most recent call last): ... TimeoutError: Timeout reached watching ... >>> with tempfile.NamedTemporaryFile("w") as file: - ... print("foo", file=file, flush=True) + ... print("foo bar baz", file=file, flush=True) ... with WatchLogFromHere(file.name) as watcher: - ... print("foo", file=file, flush=True) - ... retval = watcher.wait_for_line("foo") - >>> print(retval) - None - """ - return self._wait_for({string: None}) - - def wait_for_lines(self, strings: Dict[Union[str, Pattern], Any]) -> None: - """ - Block execution until a line of interest appears in the log file. This - function is a "multi-match" variant of `wait_for_line()` which is - useful when some action may cause several different (mutually - exclusive) messages to appear in the log file. - - `strings` is a `dict` associating each string to look for with the - value this function should return when that string is found in the log - file (strings and regular expressions are supported). If none of the - `strings` being looked for appear in the log file after timeout, a - `TimeoutError` is raised. (Catching this exception is discouraged as - it indicates that the test code did not behave as expected.) - - Since `strings` is a `dict` and preserves key order (in CPython 3.6 as - implementation detail, since 3.7 by language design), each line is - checked against each key in order until the first match. Values provided - in the `strings` dictionary (i.e. values which this function is expected - to return upon a successful match) can be of any type. - - Recommended use: - - ```python - import isctest - - def test_foo(servers): - triggers = { - "message A": "value returned when message A is found", - "message B": "value returned when message B is found", - } - with servers["ns1"].watch_log_from_here() as watcher: - # ... do stuff here ... - retval = watcher.wait_for_lines(triggers) - ``` - - One of `wait_for_line()` or `wait_for_lines()` must be called exactly - once for every `WatchLogFromHere` instance. + ... print("bar qux", file=file, flush=True) + ... match = watcher.wait_for_line("bar") + >>> print(match.string.strip()) + bar qux >>> # Different values must be returned depending on which line is >>> # found in the log file. >>> import tempfile - >>> triggers = {"foo": 42, "bar": 1337} + >>> patterns = [re.compile(r"bar ([0-9])"), "qux"] >>> with tempfile.NamedTemporaryFile("w") as file: - ... print("foo", file=file, flush=True) + ... print("foo bar 3", file=file, flush=True) ... with WatchLogFromStart(file.name) as watcher: - ... retval1 = watcher.wait_for_lines(triggers) + ... match1 = watcher.wait_for_line(patterns) ... with WatchLogFromHere(file.name) as watcher: - ... print("bar", file=file, flush=True) - ... retval2 = watcher.wait_for_lines(triggers) - >>> print(retval1) - 42 - >>> print(retval2) - 1337 + ... print("baz qux", file=file, flush=True) + ... match2 = watcher.wait_for_line(patterns) + >>> print(match1.group(1)) + 3 + >>> print(match2.group(0)) + qux """ - return self._wait_for(strings) + regexes = self._prepare_patterns(patterns) - def _wait_for(self, patterns: Dict[Union[str, Pattern], Any]) -> Any: - """ - Block execution until one of the `strings` being looked for appears in - the log file. Raise a `TimeoutError` if none of the `strings` being - looked for are found in the log file after timeout. - """ if self._wait_function_called: raise WatchLogException("wait_for_*() was already called") self._wait_function_called = True @@ -268,17 +270,12 @@ class WatchLog(abc.ABC): deadline = time.monotonic() + self._timeout while time.monotonic() < deadline: for line in self._readlines(): - for string, retval in patterns.items(): - if isinstance(string, Pattern) and string.search(line): - return retval - if isinstance(string, str) and string in line: - return retval + for regex in regexes: + match = regex.search(line) + if match: + return match time.sleep(0.1) - raise TimeoutError( - "Timeout reached watching {} for {}".format( - self._path, list(patterns.keys()) - ) - ) + raise TimeoutError(f"Timeout reached watching {self._path} for {patterns}") def __enter__(self) -> Any: self._fd = open(self._path, encoding="utf-8")