diff options
author | Michele Guerini Rocco <rnhmjoj@users.noreply.github.com> | 2020-07-01 09:04:30 +0200 |
---|---|---|
committer | GitHub <noreply@github.com> | 2020-07-01 09:04:30 +0200 |
commit | dab676b2d7bccee76d005ff1f8a9b01c21718c70 (patch) | |
tree | a1eb928902426078a1a0e2f28ebf2251702bcd01 /nixos/lib | |
parent | 8ba675379ab1af0e1b2165fb4cbaffeea2468925 (diff) | |
parent | 33c4a4bdd52a750205046cc85775e6caea342868 (diff) |
Merge pull request #65231 from buckley310/grub-password
grub: add support for passwords
Diffstat (limited to 'nixos/lib')
-rw-r--r-- | nixos/lib/test-driver/test-driver.py | 23 |
1 files changed, 23 insertions, 0 deletions
diff --git a/nixos/lib/test-driver/test-driver.py b/nixos/lib/test-driver/test-driver.py index f454b052dc31..4cb928538e7c 100644 --- a/nixos/lib/test-driver/test-driver.py +++ b/nixos/lib/test-driver/test-driver.py @@ -3,6 +3,8 @@ from contextlib import contextmanager, _GeneratorContextManager from queue import Queue, Empty from typing import Tuple, Any, Callable, Dict, Iterator, Optional, List from xml.sax.saxutils import XMLGenerator +import queue +import io import _thread import argparse import atexit @@ -672,6 +674,22 @@ class Machine: with self.nested("waiting for {} to appear on screen".format(regex)): retry(screen_matches) + def wait_for_console_text(self, regex: str) -> None: + self.log("waiting for {} to appear on console".format(regex)) + # Buffer the console output, this is needed + # to match multiline regexes. + console = io.StringIO() + while True: + try: + console.write(self.last_lines.get()) + except queue.Empty: + self.sleep(1) + continue + console.seek(0) + matches = re.search(regex, console.read()) + if matches is not None: + return + def send_key(self, key: str) -> None: key = CHAR_TO_KEY.get(key, key) self.send_monitor_command("sendkey {}".format(key)) @@ -735,11 +753,16 @@ class Machine: self.monitor, _ = self.monitor_socket.accept() self.shell, _ = self.shell_socket.accept() + # Store last serial console lines for use + # of wait_for_console_text + self.last_lines: Queue = Queue() + def process_serial_output() -> None: assert self.process.stdout is not None for _line in self.process.stdout: # Ignore undecodable bytes that may occur in boot menus line = _line.decode(errors="ignore").replace("\r", "").rstrip() + self.last_lines.put(line) eprint("{} # {}".format(self.name, line)) self.logger.enqueue({"msg": line, "machine": self.name}) |