tests: drain the replayed serial backlog and stop the pty echo loop

With socket serial output now buffered and replayed on connect, a
late-connecting client receives the whole boot backlog. The pty
interaction test had three problems with that:

- pty_read() slept a second between 512-byte reads and the loop consumed
  one chunk per two-second tick, far too slow to drain the backlog. Read
  in larger chunks without the per-read sleep and drain everything
  available each round; bound the loop so a missing marker can't run to
  the harness timeout.

- it wrote the login keystrokes before reading, so the unread backlog
  back-pressured the sender and the keystrokes never reached the prompt.
  Start reading concurrently with typing instead.

- the socat pty was created with echo on, so the replayed backlog was
  echoed back to the guest as serial input, flooding it (UART input
  overrun, login never completing). Create the pty with echo=0.

Signed-off-by: Max Makarov <maxpain@linux.com>
Assisted-by: Claude:claude-opus-4-8 [Claude Code]

cloud-hypervisor/tests/common/tests_wrappers.rs

@@ -256,6 +256,11 @@ pub(crate) fn _test_pty_interaction(pty_path: PathBuf) {
         .open(pty_path)
         .unwrap();
 
+    // Read concurrently so we drain the replayed backlog instead of leaving
+    // it stranded behind a non-reading client (which would back-pressure the
+    // sender and never let the login keystrokes through).
+    let ptyc = pty_read(cf.try_clone().unwrap());
+
     // Some dumb sleeps but we don't want to write
     // before the console is up and we don't want
     // to try and write the next line before the
@@ -268,31 +273,28 @@ pub(crate) fn _test_pty_interaction(pty_path: PathBuf) {
     assert_eq!(cf.write(b"echo test_pty_console\n").unwrap(), 22);
     thread::sleep(std::time::Duration::new(2, 0));
 
-    // read pty and ensure they have a login shell
-    // some fairly hacky workarounds to avoid looping
-    // forever in case the channel is blocked getting output
-    let ptyc = pty_read(cf);
-    let mut empty = 0;
     let mut prev = String::new();
-    loop {
+    // The console can stream continuously (e.g. journald forwarded to it), so
+    // bound the wait: a missing marker must not loop until the harness timeout.
+    for _ in 0..20 {
         thread::sleep(std::time::Duration::new(2, 0));
-        match ptyc.try_recv() {
-            Ok(line) => {
-                empty = 0;
-                prev = prev + &line;
-                if prev.contains("test_pty_console") {
-                    break;
+        // Drain everything available this round so a large replayed backlog
+        // does not take one 2s tick per chunk to get through.
+        loop {
+            match ptyc.try_recv() {
+                Ok(line) => {
+                    prev = prev + &line;
+                    if prev.contains("test_pty_console") {
+                        return;
+                    }
                 }
-            }
-            Err(mpsc::TryRecvError::Empty) => {
-                empty += 1;
-                assert!(empty <= 5, "No login on pty");
-            }
-            _ => {
-                panic!("No login on pty")
+                Err(mpsc::TryRecvError::Empty) => break,
+                Err(_) => panic!("No login on pty"),
             }
         }
     }
+    // Bounded out without ever seeing the marker: the login never completed.
+    panic!("No login on pty");
 }
 
 pub(crate) fn test_cpu_topology(

cloud-hypervisor/tests/common/utils.rs

@@ -780,14 +780,13 @@ pub(super) fn pty_read(mut pty: std::fs::File) -> Receiver<String> {
     let (tx, rx) = mpsc::channel::<String>();
     thread::spawn(move || {
         loop {
-            thread::sleep(std::time::Duration::new(1, 0));
-            let mut buf = [0; 512];
+            let mut buf = [0; 4096];
             match pty.read(&mut buf) {
-                Ok(_bytes) => {
-                    let output = std::str::from_utf8(&buf).unwrap().to_string();
-                    match tx.send(output) {
-                        Ok(_) => (),
-                        Err(_) => break,
+                Ok(0) => break,
+                Ok(bytes) => {
+                    let output = String::from_utf8_lossy(&buf[..bytes]).into_owned();
+                    if tx.send(output).is_err() {
+                        break;
                     }
                 }
                 Err(_) => break,

cloud-hypervisor/tests/integration.rs

@@ -2536,7 +2536,7 @@ mod common_parallel {
 
         let mut socat_command = Command::new("socat");
         let socat_args = [
-            &format!("pty,link={},raw", serial_socket_pty.display()),
+            &format!("pty,link={},raw,echo=0", serial_socket_pty.display()),
             &format!("UNIX-CONNECT:{}", serial_socket.display()),
         ];
         socat_command.args(socat_args);