test_infra: add polling helpers for integration tests

Add generic polling helpers for integration tests and build the
SSH wait helpers on top of them.

This lets follow-up test changes replace fixed sleeps with
condition-based waits without duplicating retry logic at each call
site.

On-behalf-of: SAP philipp.schuster@sap.com

Signed-off-by: Philipp Schuster <philipp.schuster@cyberus-technology.de>

test_infra/src/lib.rs

@@ -15,7 +15,7 @@ use std::os::unix::io::{AsRawFd, FromRawFd};
 use std::path::{Path, PathBuf};
 use std::process::{Child, Command, ExitStatus, Output, Stdio};
 use std::str::FromStr;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 use std::{env, fmt, fs, io, thread};
 
 use rand::Rng;
@@ -57,6 +57,44 @@ pub enum Error {
     WaitTimeout(#[source] WaitTimeoutError),
 }
 
+/// Polls a boolean condition until it becomes true or the timeout expires.
+pub fn wait_until<F>(timeout: Duration, mut condition: F) -> bool
+where
+    F: FnMut() -> bool,
+{
+    const INTERVAL: Duration = Duration::from_millis(50);
+    let start = Instant::now();
+
+    loop {
+        if condition() {
+            return true;
+        }
+
+        if start.elapsed() >= timeout {
+            return false;
+        }
+
+        thread::sleep(INTERVAL);
+    }
+}
+
+/// Retries an operation until it returns `Ok` or the timeout expires.
+pub fn wait_until_succeeds<F, T, E>(timeout: Duration, mut operation: F) -> Result<T, E>
+where
+    F: FnMut() -> Result<T, E>,
+{
+    const INTERVAL: Duration = Duration::from_millis(50);
+    let start = Instant::now();
+
+    loop {
+        match operation() {
+            Ok(result) => return Ok(result),
+            Err(err) if start.elapsed() >= timeout => return Err(err),
+            Err(_) => thread::sleep(INTERVAL),
+        }
+    }
+}
+
 pub struct GuestNetworkConfig {
     pub guest_ip0: String,
     pub host_ip0: String,
@@ -618,6 +656,25 @@ pub enum SshCommandError {
     WaitEof(#[source] ssh2::Error),
 }
 
+#[derive(Error, Debug)]
+pub enum WaitForSshError {
+    #[error("timed out after {timeout:?} waiting for ssh command {command:?} on {ip}: {source}")]
+    Timeout {
+        command: String,
+        ip: String,
+        timeout: Duration,
+        #[source]
+        source: SshCommandError,
+    },
+}
+
+fn default_guest_auth() -> PasswordAuth {
+    PasswordAuth {
+        username: String::from("cloud"),
+        password: String::from("cloud123"),
+    }
+}
+
 fn scp_to_guest_with_auth(
     path: &Path,
     remote_path: &Path,
@@ -791,6 +848,24 @@ pub fn ssh_command_ip(
     )
 }
 
+/// Waits until SSH to the guest becomes available.
+pub fn wait_for_ssh(
+    command: &str,
+    auth: &PasswordAuth,
+    ip: &str,
+    timeout: Duration,
+) -> Result<String, WaitForSshError> {
+    wait_until_succeeds(timeout, || {
+        ssh_command_ip_with_auth(command, auth, ip, 1, 1)
+    })
+    .map_err(|source| WaitForSshError::Timeout {
+        command: command.to_string(),
+        ip: ip.to_string(),
+        timeout,
+        source,
+    })
+}
+
 pub fn exec_host_command_with_retries(command: &str, retries: u32, interval: Duration) -> bool {
     for _ in 0..retries {
         let s = exec_host_command_output(command).status;
@@ -1093,6 +1168,37 @@ impl Guest {
         )
     }
 
+    /// Waits until SSH to the guest becomes available using the
+    /// [default guest authentication] and the default guest IP.
+    ///
+    /// [default guest authentication]: default_guest_auth
+    pub fn wait_for_ssh(&self, timeout: Duration) -> Result<(), WaitForSshError> {
+        wait_for_ssh(
+            "true",
+            &default_guest_auth(),
+            &self.network.guest_ip0,
+            timeout,
+        )
+        .map(|_| ())
+    }
+
+    /// Waits until the provided command succeeds via SSH on the guest using the
+    /// [default guest authentication] and the default guest IP.
+    ///
+    /// [default guest authentication]: default_guest_auth
+    pub fn wait_for_ssh_command(
+        &self,
+        command: &str,
+        timeout: Duration,
+    ) -> Result<String, WaitForSshError> {
+        wait_for_ssh(
+            command,
+            &default_guest_auth(),
+            &self.network.guest_ip0,
+            timeout,
+        )
+    }
+
     pub fn api_create_body(&self) -> String {
         let mut body = serde_json::json!({
             "cpus": {