bevy_net_backend 0.2.0

//! The real `RusshTransport` against the mock from `examples/mock_ssh_server.rs` (and a few
//! hostile raw TCP peers), all on 127.0.0.1 in this process. Throwaway keys and known_hosts files
//! are generated at runtime under `target/tmp`; nothing touches `~/.ssh` or an SSH agent.
//! Bounded: every wait has a frame limit.

use std::io::{Read, Write};
use std::net::TcpListener;
use std::path::PathBuf;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, Instant};

use bevy::prelude::*;
use bevy_headless_test::prelude::*;
use bevy_net_backend::*;

#[allow(dead_code)]
#[path = "../examples/mock_ssh_server.rs"]
mod mock_ssh_server;

use mock_ssh_server::{random_key, write_key, MockOptions, MockSshServer};

const USER: &str = "tester";

/// A fresh directory under `target/tmp` for one test's throwaway files.
fn scratch() -> PathBuf {
    static NEXT: AtomicUsize = AtomicUsize::new(0);
    let dir = PathBuf::from(env!("CARGO_TARGET_TMPDIR")).join(format!("ssh-loopback-{}-{}", std::process::id(), NEXT.fetch_add(1, Ordering::SeqCst)));
    std::fs::create_dir_all(&dir).unwrap_or_else(|e| panic!("{e}"));
    dir
}

/// A mock server, a client key file (encrypted with `passphrase` if given) and a known_hosts file
/// that lists the mock.
struct Setup {
    mock: MockSshServer,
    dir: PathBuf,
    key: PathBuf,
    known_hosts: PathBuf,
}

impl Setup {
    fn new(passphrase: Option<&str>) -> Self {
        Self::with(passphrase, MockOptions::default())
    }

    fn with(passphrase: Option<&str>, options: MockOptions) -> Self {
        let dir = scratch();
        let client = random_key();
        let key = dir.join("id_test");
        write_key(&client, &key, passphrase).unwrap_or_else(|e| panic!("{e}"));
        let mock = MockSshServer::start_with("127.0.0.1:0", USER, client.public_key().clone(), options).unwrap_or_else(|e| panic!("{e}"));
        let known_hosts = dir.join("known_hosts");
        std::fs::write(&known_hosts, format!("{}\n", mock.known_hosts_line("127.0.0.1"))).unwrap_or_else(|e| panic!("{e}"));
        Self { mock, dir, key, known_hosts }
    }

    fn target(&self) -> SshTarget {
        SshTarget::new("127.0.0.1", USER).with_port(self.mock.port()).with_auth(SshAuth::key_file(&self.key)).with_known_hosts_file(&self.known_hosts)
    }
}

fn app() -> TestApp {
    let mut app = TestApp::builder().frame_duration(Duration::from_millis(1)).real_pause(Duration::from_millis(2)).build();
    app.add_plugins(BackendPlugin::default());
    app.watch::<SshStateChanged>().watch::<SshOutput>().watch::<SshFinished>();
    #[cfg(feature = "sftp")]
    app.watch::<SftpFinished>().watch::<SftpProgress>();
    app
}

fn ssh(app: &TestApp) -> &SshClient {
    app.world().resource::<SshClient>()
}

/// Connect `target` as "main" and wait until it is connected or refused; the final info.
fn connect(app: &mut TestApp, target: SshTarget) -> SshConnectionInfo {
    ssh(app).connect("main", target);
    app.step();
    app.run_until(|world| world.resource::<SshConnections>().state("main").is_some_and(|s| s != SshState::Connecting), 5000);
    app.world().resource::<SshConnections>().get("main").cloned().unwrap_or_else(|| panic!("no connection info"))
}

fn connected(app: &mut TestApp, target: SshTarget) {
    let info = connect(app, target);
    assert_eq!(info.state, SshState::Connected, "{:?}", info.last_error);
}

/// Run and wait for the one answer; also the stdout and stderr text.
fn run(app: &mut TestApp, command: impl Into<SshCommand>) -> (SshFinished, String, String) {
    let id = ssh(app).run("main", command);
    wait(app, id)
}

fn wait(app: &mut TestApp, id: RequestId) -> (SshFinished, String, String) {
    app.step();
    app.run_until(|world| world.resource::<InFlight>().describe(id).is_none(), 5000);
    app.step();
    let mut answers: Vec<SshFinished> = app.all_messages::<SshFinished>().into_iter().filter(|f| f.id == id).collect();
    assert_eq!(answers.len(), 1, "{answers:?}");
    let text = |stream| app.all_messages::<SshOutput>().into_iter().filter(|o| o.id == id && o.stream == stream).map(|o| o.text()).collect::<String>();
    let (stdout, stderr) = (text(SshStream::Stdout), text(SshStream::Stderr));
    (answers.remove(0), stdout, stderr)
}

fn wait_until(mut done: impl FnMut() -> bool, limit: Duration) -> bool {
    let start = Instant::now();
    while start.elapsed() < limit {
        if done() {
            return true;
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    done()
}

#[test]
fn commands_run_and_report_output_and_exit() {
    let setup = Setup::new(None);
    let mut app = app();
    connected(&mut app, setup.target());
    let info = app.world().resource::<SshConnections>().get("main").cloned().unwrap_or_else(|| panic!("no info"));
    assert_eq!(info.fingerprint.as_deref(), Some(setup.mock.fingerprint().as_str()));

    let (answer, out, _) = run(&mut app, "echo hello world");
    assert_eq!((answer.started, out.as_str()), (Some(true), "hello world\n"));
    assert!(answer.result.as_ref().is_ok_and(SshExit::success), "{answer:?}");
    let (answer, out, _) = run(&mut app, "whoami");
    assert_eq!(out, "tester\n");
    assert_eq!(answer.result.map(|e| e.stdout_bytes).ok(), Some(7));
    let (answer, out, err) = run(&mut app, "fail");
    assert_eq!((out.as_str(), err.as_str()), ("", "mock: failed as asked\n"));
    assert_eq!(answer.result.map(|e| e.status).ok(), Some(Some(3)));
    let (answer, _, _) = run(&mut app, "signal");
    assert_eq!(answer.result.map(|e| (e.status, e.signal)).ok(), Some((None, Some("KILL".to_string()))));
    let (answer, _, _) = run(&mut app, "noexit");
    assert_eq!(answer.result.map(|e| (e.status, e.signal)).ok(), Some((None, None)));
    let (answer, out, _) = run(&mut app, SshCommand::new("cat").with_stdin(b"piped in".to_vec()));
    assert_eq!(out, "piped in");
    assert!(answer.result.is_ok_and(|e| e.success()));
    assert!(setup.mock.stats().execs.load(Ordering::SeqCst) >= 6);
}

#[test]
fn a_refused_exec_says_it_never_started() {
    let setup = Setup::new(None);
    let mut app = app();
    connected(&mut app, setup.target());
    let (answer, _, _) = run(&mut app, "refuse");
    assert!(matches!(&answer.result, Err(BackendError::Ssh(why)) if why.contains("refused")), "{answer:?}");
    assert_eq!(answer.started, Some(false));
}

#[test]
fn a_timeout_stops_the_command_and_says_it_started() {
    let setup = Setup::new(None);
    let mut app = app();
    connected(&mut app, setup.target());
    let started = Instant::now();
    let (answer, _, _) = run(&mut app, SshCommand::new("sleep 30000").with_timeout(Duration::from_millis(300)));
    assert!(started.elapsed() < Duration::from_secs(5), "took {:?}", started.elapsed());
    assert!(matches!(&answer.result, Err(BackendError::Timeout(why)) if why.contains("TERM")), "{answer:?}");
    assert_eq!(answer.started, Some(true));
    // The remote side got the TERM signal and the channel close.
    assert!(wait_until(|| setup.mock.stats().signals.load(Ordering::SeqCst) >= 1, Duration::from_secs(3)));
    // The connection is still good.
    let (answer, out, _) = run(&mut app, "echo still here");
    assert!(answer.result.is_ok() && out == "still here\n");
}

#[test]
fn cancel_stops_a_running_command() {
    let setup = Setup::new(None);
    let mut app = app();
    connected(&mut app, setup.target());
    let id = ssh(&app).run("main", "hang");
    app.step();
    assert!(wait_until(|| setup.mock.stats().execs.load(Ordering::SeqCst) >= 1, Duration::from_secs(3)));
    app.step_n(20);
    ssh(&app).cancel(id);
    let (answer, _, _) = wait(&mut app, id);
    assert_eq!((answer.result.err(), answer.started), (Some(BackendError::Cancelled), Some(true)));
    assert!(wait_until(|| setup.mock.stats().closes.load(Ordering::SeqCst) >= 1, Duration::from_secs(3)), "the channel was not closed");
}

#[test]
fn output_over_the_limit_stops_the_command() {
    let setup = Setup::new(None);
    let mut app = app();
    connected(&mut app, setup.target());
    let (answer, out, _) = run(&mut app, SshCommand::new("flood 1000000").with_max_output_bytes(64 * 1024));
    assert!(matches!(answer.result, Err(BackendError::BodyTooLarge { limit: 65_536, .. })), "{answer:?}");
    assert!(out.len() <= 64 * 1024, "{} bytes delivered", out.len());
    // Within the limit it all arrives.
    let (answer, out, _) = run(&mut app, "flood 200000");
    assert!(answer.result.is_ok());
    assert_eq!(out.len(), 200_000);
}

#[test]
fn more_commands_than_channels_wait_for_a_free_one() {
    let setup = Setup::new(None);
    let mut app = app();
    connected(&mut app, setup.target().with_max_channels(3));
    let ids: Vec<RequestId> = (0..9).map(|i| ssh(&app).run("main", format!("echo {i}"))).collect();
    app.step();
    app.run_until(|world| world.resource::<InFlight>().is_empty(), 5000);
    app.step();
    for id in ids {
        let answers: Vec<SshFinished> = app.all_messages::<SshFinished>().into_iter().filter(|f| f.id == id).collect();
        assert_eq!(answers.len(), 1);
        assert!(answers[0].result.is_ok(), "{:?}", answers[0].result);
    }
}

#[test]
fn host_keys_are_checked_strictly() {
    let setup = Setup::new(None);
    // Empty known_hosts: unknown, never trusted on first use, and the file is never written.
    let empty = setup.dir.join("empty_known_hosts");
    std::fs::write(&empty, "").unwrap_or_else(|e| panic!("{e}"));
    let mut app = app();
    let target = SshTarget::new("127.0.0.1", USER).with_port(setup.mock.port()).with_auth(SshAuth::key_file(&setup.key)).with_known_hosts_file(&empty);
    let info = connect(&mut app, target);
    assert!(
        matches!(&info.last_error, Some(BackendError::HostKey { problem: HostKeyProblem::Unknown, fingerprint, .. }) if *fingerprint == setup.mock.fingerprint()),
        "{info:?}"
    );
    assert_eq!(std::fs::read_to_string(&empty).unwrap_or_default(), "");
    // Another key for the host: changed.
    let other = random_key().public_key().to_openssh().unwrap_or_default();
    let changed = setup.dir.join("changed_known_hosts");
    std::fs::write(&changed, format!("[127.0.0.1]:{} {other}\n", setup.mock.port())).unwrap_or_else(|e| panic!("{e}"));
    let mut app2 = self::app();
    let target = SshTarget::new("127.0.0.1", USER).with_port(setup.mock.port()).with_auth(SshAuth::key_file(&setup.key)).with_known_hosts_file(&changed);
    let info = connect(&mut app2, target);
    assert!(matches!(info.last_error, Some(BackendError::HostKey { problem: HostKeyProblem::Changed, .. })), "{info:?}");
    // Revoked, even with the right line present.
    let revoked = setup.dir.join("revoked_known_hosts");
    let mock_key = setup.mock.known_hosts_line("127.0.0.1");
    let key_part = mock_key.split_once(' ').map(|(_, k)| k.to_string()).unwrap_or_default();
    std::fs::write(&revoked, format!("{mock_key}\n@revoked * {key_part}\n")).unwrap_or_else(|e| panic!("{e}"));
    let mut app3 = self::app();
    let target = SshTarget::new("127.0.0.1", USER).with_port(setup.mock.port()).with_auth(SshAuth::key_file(&setup.key)).with_known_hosts_file(&revoked);
    let info = connect(&mut app3, target);
    assert!(matches!(info.last_error, Some(BackendError::HostKey { problem: HostKeyProblem::Revoked, .. })), "{info:?}");
    // None of them got as far as logging in.
    assert_eq!(setup.mock.stats().logins.load(Ordering::SeqCst), 0);
}

#[test]
fn a_pinned_fingerprint_is_enough_and_a_wrong_one_is_refused() {
    let setup = Setup::new(None);
    let mut app = app();
    let target = SshTarget::new("127.0.0.1", USER).with_port(setup.mock.port()).with_auth(SshAuth::key_file(&setup.key));
    connected(&mut app, target.clone().trust_host_key_fingerprint(setup.mock.fingerprint()));
    let mut app2 = self::app();
    let wrong = random_key().public_key().fingerprint(Default::default()).to_string();
    let info = connect(&mut app2, target.trust_host_key_fingerprint(wrong));
    assert!(matches!(info.last_error, Some(BackendError::HostKey { problem: HostKeyProblem::Unknown, .. })), "{info:?}");
}

#[test]
fn authentication_failures_are_clear_and_never_show_secrets() {
    let setup = Setup::new(Some("fake-passphrase-1"));
    // The encrypted key with its passphrase works.
    let mut app = app();
    let target = |auth: SshAuth| SshTarget::new("127.0.0.1", USER).with_port(setup.mock.port()).with_auth(auth).with_known_hosts_file(&setup.known_hosts);
    connected(&mut app, target(SshAuth::key_file_with_passphrase(&setup.key, "fake-passphrase-1")));
    // A wrong passphrase, no passphrase, a missing file, another key: `AuthFailed`, and the
    // passphrase is in no text.
    let other = setup.dir.join("id_other");
    write_key(&random_key(), &other, None).unwrap_or_else(|e| panic!("{e}"));
    let cases = [
        SshAuth::key_file_with_passphrase(&setup.key, "fake-wrong-passphrase"),
        SshAuth::key_file(&setup.key),
        SshAuth::key_file(setup.dir.join("no-such-key")),
        SshAuth::key_file(&other),
    ];
    for auth in cases {
        let mut app = self::app();
        let info = connect(&mut app, target(auth));
        let error = info.last_error.unwrap_or_else(|| panic!("connected"));
        assert!(matches!(error, BackendError::AuthFailed(_)), "{error:?}");
        let text = format!("{error} {error:?}");
        assert!(!text.contains("fake-wrong-passphrase") && !text.contains("fake-passphrase-1"), "{text}");
        assert!(!text.contains(&*setup.dir.to_string_lossy()), "a full path leaked: {text}");
    }
}

#[test]
fn an_ssh_config_alias_is_resolved() {
    let setup = Setup::new(None);
    let config = setup.dir.join("config");
    let text = format!(
        "Host mock-box\n  HostName 127.0.0.1\n  Port {}\n  User {USER}\n  IdentityFile {}\n",
        setup.mock.port(),
        setup.key.to_string_lossy().replace('\\', "/")
    );
    std::fs::write(&config, text).unwrap_or_else(|e| panic!("{e}"));
    let mut app = app();
    connected(&mut app, SshTarget::from_ssh_config_file(&config, "mock-box").with_known_hosts_file(&setup.known_hosts));
    let (answer, out, _) = run(&mut app, "whoami");
    assert!(answer.result.is_ok() && out == "tester\n");
}

#[test]
fn a_refused_port_fails_fast() {
    let port = {
        let listener = TcpListener::bind("127.0.0.1:0").unwrap_or_else(|e| panic!("{e}"));
        listener.local_addr().map(|a| a.port()).unwrap_or(1)
    };
    let mut app = app();
    let dir = scratch();
    let started = Instant::now();
    let info = connect(
        &mut app,
        SshTarget::new("127.0.0.1", USER).with_port(port).with_auth(SshAuth::key_file(dir.join("none"))).with_known_hosts_file(dir.join("kh")),
    );
    assert!(matches!(info.last_error, Some(BackendError::Network(_)) | Some(BackendError::InvalidRequest(_))), "{info:?}");
    assert!(started.elapsed() < Duration::from_secs(5));
}

#[test]
fn a_server_without_strict_kex_connects_through_aes_gcm() {
    // The reviewer's scenario: no strict key exchange, but AES-GCM (and ChaCha20) offered.
    let setup = Setup::with(None, MockOptions { no_strict_kex: true, ..MockOptions::default() });
    let mut app = app();
    connected(&mut app, setup.target());
    let (answer, out, _) = run(&mut app, "echo terrapin-safe");
    assert!(answer.result.is_ok() && out == "terrapin-safe\n");
}

#[test]
fn chacha20_without_strict_kex_is_refused_unless_explicitly_allowed() {
    let only_chacha = MockOptions { no_strict_kex: true, ciphers: Some(vec!["chacha20-poly1305@openssh.com"]), ..MockOptions::default() };
    let setup = Setup::with(None, only_chacha);
    let mut app = app();
    let info = connect(&mut app, setup.target());
    assert!(matches!(&info.last_error, Some(BackendError::Ssh(why)) if why.contains("Terrapin") && why.contains("allow_terrapin_vulnerable")), "{info:?}");
    assert_eq!(setup.mock.stats().logins.load(Ordering::SeqCst), 0, "refused before authentication");
    let mut app2 = self::app();
    connected(&mut app2, setup.target().allow_terrapin_vulnerable(true));
    // With strict key exchange, ChaCha20 alone is fine.
    let strict = Setup::with(None, MockOptions { ciphers: Some(vec!["chacha20-poly1305@openssh.com"]), ..MockOptions::default() });
    let mut app3 = self::app();
    connected(&mut app3, strict.target());
}

#[test]
fn the_key_type_in_known_hosts_is_preferred_and_another_type_is_unknown_not_changed() {
    // The server has ed25519 AND ECDSA keys; known_hosts lists only its ECDSA key: that is used.
    let setup = Setup::with(None, MockOptions { ecdsa_host_key: true, ..MockOptions::default() });
    let ecdsa = setup.mock.host_keys().get(1).cloned().unwrap_or_else(|| panic!("no ecdsa key"));
    let known = setup.dir.join("ecdsa_known_hosts");
    std::fs::write(&known, format!("{}\n", setup.mock.known_hosts_line_for("127.0.0.1", &ecdsa))).unwrap_or_else(|e| panic!("{e}"));
    let target = SshTarget::new("127.0.0.1", USER).with_port(setup.mock.port()).with_auth(SshAuth::key_file(&setup.key));
    let mut app = app();
    connected(&mut app, target.clone().with_known_hosts_file(&known));
    let fingerprint = app.world().resource::<SshConnections>().get("main").and_then(|c| c.fingerprint.clone());
    assert_eq!(fingerprint, Some(ecdsa.fingerprint(Default::default()).to_string()));
    // A server offering only ECDSA while known_hosts has only an ed25519 key for it: unknown.
    let only = Setup::with(None, MockOptions { only_ecdsa_host_key: true, ..MockOptions::default() });
    let old = only.dir.join("old_type_known_hosts");
    let other = random_key().public_key().to_openssh().unwrap_or_default();
    std::fs::write(&old, format!("[127.0.0.1]:{} {other}\n", only.mock.port())).unwrap_or_else(|e| panic!("{e}"));
    let mut app2 = self::app();
    let info =
        connect(&mut app2, SshTarget::new("127.0.0.1", USER).with_port(only.mock.port()).with_auth(SshAuth::key_file(&only.key)).with_known_hosts_file(&old));
    assert!(matches!(info.last_error, Some(BackendError::HostKey { problem: HostKeyProblem::Unknown, .. })), "{info:?}");
}

#[test]
fn password_and_keyboard_interactive_logins_work_and_never_leak() {
    let options = MockOptions {
        password: Some("fake-password-7".into()),
        keyboard_interactive: Some(("fake-password-7".into(), "424242".into())),
        ..MockOptions::default()
    };
    let setup = Setup::with(None, options);
    let target = |auth: SshAuth| SshTarget::new("127.0.0.1", USER).with_port(setup.mock.port()).with_auth(auth).with_known_hosts_file(&setup.known_hosts);
    let mut app = app();
    connected(&mut app, target(SshAuth::password("fake-password-7")));
    let (answer, out, _) = run(&mut app, "whoami");
    assert!(answer.result.is_ok() && out == "tester\n");
    let mut app2 = self::app();
    connected(
        &mut app2,
        target(SshAuth::keyboard_interactive(SshPromptAnswers::new().answer_containing("password", "fake-password-7").answer_containing("code", "424242"))),
    );
    for (auth, what) in [
        (SshAuth::password("fake-wrong-9"), "password"),
        (
            SshAuth::keyboard_interactive(SshPromptAnswers::new().answer_containing("password", "fake-password-7").answer_containing("code", "000000")),
            "keyboard-interactive",
        ),
        (SshAuth::keyboard_interactive(SshPromptAnswers::new().answer_containing("password", "fake-password-7")), "keyboard-interactive"),
    ] {
        let mut app = self::app();
        let info = connect(&mut app, target(auth));
        let error = info.last_error.unwrap_or_else(|| panic!("connected"));
        let text = format!("{error} {error:?}");
        assert!(matches!(error, BackendError::AuthFailed(_)) && text.contains(what), "{text}");
        assert!(!text.contains("fake-password-7") && !text.contains("fake-wrong-9") && !text.contains("424242"), "{text}");
    }
}

#[test]
fn an_ssh_config_that_includes_itself_is_an_error_not_a_crash() {
    let setup = Setup::new(None);
    let config = setup.dir.join("config");
    let path = config.to_string_lossy().replace('\\', "/");
    std::fs::write(&config, format!("Include {path}\nHost mock-box\n  HostName 127.0.0.1\n")).unwrap_or_else(|e| panic!("{e}"));
    let mut app = app();
    let info = connect(&mut app, SshTarget::from_ssh_config_file(&config, "mock-box").with_known_hosts_file(&setup.known_hosts));
    assert!(matches!(&info.last_error, Some(BackendError::InvalidRequest(why)) if why.contains("deeper than 16")), "{info:?}");
}

/// A hostile peer on 127.0.0.1: accepts one connection and plays `behave` on it (bounded by the
/// test's own lifetime: the thread ends once the client is gone or after 20 s).
fn hostile(behave: fn(&mut std::net::TcpStream)) -> u16 {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap_or_else(|e| panic!("{e}"));
    let port = listener.local_addr().map(|a| a.port()).unwrap_or(0);
    std::thread::spawn(move || {
        if let Ok((mut stream, _)) = listener.accept() {
            let _ = stream.set_write_timeout(Some(Duration::from_secs(1)));
            let _ = stream.set_read_timeout(Some(Duration::from_millis(100)));
            behave(&mut stream);
        }
    });
    port
}

fn connect_hostile(port: u16) -> (SshConnectionInfo, Duration) {
    let dir = scratch();
    let known = dir.join("known_hosts");
    std::fs::write(&known, "").unwrap_or_else(|e| panic!("{e}"));
    let mut app = app();
    let started = Instant::now();
    let target = SshTarget::new("127.0.0.1", USER)
        .with_port(port)
        .with_auth(SshAuth::key_file(dir.join("none")))
        .with_known_hosts_file(known)
        .with_connect_timeout(Duration::from_secs(1));
    let info = connect(&mut app, target);
    (info, started.elapsed())
}

fn trickle(stream: &mut std::net::TcpStream, bytes: &[u8], every: Duration, limit: Duration) {
    let start = Instant::now();
    for byte in bytes.iter().cycle() {
        if start.elapsed() > limit || stream.write_all(&[*byte]).is_err() {
            return;
        }
        std::thread::sleep(every);
        let mut sink = [0u8; 256];
        let _ = stream.read(&mut sink);
    }
}

#[test]
fn a_silent_server_is_given_up_at_the_connect_deadline() {
    let port = hostile(|stream| {
        let start = Instant::now();
        let mut sink = [0u8; 256];
        while start.elapsed() < Duration::from_secs(20) {
            match stream.read(&mut sink) {
                Ok(0) => return,
                _ => std::thread::sleep(Duration::from_millis(20)),
            }
        }
    });
    let (info, took) = connect_hostile(port);
    assert!(matches!(&info.last_error, Some(BackendError::Timeout(why)) if why.contains("1s")), "{info:?}");
    assert!(took < Duration::from_secs(4), "took {took:?}");
}

#[test]
fn a_trickled_banner_cannot_stretch_the_connect_deadline() {
    // A valid-looking banner, one byte every 150 ms, forever.
    let port =
        hostile(|stream| trickle(stream, b"SSH-2.0-OpenSSH_9.9 trickle trickle trickle trickle\r\n", Duration::from_millis(150), Duration::from_secs(20)));
    let (info, took) = connect_hostile(port);
    assert!(matches!(info.last_error, Some(BackendError::Timeout(_))), "{info:?}");
    assert!(took < Duration::from_secs(4), "took {took:?}");
}

#[test]
fn a_banner_then_silence_during_key_exchange_hits_the_deadline() {
    let port = hostile(|stream| {
        let _ = stream.write_all(b"SSH-2.0-OpenSSH_9.9\r\n");
        let start = Instant::now();
        let mut sink = [0u8; 4096];
        while start.elapsed() < Duration::from_secs(20) {
            match stream.read(&mut sink) {
                Ok(0) => return,
                _ => std::thread::sleep(Duration::from_millis(20)),
            }
        }
    });
    let (info, took) = connect_hostile(port);
    assert!(matches!(info.last_error, Some(BackendError::Timeout(_))), "{info:?}");
    assert!(took < Duration::from_secs(4), "took {took:?}");
}

#[test]
fn a_huge_banner_is_an_error_not_a_hang() {
    let port = hostile(|stream| {
        let line = vec![b'A'; 64 * 1024];
        for _ in 0..30 {
            if stream.write_all(&line).is_err() {
                return;
            }
        }
    });
    let (info, took) = connect_hostile(port);
    assert!(info.last_error.is_some(), "{info:?}");
    assert!(took < Duration::from_secs(4), "took {took:?}");
}

#[test]
fn a_server_that_goes_away_is_reported_lost() {
    let setup = Setup::new(None);
    let mut app = app();
    connected(&mut app, setup.target().with_keepalive(Duration::from_secs(1), 2));
    let id = ssh(&app).run("main", "hang");
    app.step_n(20);
    drop(setup.mock);
    app.run_until(|world| world.resource::<SshConnections>().state("main") == Some(SshState::Disconnected), 5000);
    assert_eq!(app.world().resource::<SshConnections>().state("main"), Some(SshState::Disconnected));
    let (answer, _, _) = wait(&mut app, id);
    assert!(matches!(answer.result, Err(BackendError::Disconnected { .. })), "{answer:?}");
    let _ = (setup.dir, setup.key, setup.known_hosts);
}

#[test]
fn app_exit_answers_at_once_and_does_not_wait_for_the_server() {
    let setup = Setup::new(None);
    let mut app = app();
    connected(&mut app, setup.target());
    let id = ssh(&app).run("main", "hang");
    assert!(wait_until(
        || {
            app.step();
            setup.mock.stats().execs.load(Ordering::SeqCst) >= 1
        },
        Duration::from_secs(3)
    ));
    let late = ssh(&app).run("main", "echo never");
    let execs = setup.mock.stats().execs.load(Ordering::SeqCst);
    app.world_mut().write_message(AppExit::Success);
    let started = Instant::now();
    app.step();
    assert!(started.elapsed() < Duration::from_millis(500), "exit took {:?}", started.elapsed());
    let answers = app.all_messages::<SshFinished>();
    let find = |id| answers.iter().find(|f| f.id == id).map(|f| (f.result.clone().err(), f.started));
    assert_eq!(find(id), Some((Some(BackendError::Shutdown), Some(true))));
    assert_eq!(find(late), Some((Some(BackendError::Shutdown), Some(false))));
    std::thread::sleep(Duration::from_millis(300));
    assert_eq!(setup.mock.stats().execs.load(Ordering::SeqCst), execs, "a command of the exit frame reached the server");
}

#[test]
fn disconnect_closes_the_connection() {
    let setup = Setup::new(None);
    let mut app = app();
    connected(&mut app, setup.target());
    ssh(&app).disconnect("main");
    app.step_n(3);
    assert_eq!(app.world().resource::<SshConnections>().state("main"), Some(SshState::Disconnected));
    // The server sees the connection end (its session ends, so it accepts a new one later).
    let mut app2 = self::app();
    connected(&mut app2, setup.target());
}

#[cfg(feature = "sftp")]
mod sftp {
    use std::path::Path;

    use super::*;

    fn op(app: &mut TestApp, id: RequestId) -> SftpFinished {
        app.step();
        app.run_until(|world| world.resource::<InFlight>().describe(id).is_none(), 5000);
        app.step();
        let mut answers: Vec<SftpFinished> = app.all_messages::<SftpFinished>().into_iter().filter(|f| f.id == id).collect();
        assert_eq!(answers.len(), 1, "{answers:?}");
        answers.remove(0)
    }

    #[test]
    fn upload_list_download_rename_remove() {
        let setup = Setup::new(None);
        let mut app = app();
        connected(&mut app, setup.target());
        let data: Vec<u8> = (0..300_000u32).map(|i| (i % 251) as u8).collect();
        let id = ssh(&app).create_dir("main", "work");
        assert_eq!(op(&mut app, id).result, Ok(SftpOutcome::Done));
        let id = ssh(&app).upload("main", "work/data.bin", data.clone());
        let answer = op(&mut app, id);
        assert_eq!((answer.result, answer.started), (Ok(SftpOutcome::Uploaded { bytes: 300_000 }), Some(true)));
        assert_eq!(setup.mock.file("work/data.bin"), Some(data.clone()));
        assert!(app.all_messages::<SftpProgress>().iter().any(|p| p.id == id && p.done == 300_000));
        let id = ssh(&app).list_dir("main", "work");
        match op(&mut app, id).result {
            Ok(SftpOutcome::Listing(entries)) => {
                assert_eq!(entries.len(), 1);
                assert_eq!((entries[0].name.as_str(), entries[0].kind, entries[0].size), ("data.bin", SftpEntryKind::File, Some(300_000)));
            }
            other => panic!("{other:?}"),
        }
        let id = ssh(&app).download("main", "work/data.bin");
        assert_eq!(op(&mut app, id).result, Ok(SftpOutcome::Data(data.clone())));
        let local = setup.dir.join("downloaded.bin");
        let id = ssh(&app).download_file("main", "work/data.bin", &local);
        assert_eq!(op(&mut app, id).result, Ok(SftpOutcome::Downloaded { bytes: 300_000 }));
        assert_eq!(std::fs::read(&local).ok(), Some(data.clone()));
        let id = ssh(&app).upload_file("main", &local, "work/copy.bin");
        assert_eq!(op(&mut app, id).result, Ok(SftpOutcome::Uploaded { bytes: 300_000 }));
        let id = ssh(&app).rename("main", "work/copy.bin", "work/moved.bin");
        assert_eq!(op(&mut app, id).result, Ok(SftpOutcome::Done));
        assert_eq!(setup.mock.file("work/moved.bin"), Some(data));
        for id in [ssh(&app).remove_file("main", "work/data.bin"), ssh(&app).remove_file("main", "work/moved.bin")] {
            assert_eq!(op(&mut app, id).result, Ok(SftpOutcome::Done));
        }
        let id = ssh(&app).remove_dir("main", "work");
        assert_eq!(op(&mut app, id).result, Ok(SftpOutcome::Done));
    }

    #[test]
    fn errors_carry_the_servers_words_and_limits_hold() {
        let setup = Setup::new(None);
        let mut app = app();
        connected(&mut app, setup.target().with_max_transfer_bytes(64 * 1024));
        let id = ssh(&app).download("main", "missing.txt");
        assert!(matches!(op(&mut app, id).result, Err(BackendError::Ssh(why)) if why.starts_with("SFTP:")));
        setup.mock.put_file("big.bin", &vec![7u8; 200_000]);
        let id = ssh(&app).download("main", "big.bin");
        assert!(matches!(op(&mut app, id).result, Err(BackendError::BodyTooLarge { limit: 65_536, .. })));
        // A failed download to a file leaves neither the file nor its part file.
        let local = setup.dir.join("big.bin");
        let id = ssh(&app).download_file("main", "big.bin", &local);
        assert!(matches!(op(&mut app, id).result, Err(BackendError::BodyTooLarge { .. })));
        assert!(!local.exists() && !Path::new(&format!("{}.part", local.display())).exists());
        // Uploads over the limit, from memory and from a file, get the same answer and are
        // refused before anything is sent.
        let id = ssh(&app).upload("main", "too-big.bin", vec![0u8; 100_000]);
        let answer = op(&mut app, id);
        assert!(matches!(answer.result, Err(BackendError::RequestTooLarge { limit: 65_536, size: 100_000, .. })), "{:?}", answer.result);
        assert_eq!(answer.started, Some(false));
        let big_local = setup.dir.join("too-big-local.bin");
        std::fs::write(&big_local, vec![0u8; 100_000]).unwrap_or_else(|e| panic!("{e}"));
        let id = ssh(&app).upload_file("main", &big_local, "too-big-file.bin");
        let answer = op(&mut app, id);
        assert!(matches!(answer.result, Err(BackendError::RequestTooLarge { limit: 65_536, size: 100_000, .. })), "{:?}", answer.result);
        assert_eq!(answer.started, Some(false));
        assert_eq!(setup.mock.file("too-big.bin"), None);
        assert_eq!(setup.mock.file("too-big-file.bin"), None);
    }

    /// Pseudo-random file content (every byte position differs, so a misplaced chunk shows).
    fn content(len: usize) -> Vec<u8> {
        let mut state = 0x2545_F491_4F6C_DD1Du64;
        (0..len)
            .map(|_| {
                state ^= state << 13;
                state ^= state >> 7;
                state ^= state << 17;
                (state >> 32) as u8
            })
            .collect()
    }

    fn sha1_hex(bytes: &[u8]) -> String {
        use sha1::Digest;
        sha1::Sha1::digest(bytes).iter().map(|b| format!("{b:02x}")).collect()
    }

    /// SFTP handles the mock has open right now.
    fn open_handles(setup: &Setup) -> usize {
        setup.mock.stats().sftp_handles.load(Ordering::SeqCst)
    }

    /// Files in `dir` whose name ends with `.part`.
    fn part_files(dir: &Path) -> Vec<String> {
        std::fs::read_dir(dir)
            .map(|entries| entries.filter_map(Result::ok).map(|e| e.file_name().to_string_lossy().into_owned()).filter(|n| n.ends_with(".part")).collect())
            .unwrap_or_default()
    }

    /// Waits for the one answer, up to `limit` (big transfers in a debug build take a while).
    fn op_within(app: &mut TestApp, id: RequestId, limit: Duration) -> SftpFinished {
        app.step();
        let start = Instant::now();
        while app.world().resource::<InFlight>().describe(id).is_some() && start.elapsed() < limit {
            app.step();
        }
        app.step();
        let mut answers: Vec<SftpFinished> = app.all_messages::<SftpFinished>().into_iter().filter(|f| f.id == id).collect();
        assert_eq!(answers.len(), 1, "{answers:?}");
        answers.remove(0)
    }

    /// Progress of `id`: only grows, never past the total, the size the server reported.
    fn check_progress(app: &TestApp, id: RequestId, size: u64) -> Vec<u64> {
        let progress: Vec<SftpProgress> = app.all_messages::<SftpProgress>().into_iter().filter(|p| p.id == id).collect();
        let done: Vec<u64> = progress.iter().map(|p| p.done).collect();
        assert!(done.windows(2).all(|w| w[0] <= w[1]), "progress went backwards: {done:?}");
        assert!(progress.iter().all(|p| p.total == Some(size) && p.done <= size), "{progress:?}");
        done
    }

    /// The heavy one: a 256 MiB file to disk and back into memory, compared by SHA-1 and byte by
    /// byte; the progress is checked and the speed printed.
    #[test]
    fn a_256_mib_download_is_byte_exact() {
        const SIZE: usize = 256 * 1024 * 1024;
        let setup = Setup::new(None);
        let mut app = app();
        connected(&mut app, setup.target().with_max_transfer_bytes(512 * 1024 * 1024).with_sftp_timeout(Duration::from_secs(600)));
        let data = content(SIZE);
        let hash = sha1_hex(&data);
        setup.mock.put_file("big.bin", &data);
        let local = setup.dir.join("big.bin");
        let start = Instant::now();
        let id = ssh(&app).download_file("main", "big.bin", &local);
        let answer = op_within(&mut app, id, Duration::from_secs(600));
        let seconds = start.elapsed().as_secs_f64();
        assert_eq!((answer.result, answer.started), (Ok(SftpOutcome::Downloaded { bytes: SIZE as u64 }), Some(true)));
        let got = std::fs::read(&local).unwrap_or_else(|e| panic!("{e}"));
        assert_eq!(sha1_hex(&got), hash);
        assert!(got == data, "byte-for-byte");
        drop(got);
        let done = check_progress(&app, id, SIZE as u64);
        assert_eq!(done.last(), Some(&(SIZE as u64)), "the last progress is the whole file");
        assert!(!done.is_empty(), "progress while it ran: {done:?}");
        assert!(part_files(&setup.dir).is_empty());
        eprintln!("256 MiB to a file over loopback (debug build): {seconds:.1} s = {:.1} MB/s", SIZE as f64 / 1e6 / seconds);

        let start = Instant::now();
        let id = ssh(&app).download("main", "big.bin");
        let answer = op_within(&mut app, id, Duration::from_secs(600));
        let seconds = start.elapsed().as_secs_f64();
        match answer.result {
            Ok(SftpOutcome::Data(bytes)) => {
                assert_eq!(sha1_hex(&bytes), hash);
                assert!(bytes == data);
            }
            other => panic!("{:?}", other.map(|_| "not data")),
        }
        eprintln!("256 MiB into memory over loopback (debug build): {seconds:.1} s = {:.1} MB/s", SIZE as f64 / 1e6 / seconds);
    }

    #[test]
    fn small_empty_and_boundary_sizes_arrive_exactly() {
        let setup = Setup::new(None);
        let mut app = app();
        connected(&mut app, setup.target());
        for size in [0usize, 1, 2, 4095, 65_535, 65_536, 65_537, 16 * 65_536, 16 * 65_536 + 1, 3_000_017] {
            let data = content(size);
            let remote = format!("size-{size}.bin");
            setup.mock.put_file(&remote, &data);
            let id = ssh(&app).download("main", &remote);
            assert_eq!(op(&mut app, id).result, Ok(SftpOutcome::Data(data.clone())), "{size} bytes into memory");
            let local = setup.dir.join(&remote);
            let id = ssh(&app).download_file("main", &remote, &local);
            assert_eq!(op(&mut app, id).result, Ok(SftpOutcome::Downloaded { bytes: size as u64 }), "{size} bytes to a file");
            assert!(std::fs::read(&local).ok() == Some(data), "{size} bytes: content");
            let done = check_progress(&app, id, size as u64);
            assert_eq!(done.last(), Some(&(size as u64)), "{size} bytes: final progress");
        }
        assert!(part_files(&setup.dir).is_empty());
    }

    #[test]
    fn short_reads_from_the_server_still_give_the_whole_file() {
        let options = MockOptions { sftp_short_reads: true, ..MockOptions::default() };
        let setup = Setup::with(None, options);
        let mut app = app();
        connected(&mut app, setup.target());
        let data = content(5 * 1024 * 1024 + 333);
        setup.mock.put_file("short.bin", &data);
        let local = setup.dir.join("short.bin");
        let id = ssh(&app).download_file("main", "short.bin", &local);
        assert_eq!(op_within(&mut app, id, Duration::from_secs(120)).result, Ok(SftpOutcome::Downloaded { bytes: data.len() as u64 }));
        assert_eq!(std::fs::read(&local).map(|b| sha1_hex(&b)).ok(), Some(sha1_hex(&data)));
    }

    #[test]
    fn a_server_error_midway_leaves_no_file_and_no_part_file() {
        let options = MockOptions { sftp_fail_reads_at: Some(3 * 1024 * 1024), ..MockOptions::default() };
        let setup = Setup::with(None, options);
        let mut app = app();
        connected(&mut app, setup.target());
        setup.mock.put_file("broken.bin", &content(8 * 1024 * 1024));
        let local = setup.dir.join("broken.bin");
        let id = ssh(&app).download_file("main", "broken.bin", &local);
        let answer = op_within(&mut app, id, Duration::from_secs(60));
        assert!(matches!(&answer.result, Err(BackendError::Ssh(why)) if why.starts_with("SFTP:")), "{:?}", answer.result);
        assert_eq!(answer.started, Some(true));
        assert!(!local.exists(), "no file under the final name");
        assert!(wait_until(|| part_files(&setup.dir).is_empty(), Duration::from_secs(5)), "{:?}", part_files(&setup.dir));
        assert!(wait_until(|| open_handles(&setup) == 0, Duration::from_secs(5)), "the remote handle stayed open");
        let id = ssh(&app).download("main", "broken.bin");
        assert!(matches!(op_within(&mut app, id, Duration::from_secs(60)).result, Err(BackendError::Ssh(_))));
        // The connection and its SFTP session still work.
        setup.mock.put_file("fine.bin", b"still fine");
        let id = ssh(&app).download("main", "fine.bin");
        assert_eq!(op(&mut app, id).result, Ok(SftpOutcome::Data(b"still fine".to_vec())));
    }

    #[test]
    fn cancel_and_timeout_midway_leave_no_file_and_no_part_file() {
        let options = MockOptions { sftp_read_delay: Some(Duration::from_millis(15)), ..MockOptions::default() };
        let setup = Setup::with(None, options);
        let mut app = app();
        connected(&mut app, setup.target().with_sftp_timeout(Duration::from_secs(120)));
        setup.mock.put_file("slow.bin", &content(16 * 1024 * 1024));
        let local = setup.dir.join("slow.bin");
        let id = ssh(&app).download_file("main", "slow.bin", &local);
        app.step();
        // Cancel once some data has arrived.
        let start = Instant::now();
        while !app.all_messages::<SftpProgress>().iter().any(|p| p.id == id && p.done > 0) && start.elapsed() < Duration::from_secs(30) {
            app.step();
        }
        ssh(&app).cancel(id);
        let answer = op_within(&mut app, id, Duration::from_secs(30));
        assert_eq!(answer.result, Err(BackendError::Cancelled));
        assert!(!local.exists());
        assert!(wait_until(|| part_files(&setup.dir).is_empty(), Duration::from_secs(5)), "{:?}", part_files(&setup.dir));
        // The connection stays open: only the guard's close request can bring this to 0.
        assert!(wait_until(|| open_handles(&setup) == 0, Duration::from_secs(5)), "the remote handle stayed open after the cancel");

        // A timeout in the middle: the same, and the answer says it had started.
        let mut app2 = self::app();
        connected(&mut app2, setup.target().with_sftp_timeout(Duration::from_millis(400)));
        let id = ssh(&app2).download_file("main", "slow.bin", &local);
        let answer = op_within(&mut app2, id, Duration::from_secs(30));
        assert!(matches!(&answer.result, Err(BackendError::Timeout(why)) if !why.starts_with("not sent")), "{:?}", answer.result);
        assert!(!local.exists());
        assert!(wait_until(|| part_files(&setup.dir).is_empty(), Duration::from_secs(5)), "{:?}", part_files(&setup.dir));
        assert!(wait_until(|| open_handles(&setup) == 0, Duration::from_secs(5)), "the remote handle stayed open after the timeout");

        // The first connection still downloads after its cancelled transfer.
        setup.mock.put_file("after.bin", b"after the cancel");
        let id = ssh(&app).download("main", "after.bin");
        assert_eq!(op_within(&mut app, id, Duration::from_secs(30)).result, Ok(SftpOutcome::Data(b"after the cancel".to_vec())));
    }

    /// Starts a slow download of `remote` (8 MiB, 15 ms per read on the mock) to `local` and
    /// returns once its first bytes arrived.
    fn download_until_progress(app: &mut TestApp, remote: &str, local: &Path) -> RequestId {
        let id = ssh(app).download_file("main", remote, local);
        app.step();
        let start = Instant::now();
        while !app.all_messages::<SftpProgress>().iter().any(|p| p.id == id && p.done > 0) && start.elapsed() < Duration::from_secs(30) {
            app.step();
        }
        assert!(app.all_messages::<SftpProgress>().iter().any(|p| p.id == id && p.done > 0), "no progress within 30 s");
        id
    }

    #[test]
    fn a_connection_lost_midway_ends_a_transfer_disconnected_and_leaves_no_file() {
        let options = MockOptions { sftp_read_delay: Some(Duration::from_millis(15)), ..MockOptions::default() };
        let setup = Setup::with(None, options);
        let mut app = app();
        connected(&mut app, setup.target().with_sftp_timeout(Duration::from_secs(120)));
        setup.mock.put_file("slow.bin", &content(8 * 1024 * 1024));
        let local = setup.dir.join("slow.bin");
        let id = download_until_progress(&mut app, "slow.bin", &local);
        // The server goes away (TCP closed, no SSH disconnect message).
        setup.mock.drop_connections();
        let answer = op_within(&mut app, id, Duration::from_secs(30));
        assert!(matches!(&answer.result, Err(BackendError::Disconnected { sent: Some(true), .. })), "{:?}", answer.result);
        assert_eq!(answer.result.as_ref().err().and_then(BackendError::was_sent), Some(true));
        assert_eq!(answer.started, Some(true));
        assert!(!local.exists(), "no file under the final name");
        assert!(wait_until(|| part_files(&setup.dir).is_empty(), Duration::from_secs(5)), "{:?}", part_files(&setup.dir));
        // The connection is gone (no reconnect asked for): a new operation is never sent.
        app.run_until(|world| world.resource::<SshConnections>().state("main") == Some(SshState::Disconnected), 5000);
        let id = ssh(&app).list_dir("main", ".");
        let answer = op_within(&mut app, id, Duration::from_secs(10));
        assert!(matches!(&answer.result, Err(BackendError::Disconnected { sent: Some(false), .. })), "{:?}", answer.result);
        assert_eq!(answer.started, Some(false));
    }

    /// Only the SFTP channel ends (the connection stays): the transfer is answered by the
    /// transport itself, `Disconnected` with `sent: Some(true)` (not the SFTP library's internal
    /// "sender dropped"), and the next operation opens a new SFTP channel.
    #[test]
    fn an_sftp_channel_that_ends_midway_is_a_disconnect_and_the_next_operation_opens_a_new_one() {
        let options = MockOptions { sftp_close_channel_at: Some(1024 * 1024), ..MockOptions::default() };
        let setup = Setup::with(None, options);
        let mut app = app();
        connected(&mut app, setup.target());
        setup.mock.put_file("big.bin", &content(4 * 1024 * 1024));
        let local = setup.dir.join("big.bin");
        let id = ssh(&app).download_file("main", "big.bin", &local);
        let answer = op_within(&mut app, id, Duration::from_secs(30));
        assert!(
            matches!(&answer.result, Err(BackendError::Disconnected { reason, sent: Some(true), .. }) if reason.contains("SFTP channel ended")),
            "{:?}",
            answer.result
        );
        assert_eq!(answer.started, Some(true));
        assert!(!local.exists(), "no file under the final name");
        assert!(wait_until(|| part_files(&setup.dir).is_empty(), Duration::from_secs(5)), "{:?}", part_files(&setup.dir));
        assert_eq!(app.world().resource::<SshConnections>().state("main"), Some(SshState::Connected));
        setup.mock.put_file("small.bin", b"a new channel");
        let id = ssh(&app).download("main", "small.bin");
        assert_eq!(op_within(&mut app, id, Duration::from_secs(30)).result, Ok(SftpOutcome::Data(b"a new channel".to_vec())));
    }

    #[test]
    fn a_remote_file_cut_short_during_a_download_is_an_error_and_leaves_no_file() {
        let options = MockOptions { sftp_read_delay: Some(Duration::from_millis(15)), ..MockOptions::default() };
        let setup = Setup::with(None, options);
        let mut app = app();
        connected(&mut app, setup.target().with_sftp_timeout(Duration::from_secs(120)));
        let data = content(8 * 1024 * 1024);
        setup.mock.put_file("cut.bin", &data);
        let local = setup.dir.join("cut.bin");
        let id = download_until_progress(&mut app, "cut.bin", &local);
        // Truncated on the server while it is read (the handle stays open).
        setup.mock.put_file("cut.bin", &data[..1024 * 1024]);
        let answer = op_within(&mut app, id, Duration::from_secs(60));
        let expected = format!("expected {} bytes, its size when it was opened; received ", data.len());
        assert!(
            matches!(&answer.result, Err(BackendError::Ssh(why)) if why.contains(&expected) && why.contains("cut short during the download")),
            "{:?}",
            answer.result
        );
        assert_eq!(answer.started, Some(true));
        assert!(!local.exists(), "no file under the final name");
        assert!(wait_until(|| part_files(&setup.dir).is_empty(), Duration::from_secs(5)), "{:?}", part_files(&setup.dir));
        assert!(wait_until(|| open_handles(&setup) == 0, Duration::from_secs(5)), "the remote handle stayed open");
        // Into memory: the same answer for a file that is shorter than its size at open.
        setup.mock.put_file("cut2.bin", &data);
        let id = ssh(&app).download("main", "cut2.bin");
        app.step();
        let start = Instant::now();
        while !app.all_messages::<SftpProgress>().iter().any(|p| p.id == id && p.done > 0) && start.elapsed() < Duration::from_secs(30) {
            app.step();
        }
        setup.mock.put_file("cut2.bin", b"short");
        let answer = op_within(&mut app, id, Duration::from_secs(60));
        assert!(matches!(&answer.result, Err(BackendError::Ssh(why)) if why.contains("cut short")), "{:?}", answer.result);
    }

    #[test]
    fn files_that_report_size_zero_or_no_size_are_read_to_the_end() {
        let data = content(300_000);
        for reported in [Some(0), None] {
            let options = MockOptions { sftp_fstat_size: Some(reported), ..MockOptions::default() };
            let setup = Setup::with(None, options);
            let mut app = app();
            connected(&mut app, setup.target());
            setup.mock.put_file("special.bin", &data);
            let id = ssh(&app).download("main", "special.bin");
            assert_eq!(op(&mut app, id).result, Ok(SftpOutcome::Data(data.clone())), "size reported: {reported:?}");
            let local = setup.dir.join("special.bin");
            let id = ssh(&app).download_file("main", "special.bin", &local);
            assert_eq!(op(&mut app, id).result, Ok(SftpOutcome::Downloaded { bytes: data.len() as u64 }), "size reported: {reported:?}");
            assert_eq!(std::fs::read(&local).ok(), Some(data.clone()));
        }
    }
}