e2e_tests: Extend exec_in_guest functionality

This is a refactoring that adds the following functionality to
exec_in_guest:

- Receive exit codes from executed commands and by default check
  if they are zero.
- Add timeouts to all communication with the guest
- Allow commands to be executed asynchronously, to allow the test
  code to perform actions while the guest command is running.
- Add anyhow error handling with contexts for more helpful error
  messages

BUG=b:269137600
BUG=b:268539201

Change-Id: Ic069c3ccae363790a124db10a0db76e4ad4384c3
Reviewed-on: https://chromium-review.googlesource.com/c/crosvm/crosvm/+/4251717
Reviewed-by: Keiichi Watanabe <keiichiw@chromium.org>
Reviewed-by: Zihan Chen <zihanchen@google.com>

e2e_tests/guest_under_test/PREBUILT_VERSION

@@ -1 +1 @@
-r0003
+r0004

e2e_tests/guest_under_test/delegate.rs

@@ -7,26 +7,32 @@
 use std::fs::File;
 use std::io;
 use std::io::prelude::*;
+use std::os::unix::process::ExitStatusExt;
 use std::path::Path;
 use std::process::Command;
 use std::process::Stdio;
 use std::str;
 
 /// Device file to read from and write to.
-const CONSOLE_FILE: &str = "/dev/ttyS1";
+const CONSOLE_FILE: &'static str = "/dev/ttyS1";
 
-/// Magic line sent when we are ready to receive a command.
+/// Line sent when we are ready to receive a command.
 /// \x05 is the ENQ (enquiry) character, which is rarely used and 'should'
 /// not appear in command output.
-const MAGIC_LINE: &str = "\x05Ready";
+const READY_LINE: &'static str = "\x05READY";
 
-/// When ready to receive a command, the `MAGIC_LINE` is written to `input`.
+/// Line sent containing the exit code of the program
-/// The received command is executed via /bin/sh/ and it's stdout is written
+/// \x05 is the ENQ (enquiry) character, which is rarely used and 'should'
-/// back to `output`, terminated by `MAGIC_LINE`.
+/// not appear in command output.
+const EXIT_CODE_LINE: &'static str = "\x05EXIT_CODE";
+
+/// When ready to receive a command, the `READY_LINE` is written to `input`.
+/// The received command is executed via /bin/sh and it's stdout is written
+/// back to `output`, terminated by `EXIT_CODE_LINE ${exit_code}`.
 fn listen(input: Box<dyn io::Read>, mut output: Box<dyn io::Write>) -> io::Result<()> {
     let mut reader = io::BufReader::new(input);
     loop {
-        writeln!(&mut output, "{}", MAGIC_LINE).unwrap();
+        writeln!(&mut output, "{}", READY_LINE).unwrap();
 
         let mut command = String::new();
         reader.read_line(&mut command)?;
@@ -40,8 +46,14 @@ fn listen(input: Box<dyn io::Read>, mut output: Box<dyn io::Write>) -> io::Resul
             .stderr(Stdio::inherit())
             .output()
             .unwrap();
+        let exit_code = match result.status.code() {
+            Some(code) => code,
+            None => -result.status.signal().unwrap(),
+        };
 
         output.write(&result.stdout)?;
+        println!("<- {}", exit_code);
+        writeln!(&mut output, "{EXIT_CODE_LINE} {exit_code}")?;
     }
     Ok(())
 }

e2e_tests/tests/boot.rs

@@ -7,9 +7,10 @@ use fixture::vm::Config;
 use fixture::vm::TestVm;
 
 #[test]
-fn boot_test_vm() {
+fn boot_test_vm() -> anyhow::Result<()> {
     let mut vm = TestVm::new(Config::new()).unwrap();
-    assert_eq!(vm.exec_in_guest("echo 42").unwrap().trim(), "42");
+    assert_eq!(vm.exec_in_guest("echo 42")?.trim(), "42");
+    Ok(())
 }
 
 #[cfg(unix)]

e2e_tests/tests/fixture/sys/unix.rs

@@ -6,7 +6,6 @@ use std::ffi::CString;
 use std::fs::File;
 use std::fs::OpenOptions;
 use std::io;
-use std::io::BufRead;
 use std::io::BufReader;
 use std::io::Write;
 use std::os::unix::fs::OpenOptionsExt;
@@ -16,19 +15,21 @@ use std::process::Child;
 use std::process::Command;
 use std::process::Stdio;
 use std::str::from_utf8;
+use std::sync::Arc;
+use std::sync::Mutex;
 use std::time::Duration;
 
 use anyhow::anyhow;
+use anyhow::Context;
 use anyhow::Result;
 use libc::O_DIRECT;
 use tempfile::TempDir;
 
 use crate::fixture::utils::find_crosvm_binary;
+use crate::fixture::utils::run_with_timeout;
 use crate::fixture::vm::kernel_path;
 use crate::fixture::vm::rootfs_path;
-use crate::fixture::vm::run_with_timeout;
 use crate::fixture::vm::Config;
-use crate::fixture::vm::TestVm;
 
 const FROM_GUEST_PIPE: &str = "from_guest";
 const TO_GUEST_PIPE: &str = "to_guest";
@@ -66,8 +67,8 @@ pub struct TestVmSys {
     /// Maintain ownership of test_dir until the vm is destroyed.
     #[allow(dead_code)]
     pub test_dir: TempDir,
-    pub from_guest_reader: BufReader<File>,
+    pub from_guest_reader: Arc<Mutex<BufReader<File>>>,
-    pub to_guest: File,
+    pub to_guest: Arc<Mutex<File>>,
     pub control_socket_path: PathBuf,
     pub process: Option<Child>, // Use `Option` to allow taking the ownership in `Drop::drop()`.
 }
@@ -148,11 +149,15 @@ impl TestVmSys {
 
         let mut process = Some(command.spawn()?);
 
-        // Open pipes. Panic if we cannot connect after a timeout.
+        // Open pipes. Apply timeout to `from_guest` since it will block until crosvm opens the
-        let (to_guest, from_guest) = run_with_timeout(
+        // other end.
-            move || (File::create(to_guest_pipe), File::open(from_guest_pipe)),
+        let to_guest = File::create(to_guest_pipe)?;
+        let from_guest = match run_with_timeout(
+            move || File::open(from_guest_pipe),
             VM_COMMUNICATION_TIMEOUT,
-            || {
+        ) {
+            Ok(from_guest) => from_guest.with_context(|| "Cannot open from_guest pipe")?,
+            Err(error) => {
                 let mut process = process.take().unwrap();
                 process.kill().unwrap();
                 let output = process.wait_with_output().unwrap();
@@ -167,19 +172,14 @@ impl TestVmSys {
                     "TestVm stderr:\n{}",
                     std::str::from_utf8(&output.stderr).unwrap()
                 );
-            },
+                panic!("Cannot connect to VM: {}", error);
-        );
+            }
-
+        };
-        // Wait for magic line to be received, indicating the delegate is ready.
-        let mut from_guest_reader = BufReader::new(from_guest?);
-        let mut magic_line = String::new();
-        from_guest_reader.read_line(&mut magic_line)?;
-        assert_eq!(magic_line.trim(), TestVm::MAGIC_LINE);
 
         Ok(TestVmSys {
             test_dir,
-            from_guest_reader,
+            from_guest_reader: Arc::new(Mutex::new(BufReader::new(from_guest))),
-            to_guest: to_guest?,
+            to_guest: Arc::new(Mutex::new(to_guest)),
             control_socket_path,
             process,
         })

e2e_tests/tests/fixture/sys/windows.rs

@@ -8,13 +8,14 @@
 use std::env;
 use std::fs::File;
 use std::fs::OpenOptions;
-use std::io::BufRead;
 use std::io::BufReader;
 use std::io::Write;
 use std::path::Path;
 use std::path::PathBuf;
 use std::process::Child;
 use std::process::Command;
+use std::sync::Arc;
+use std::sync::Mutex;
 use std::time::Duration;
 
 use anyhow::Result;
@@ -25,9 +26,7 @@ use rand::Rng;
 use crate::fixture::utils::find_crosvm_binary;
 use crate::fixture::vm::kernel_path;
 use crate::fixture::vm::rootfs_path;
-use crate::fixture::vm::run_with_timeout;
 use crate::fixture::vm::Config;
-use crate::fixture::vm::TestVm;
 
 const GUEST_EARLYCON: &str = "guest_earlycon.log";
 const GUEST_CONSOLE: &str = "guest_latecon.log";
@@ -40,10 +39,6 @@ const SLEEP_TIMEOUT: Duration = Duration::from_millis(500);
 // presubmit runs.
 const RETRY_COUNT: u16 = 600;
 
-/// Timeout for communicating with the VM. If we do not hear back, panic so we
-/// do not block the tests.
-const VM_COMMUNICATION_TIMEOUT: Duration = Duration::from_secs(60);
-
 pub struct SerialArgs {
     // This pipe is used to communicate to/from guest.
     from_guest_pipe: PathBuf,
@@ -129,8 +124,8 @@ fn create_client_pipe_helper(from_guest_pipe: &str, logs_dir: &str) -> PipeConne
 
 #[cfg(test)]
 pub struct TestVmSys {
-    pub(crate) from_guest_reader: BufReader<PipeConnection>,
+    pub(crate) from_guest_reader: Arc<Mutex<BufReader<PipeConnection>>>,
-    pub(crate) to_guest: PipeConnection,
+    pub(crate) to_guest: Arc<Mutex<PipeConnection>>,
     pub(crate) process: Option<Child>, // Use `Option` to allow taking the ownership in `Drop::drop()`.
 }
 
@@ -221,29 +216,11 @@ impl TestVmSys {
         let process = Some(command.spawn().unwrap());
 
         let to_guest = create_client_pipe_helper(&from_guest_path, logs_dir);
-
+        let from_guest_reader = BufReader::new(to_guest.try_clone().unwrap());
-        // Wait for magic line to be received, indicating the delegate is ready.
-        let mut from_guest_reader = BufReader::new(to_guest.try_clone().unwrap());
-        let from_guest_reader = run_with_timeout(
-            move || {
-                let mut magic_line = String::new();
-                if from_guest_reader.read_line(&mut magic_line).is_err() {
-                    dump_logs(logs_dir);
-                    panic!("failed to read from pipe");
-                }
-                assert_eq!(magic_line.trim(), TestVm::MAGIC_LINE);
-                from_guest_reader
-            },
-            VM_COMMUNICATION_TIMEOUT,
-            || {
-                dump_logs(logs_dir);
-                panic!("Timeout while waiting to read magic line");
-            },
-        );
 
         Ok(TestVmSys {
-            from_guest_reader,
+            from_guest_reader: Arc::new(Mutex::new(from_guest_reader)),
-            to_guest,
+            to_guest: Arc::new(Mutex::new(to_guest)),
             process,
         })
     }
@@ -350,13 +327,7 @@ impl TestVmSys {
         Ok(())
     }
 
-    pub fn crosvm_command(&mut self, command: &str, mut _args: Vec<String>) -> Result<()> {
+    pub fn crosvm_command(&mut self, _command: &str, mut _args: Vec<String>) -> Result<()> {
-        writeln!(&mut self.to_guest, "{}", command)?;
+        unimplemented!()
-
-        // We will receive an echo of what we have written on the pipe.
-        let mut echo = String::new();
-        self.from_guest_reader.read_line(&mut echo)?;
-        assert_eq!(echo.trim(), command);
-        Ok(())
     }
 }

e2e_tests/tests/fixture/utils.rs

@@ -6,6 +6,13 @@
 
 use std::env;
 use std::path::PathBuf;
+use std::sync::mpsc::sync_channel;
+use std::sync::mpsc::RecvTimeoutError;
+use std::thread;
+use std::time::Duration;
+
+use anyhow::anyhow;
+use anyhow::Result;
 
 use crate::fixture::sys::binary_name;
 
@@ -30,3 +37,27 @@ pub fn find_crosvm_binary() -> PathBuf {
         binary_name
     );
 }
+
+/// Run the provided closure in a separate thread and return it's result. If the closure does not
+/// finish before the timeout is reached, an Error is returned instead.
+///
+/// WARNING: It is not possible to kill the closure if a timeout occurs. It is advised to panic
+/// when an error is returned.
+pub(super) fn run_with_timeout<F, U>(closure: F, timeout: Duration) -> Result<U>
+where
+    F: FnOnce() -> U + Send + 'static,
+    U: Send + 'static,
+{
+    let (tx, rx) = sync_channel::<()>(1);
+    let handle = thread::spawn(move || {
+        let result = closure();
+        // Notify main thread the closure is done. Fail silently if it's not listening anymore.
+        let _ = tx.send(());
+        result
+    });
+    match rx.recv_timeout(timeout) {
+        Ok(_) => Ok(handle.join().unwrap()),
+        Err(RecvTimeoutError::Timeout) => Err(anyhow!("closure timed out after {timeout:?}")),
+        Err(RecvTimeoutError::Disconnected) => Err(anyhow!("closure paniced")),
+    }
+}

e2e_tests/tests/fixture/vm.rs

@@ -8,17 +8,19 @@ use std::io::Write;
 use std::path::PathBuf;
 use std::process::Command;
 use std::str::from_utf8;
-use std::sync::mpsc::sync_channel;
 use std::sync::Once;
-use std::thread;
 use std::time::Duration;
 
+use anyhow::anyhow;
+use anyhow::bail;
+use anyhow::Context;
 use anyhow::Result;
 use base::syslog;
 use prebuilts::download_file;
 
 use crate::fixture::sys::SerialArgs;
 use crate::fixture::sys::TestVmSys;
+use crate::fixture::utils::run_with_timeout;
 
 const PREBUILT_URL: &str = "https://storage.googleapis.com/crosvm/integration_tests";
 
@@ -29,6 +31,15 @@ const ARCH: &str = "arm";
 #[cfg(target_arch = "aarch64")]
 const ARCH: &str = "aarch64";
 
+/// Timeout when waiting for pipes that are expected to be ready.
+const COMMUNICATION_TIMEOUT: Duration = Duration::from_secs(5);
+
+/// Timeout for the VM to boot and the delegate to report that it's ready.
+const BOOT_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Default timeout when waiting for guest commands to execute
+const DEFAULT_COMMAND_TIMEOUT: Duration = Duration::from_secs(10);
+
 fn prebuilt_version() -> &'static str {
     include_str!("../../guest_under_test/PREBUILT_VERSION").trim()
 }
@@ -73,26 +84,73 @@ pub(super) fn rootfs_path() -> PathBuf {
     }
 }
 
-/// Run the provided closure, but panic if it does not complete until the timeout has passed.
+/// Represents a command running in the guest. See `TestVm::exec_in_guest_async()`
-/// We should panic here, as we cannot gracefully stop the closure from running.
+#[must_use]
-/// `on_timeout` will be called before panic to allow printing debug information.
+pub struct GuestProcess {
-pub(super) fn run_with_timeout<F, G, U>(closure: F, timeout: Duration, on_timeout: G) -> U
+    command: String,
-where
+    timeout: Duration,
-    F: FnOnce() -> U + Send + 'static,
+}
-    G: FnOnce(),
+
-    U: Send + 'static,
+impl GuestProcess {
-{
+    pub fn with_timeout(self, duration: Duration) -> Self {
-    let (tx, rx) = sync_channel::<()>(1);
+        Self {
-    let handle = thread::spawn(move || {
+            timeout: duration,
-        let result = closure();
+            ..self
-        tx.send(()).unwrap();
+        }
-        result
+    }
-    });
+
-    if rx.recv_timeout(timeout).is_err() {
+    /// Waits for the process to finish execution and return the produced stdout.
-        on_timeout();
+    /// Will fail on a non-zero exit code.
-        panic!("Operation timed out or closure paniced.");
+    pub fn wait(self, vm: &mut TestVm) -> Result<String> {
+        let command = self.command.clone();
+        let (exit_code, output) = self.wait_unchecked(vm)?;
+        if exit_code != 0 {
+            bail!(
+                "Command `{}` terminated with exit code {}",
+                command,
+                exit_code
+            );
+        }
+        Ok(output)
+    }
+
+    /// Same as `wait` but will return a tuple of (exit code, output) instead of failing
+    /// on a non-zero exit code.
+    pub fn wait_unchecked(self, vm: &mut TestVm) -> Result<(i32, String)> {
+        // First read echo of command
+        let echo = vm
+            .read_line_from_guest(COMMUNICATION_TIMEOUT)
+            .with_context(|| {
+                format!(
+                    "Command `{}`: Failed to read echo from guest pipe",
+                    self.command
+                )
+            })?;
+        assert_eq!(echo.trim(), self.command.trim());
+
+        // Then read stdout and exit code
+        let mut output = Vec::<String>::new();
+        let exit_code = loop {
+            let line = vm.read_line_from_guest(self.timeout).with_context(|| {
+                format!(
+                    "Command `{}`: Failed to read response from guest",
+                    self.command
+                )
+            })?;
+            let trimmed = line.trim();
+            if trimmed.starts_with(TestVm::EXIT_CODE_LINE) {
+                let exit_code_str = &trimmed[(TestVm::EXIT_CODE_LINE.len() + 1)..];
+                break exit_code_str.parse::<i32>().unwrap();
+            }
+            output.push(trimmed.to_owned());
+        };
+
+        // Finally get the VM in a ready state again.
+        vm.wait_for_guest(COMMUNICATION_TIMEOUT)
+            .with_context(|| format!("Command `{}`: Failed to wait for guest", self.command))?;
+
+        Ok((exit_code, output.join("\n")))
     }
-    handle.join().unwrap()
 }
 
 /// Configuration to start `TestVm`.
@@ -141,12 +199,17 @@ static PREP_ONCE: Once = Once::new();
 /// when this instance is dropped.
 #[cfg(test)]
 pub struct TestVm {
+    // Platform-dependent bits
     sys: TestVmSys,
+    // The guest is ready to receive a command.
+    ready: bool,
 }
 
 impl TestVm {
-    /// Magic line sent by the delegate binary when the guest is ready.
+    /// Line sent by the delegate binary when the guest is ready.
-    pub(super) const MAGIC_LINE: &'static str = "\x05Ready";
+    const READY_LINE: &'static str = "\x05READY";
+    /// Line sent by the delegate binary to terminate the stdout and send the exit code.
+    const EXIT_CODE_LINE: &'static str = "\x05EXIT_CODE";
 
     /// Downloads prebuilts if needed.
     fn initialize_once() {
@@ -196,10 +259,13 @@ impl TestVm {
         F: FnOnce(&mut Command, &SerialArgs, &Config) -> Result<()>,
     {
         PREP_ONCE.call_once(TestVm::initialize_once);
-
+        let mut vm = TestVm {
-        Ok(TestVm {
+            sys: TestVmSys::new_generic(f, cfg).with_context(|| "Could not start crosvm")?,
-            sys: TestVmSys::new_generic(f, cfg)?,
+            ready: false,
-        })
+        };
+        vm.wait_for_guest(BOOT_TIMEOUT)
+            .with_context(|| "Guest did not become ready after boot")?;
+        Ok(vm)
     }
 
     pub fn new(cfg: Config) -> Result<TestVm> {
@@ -212,50 +278,77 @@ impl TestVm {
         TestVm::new_generic(TestVmSys::append_config_file_arg, cfg)
     }
 
-    /// Executes the shell command `command` and returns the programs stdout.
+    /// Executes the provided command in the guest.
+    /// Returns the stdout that was produced by the command, or a GuestProcessError::ExitCode if
+    /// the program did not exit with 0.
     pub fn exec_in_guest(&mut self, command: &str) -> Result<String> {
-        self.exec_command(command)?;
+        self.exec_in_guest_async(command)?.wait(self)
-        self.wait_for_guest()
     }
 
-    pub fn exec_command(&mut self, command: &str) -> Result<()> {
+    /// Same as `exec_in_guest` but will return a tuple of (exit code, output) instead of failing
-        // Write command to serial port.
+    /// on a non-zero exit code.
-        writeln!(&mut self.sys.to_guest, "{}", command)?;
+    pub fn exec_in_guest_unchecked(&mut self, command: &str) -> Result<(i32, String)> {
-
+        self.exec_in_guest_async(command)?.wait_unchecked(self)
-        // We will receive an echo of what we have written on the pipe.
-        let mut echo = String::new();
-        self.sys.from_guest_reader.read_line(&mut echo)?;
-        assert_eq!(echo.trim(), command);
-        Ok(())
     }
 
-    /// Executes the shell command `command` async, allowing for calls other actions between the
+    /// Executes the provided command in the guest asynchronously.
-    /// command call and the result, and returns the programs stdout.
+    /// The command will be run in the guest, but output will not be read until GuestProcess::wait
-    pub fn exec_command_async(&mut self, command: &str, block: impl Fn(&mut Self)) -> Result<()> {
+    /// is called.
-        // Write command to serial port.
+    pub fn exec_in_guest_async(&mut self, command: &str) -> Result<GuestProcess> {
-        writeln!(&mut self.sys.to_guest, "{}", command)?;
+        assert!(self.ready);
-        block(self);
+        self.ready = false;
-        let mut echo = String::new();
+
-        self.sys.from_guest_reader.read_line(&mut echo)?;
+        // Send command and read echo from the pipe
-        assert_eq!(echo.trim(), command);
+        self.write_line_to_guest(command, COMMUNICATION_TIMEOUT)
-        Ok(())
+            .with_context(|| format!("Command `{}`: Failed to write to guest pipe", command))?;
+
+        Ok(GuestProcess {
+            command: command.to_owned(),
+            timeout: DEFAULT_COMMAND_TIMEOUT,
+        })
     }
 
-    pub fn wait_for_guest(&mut self) -> Result<String> {
+    // Waits for the guest to be ready to receive commands
-        // Return all remaining lines until we receive the MAGIC_LINE
+    fn wait_for_guest(&mut self, timeout: Duration) -> Result<()> {
-        let mut output = String::new();
+        assert!(!self.ready);
-        loop {
+        let line = self.read_line_from_guest(timeout)?;
-            let mut line = String::new();
+        if line.trim() == TestVm::READY_LINE {
-            self.sys.from_guest_reader.read_line(&mut line)?;
+            self.ready = true;
-            if line.trim() == TestVm::MAGIC_LINE {
+            Ok(())
-                break;
+        } else {
-            }
+            Err(anyhow!(
-            output.push_str(&line);
+                "Expected READY line from delegate, but got: {:?}",
+                line.trim()
+            ))
         }
-        let trimmed = output.trim();
+    }
-        println!("<- {:?}", trimmed);
 
-        Ok(trimmed.to_string())
+    /// Reads one line via the `from_guest` pipe from the guest delegate.
+    fn read_line_from_guest(&mut self, timeout: Duration) -> Result<String> {
+        let reader = self.sys.from_guest_reader.clone();
+        run_with_timeout(
+            move || {
+                let mut data = String::new();
+                reader.lock().unwrap().read_line(&mut data)?;
+                println!("<- {:?}", data);
+                Ok(data)
+            },
+            timeout,
+        )?
+    }
+
+    /// Send one line via the `to_guest` pipe to the guest delegate.
+    fn write_line_to_guest(&mut self, data: &str, timeout: Duration) -> Result<()> {
+        let writer = self.sys.to_guest.clone();
+        let data = data.to_owned();
+        run_with_timeout(
+            move || -> Result<()> {
+                println!("-> {:?}", data);
+                writeln!(writer.lock().unwrap(), "{}", data)?;
+                Ok(())
+            },
+            timeout,
+        )?
     }
 
     pub fn stop(&mut self) -> Result<()> {

e2e_tests/tests/suspend_resume.rs

@@ -3,11 +3,12 @@
 // found in the LICENSE file.
 
 pub mod fixture;
-use crate::fixture::vm::Config;
-use crate::fixture::vm::TestVm;
 
 use tempfile::tempdir;
 
+use crate::fixture::vm::Config;
+use crate::fixture::vm::TestVm;
+
 // Tests for suspend/resume.
 //
 // System-wide suspend/resume, snapshot/restore.
@@ -57,15 +58,13 @@ fn suspend_resume_system(vm: &mut TestVm) -> anyhow::Result<()> {
     // This command will get queued and not run while the VM is suspended. The command is saved in
     // the serial device. After the snapshot is taken, the VM is resumed. At that point, the
     // command runs and is validated.
-    vm.exec_command_async("echo 42", |vm| {
+    let echo_cmd = vm.exec_in_guest_async("echo 42").unwrap();
-        // Take snapshot of modified VM
+    // Take snapshot of modified VM
-        println!("snapshotting VM - mod state");
+    println!("snapshotting VM - mod state");
-        vm.snapshot(&snap2_path).unwrap();
+    vm.snapshot(&snap2_path).unwrap();
 
-        vm.resume().unwrap();
+    vm.resume().unwrap();
-    })
+    assert_eq!("42", echo_cmd.wait(vm).unwrap());
-    .unwrap();
-    assert_eq!("42", vm.wait_for_guest().unwrap());
 
     // suspend VM
     vm.suspend().unwrap();

e2e_tests/tests/vsock.rs

@@ -13,10 +13,9 @@ use std::process::Stdio;
 use std::thread;
 use std::time::Duration;
 
-use rand::Rng;
-
 use fixture::vm::Config;
 use fixture::vm::TestVm;
+use rand::Rng;
 
 const HOST_CID: u64 = 2;
 const VSOCK_COM_PORT: u64 = 11111;
@@ -50,12 +49,11 @@ fn host_to_guest_connection(config: Config) {
     let config = config.extra_args(vec!["--cid".to_string(), guest_cid.to_string()]);
     let mut vm = TestVm::new(config).unwrap();
 
-    let handle_guest = thread::spawn(move || {
+    let guest_cmd = vm
-        let cmd = format!(
+        .exec_in_guest_async(&format!(
-            "echo {MESSAGE_TO_HOST} | timeout {SERVER_TIMEOUT_IN_SEC}s ncat -l --vsock {VSOCK_COM_PORT}",
+            "echo {MESSAGE_TO_HOST} | ncat -l --vsock {VSOCK_COM_PORT}"
-        );
+        ))
-        vm.exec_in_guest(&cmd).unwrap();
+        .unwrap();
-    });
 
     // wait until the server is ready
     thread::sleep(CLIENT_WAIT_DURATION);
@@ -69,11 +67,11 @@ fn host_to_guest_connection(config: Config) {
         ])
         .output()
         .expect("failed to execute process");
+
     let host_stdout = std::str::from_utf8(&output.stdout).unwrap();
-
-    handle_guest.join().unwrap();
-
     assert_eq!(host_stdout.trim(), MESSAGE_TO_HOST);
+
+    guest_cmd.wait(&mut vm).unwrap();
 }
 
 #[test]
@@ -111,7 +109,14 @@ fn guest_to_host_connection(config: Config) {
     thread::sleep(CLIENT_WAIT_DURATION);
 
     let cmd = format!("ncat --idle-timeout 1 --vsock {HOST_CID} {VSOCK_COM_PORT}");
-    let guest_stdout = vm.exec_in_guest(&cmd).unwrap();
+    let (exit_code, guest_stdout) = vm
+        .exec_in_guest_async(&cmd)
+        .unwrap()
+        .with_timeout(Duration::from_secs(2))
+        .wait_unchecked(&mut vm)
+        .unwrap();
+    // We expect to hit the idle-timeout, which will exit with code 1.
+    assert_eq!(exit_code, 1);
 
     handle_host.wait().unwrap();