nixos/test-driver: add timeout option for wait_for_console_text
Previously, `wait_for_console_text` would block indefinitely until there were lines shown in the buffer. This is highly annoying when testing for things that can just hang for some reasons. This introduces a classical timeout mechanism via non-blocking get on the Queue.
This commit is contained in:
@@ -855,17 +855,30 @@ class Machine:
|
|||||||
with self.nested(f"waiting for {regex} to appear on screen"):
|
with self.nested(f"waiting for {regex} to appear on screen"):
|
||||||
retry(screen_matches)
|
retry(screen_matches)
|
||||||
|
|
||||||
def wait_for_console_text(self, regex: str) -> None:
|
def wait_for_console_text(self, regex: str, timeout: float | None = None) -> None:
|
||||||
|
"""
|
||||||
|
Wait for the provided regex to appear on console.
|
||||||
|
For each reads,
|
||||||
|
|
||||||
|
If timeout is None, timeout is infinite.
|
||||||
|
|
||||||
|
`timeout` is in seconds.
|
||||||
|
"""
|
||||||
with self.nested(f"waiting for {regex} to appear on console"):
|
with self.nested(f"waiting for {regex} to appear on console"):
|
||||||
# Buffer the console output, this is needed
|
# Buffer the console output, this is needed
|
||||||
# to match multiline regexes.
|
# to match multiline regexes.
|
||||||
console = io.StringIO()
|
console = io.StringIO()
|
||||||
|
start = time.time()
|
||||||
while True:
|
while True:
|
||||||
try:
|
try:
|
||||||
console.write(self.last_lines.get())
|
# This will return as soon as possible and
|
||||||
|
# sleep 1 second.
|
||||||
|
console.write(self.last_lines.get(block=False))
|
||||||
except queue.Empty:
|
except queue.Empty:
|
||||||
self.sleep(1)
|
time.sleep(1)
|
||||||
continue
|
if timeout is not None and time.time() - start >= timeout:
|
||||||
|
# If we reached here, we didn't honor our timeout constraint.
|
||||||
|
raise Exception(f"`wait_for_console_text` did not match `{regex}` after {timeout} seconds")
|
||||||
console.seek(0)
|
console.seek(0)
|
||||||
matches = re.search(regex, console.read())
|
matches = re.search(regex, console.read())
|
||||||
if matches is not None:
|
if matches is not None:
|
||||||
|
|||||||
Reference in New Issue
Block a user