nixos/test-runner: Allow writing to qemu stdin

nixos/lib/test-driver/test_driver/machine.py

@@ -198,7 +198,7 @@ class StartCommand:
     ) -> subprocess.Popen:
         return subprocess.Popen(
             self.cmd(monitor_socket_path, shell_socket_path),
-            stdin=subprocess.DEVNULL,
+            stdin=subprocess.PIPE,
             stdout=subprocess.PIPE,
             stderr=subprocess.STDOUT,
             shell=True,
@@ -558,6 +558,28 @@ class Machine:
             pass_fds=[self.shell.fileno()],
         )
 
+    def console_interact(self) -> None:
+        """Allows you to interact with QEMU's stdin
+
+        The shell can be exited with Ctrl+D. Note that Ctrl+C is not allowed to be used.
+        QEMU's stdout is read line-wise.
+
+        Should only be used during test development, not in the production test."""
+        self.log("Terminal is ready (there is no prompt):")
+
+        assert self.process
+        assert self.process.stdin
+
+        while True:
+            try:
+                char = sys.stdin.buffer.read(1)
+            except KeyboardInterrupt:
+                break
+            if char == b"":  # ctrl+d
+                self.log("Closing connection to the console")
+                break
+            self.send_console(char.decode())
+
     def succeed(self, *commands: str, timeout: Optional[int] = None) -> str:
         """Execute each command and check that it succeeds."""
         output = ""
@@ -834,6 +856,12 @@ class Machine:
         self.send_monitor_command("sendkey {}".format(key))
         time.sleep(0.01)
 
+    def send_console(self, chars: str) -> None:
+        assert self.process
+        assert self.process.stdin
+        self.process.stdin.write(chars.encode())
+        self.process.stdin.flush()
+
     def start(self) -> None:
         if self.booted:
             return