Loading nixos/lib/test-driver/test_driver/machine.py +29 −14 Original line number Diff line number Diff line Loading @@ -856,21 +856,36 @@ class Machine: with self.nested(f"waiting for {regex} to appear on screen"): retry(screen_matches) def wait_for_console_text(self, regex: str) -> None: with self.nested(f"waiting for {regex} to appear on console"): def wait_for_console_text(self, regex: str, timeout: int | 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. """ # Buffer the console output, this is needed # to match multiline regexes. console = io.StringIO() while True: def console_matches() -> bool: nonlocal console 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: self.sleep(1) continue pass console.seek(0) matches = re.search(regex, console.read()) if matches is not None: return return (matches is not None) with self.nested(f"waiting for {regex} to appear on console"): if timeout is not None: retry(console_matches, timeout) else: while not console_matches(): pass def send_key( self, key: str, delay: Optional[float] = 0.01, log: Optional[bool] = True Loading Loading
nixos/lib/test-driver/test_driver/machine.py +29 −14 Original line number Diff line number Diff line Loading @@ -856,21 +856,36 @@ class Machine: with self.nested(f"waiting for {regex} to appear on screen"): retry(screen_matches) def wait_for_console_text(self, regex: str) -> None: with self.nested(f"waiting for {regex} to appear on console"): def wait_for_console_text(self, regex: str, timeout: int | 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. """ # Buffer the console output, this is needed # to match multiline regexes. console = io.StringIO() while True: def console_matches() -> bool: nonlocal console 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: self.sleep(1) continue pass console.seek(0) matches = re.search(regex, console.read()) if matches is not None: return return (matches is not None) with self.nested(f"waiting for {regex} to appear on console"): if timeout is not None: retry(console_matches, timeout) else: while not console_matches(): pass def send_key( self, key: str, delay: Optional[float] = 0.01, log: Optional[bool] = True Loading