diff --git a/nixos/lib/test-driver/src/test_driver/machine/__init__.py b/nixos/lib/test-driver/src/test_driver/machine/__init__.py index e78d9af4567a..9a80e2b88c6a 100644 --- a/nixos/lib/test-driver/src/test_driver/machine/__init__.py +++ b/nixos/lib/test-driver/src/test_driver/machine/__init__.py @@ -551,40 +551,6 @@ class BaseMachine(ABC): """ return self.systemctl(f"stop {jobname}", user) - def wait_for_job(self, jobname: str) -> None: - self.wait_for_unit(jobname) - - def get_tty_text(self, tty: str) -> str: - """ - Get the output printed to a given TTY. - """ - status, output = self.execute( - f"fold -w$(stty -F /dev/tty{tty} size | awk '{{print $2}}') /dev/vcs{tty}" - ) - return output - - def wait_until_tty_matches(self, tty: str, regexp: str, timeout: int = 900) -> None: - """Wait until the visible output on the chosen TTY matches regular - expression. Throws an exception on timeout. - """ - matcher = re.compile(regexp) - - def tty_matches(last_try: bool) -> bool: - text = self.get_tty_text(tty) - if last_try: - self.log( - f"Last chance to match /{regexp}/ on TTY{tty}, " - f"which currently contains: {text}" - ) - return len(matcher.findall(text)) > 0 - - with self.nested(f"waiting for {regexp} to appear on tty {tty}"): - retry(tty_matches, timeout) - - def dump_tty_contents(self, tty: str) -> None: - """Debugging: Dump the contents of the TTY""" - self.execute(f"fold -w 80 /dev/vcs{tty} | systemd-cat") - def execute( self, command: str, @@ -804,6 +770,37 @@ class QemuMachine(BaseMachine): break return "".join(output_buffer) + def get_tty_text(self, tty: str) -> str: + """ + Get the output printed to a given TTY. + """ + status, output = self.execute( + f"fold -w$(stty -F /dev/tty{tty} size | awk '{{print $2}}') /dev/vcs{tty}" + ) + return output + + def wait_until_tty_matches(self, tty: str, regexp: str, timeout: int = 900) -> None: + """Wait until the visible output on the chosen TTY matches regular + expression. Throws an exception on timeout. + """ + matcher = re.compile(regexp) + + def tty_matches(last_try: bool) -> bool: + text = self.get_tty_text(tty) + if last_try: + self.log( + f"Last chance to match /{regexp}/ on TTY{tty}, " + f"which currently contains: {text}" + ) + return len(matcher.findall(text)) > 0 + + with self.nested(f"waiting for {regexp} to appear on tty {tty}"): + retry(tty_matches, timeout) + + def dump_tty_contents(self, tty: str) -> None: + """Debugging: Dump the contents of the TTY""" + self.execute(f"fold -w 80 /dev/vcs{tty} | systemd-cat") + def _execute( self, command: str,