# See the COPYRIGHT file distributed with this work for additional
# information regarding copyright ownership.
-from typing import Iterator, Optional, TextIO, Dict, Any
+from typing import Iterator, Optional, TextIO, Dict, Any, Union, Pattern
import abc
import os
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` seconds (default: 10) if `string` does
- not appear in the log file. (Catching this exception is discouraged as
- it indicates that the test code did not behave as expected.)
+ 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.)
Recommended use:
"""
return self._wait_for({string: None}, timeout)
- def wait_for_lines(self, strings: Dict[str, Any], timeout: int = 10) -> None:
+ def wait_for_lines(
+ self, strings: Dict[Union[str, Pattern], Any], timeout: int = 10
+ ) -> 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
`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. If none of the `strings` being looked for appear in the log file
- after `timeout` seconds, a `TimeoutError` is raised.
- (Catching this exception is discouraged as it indicates that the test
- code did not behave as expected.)
+ file (strings and regular expressions are supported). If none of the
+ `strings` being looked for appear in the log file after `timeout`
+ seconds, 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
"""
return self._wait_for(strings, timeout)
- def _wait_for(self, strings: Dict[str, Any], timeout: int) -> Any:
+ def _wait_for(self, patterns: Dict[Union[str, Pattern], Any], timeout: int) -> 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
else:
line = leftover + line
leftover = ""
- for string, retval in strings.items():
- if string in line:
+ 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
time.sleep(0.1)
raise TimeoutError(
"Timeout reached watching {} for {}".format(
- self._path, list(strings.keys())
+ self._path, list(patterns.keys())
)
)