carte 0.4.0

use std::io::{Read, Write};
use std::net::TcpStream;
use std::os::unix::process::ExitStatusExt;
use std::path::{Path, PathBuf};
use std::process::ExitStatus;
use std::sync::OnceLock;
use std::time::{Duration, Instant};

use carte::process::{IoMode, Output, ProcessHandle, SpawnSpec, Spawner};
use nix::sys::signal::{Signal, kill};
use nix::unistd::Pid;
use tokio::io::AsyncReadExt;

fn fixture() -> &'static Path {
    static PATH: OnceLock<PathBuf> = OnceLock::new();
    PATH.get_or_init(|| {
        let status = std::process::Command::new(env!("CARGO"))
            .args(["build", "--quiet", "-p", "fixture-app"])
            .status()
            .expect("cargo build fixture-app");
        assert!(status.success());
        let exe = std::env::current_exe().unwrap();
        exe.parent().unwrap().parent().unwrap().join("fixture-app")
    })
}

fn spawner() -> Spawner {
    Spawner::new(env!("CARGO_BIN_EXE_carte"))
}

fn spec() -> SpawnSpec {
    SpawnSpec::new(fixture().display().to_string(), std::env::temp_dir())
}

async fn read_until(output: &mut Output, seen: &mut Vec<u8>, needle: &str) {
    let found = tokio::time::timeout(Duration::from_secs(10), async {
        let mut buf = [0u8; 4096];
        while !String::from_utf8_lossy(seen).contains(needle) {
            let n = output.read(&mut buf).await.unwrap();
            assert!(n > 0, "EOF before {needle:?}");
            seen.extend_from_slice(&buf[..n]);
        }
    })
    .await;
    assert!(
        found.is_ok(),
        "timed out waiting for {needle:?}, got {:?}",
        String::from_utf8_lossy(seen)
    );
}

async fn read_to_end(output: &mut Output) -> String {
    let mut seen = Vec::new();
    tokio::time::timeout(Duration::from_secs(10), output.read_to_end(&mut seen))
        .await
        .expect("output EOF")
        .unwrap();
    String::from_utf8_lossy(&seen).into_owned()
}

async fn run_to_end(handle: &mut ProcessHandle) -> (ExitStatus, String) {
    let mut output = handle.take_output().unwrap();
    let (status, text) = tokio::join!(handle.wait(), read_to_end(&mut output));
    (status.unwrap(), text)
}

fn drain(mut output: Output) {
    tokio::spawn(async move { while output.read(&mut [0u8; 4096]).await.is_ok_and(|n| n > 0) {} });
}

fn alive(pid: u32) -> bool {
    kill(Pid::from_raw(pid as i32), None).is_ok()
}

async fn wait_dead(pid: u32) {
    let deadline = Instant::now() + Duration::from_secs(10);
    while alive(pid) {
        assert!(Instant::now() < deadline, "pid {pid} still alive");
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
}

async fn read_pidfile(path: &Path) -> u32 {
    let deadline = Instant::now() + Duration::from_secs(10);
    loop {
        if let Ok(s) = std::fs::read_to_string(path)
            && let Ok(pid) = s.trim().parse()
        {
            return pid;
        }
        assert!(
            Instant::now() < deadline,
            "no pidfile at {}",
            path.display()
        );
        tokio::time::sleep(Duration::from_millis(20)).await;
    }
}

async fn stop(mut handle: ProcessHandle) {
    handle.stop(Duration::from_secs(2)).await.unwrap();
}

#[tokio::test]
async fn pty_output_is_line_buffered_and_a_tty() {
    let mut handle = spawner()
        .spawn(
            &spec()
                .env("FIXTURE_LINES", "3")
                .env("FIXTURE_PARTIAL", "partial>"),
        )
        .await
        .unwrap();
    assert_ne!(handle.pid(), handle.shim_pid());
    let mut output = handle.take_output().unwrap();
    let mut seen = Vec::new();
    read_until(&mut output, &mut seen, "partial>").await;
    drain(output);
    let text = String::from_utf8_lossy(&seen);
    assert!(
        text.starts_with("tty: true\r\nline 1\r\nline 2\r\nline 3\r\n"),
        "{text:?}"
    );
    assert!(alive(handle.pid()));
    stop(handle).await;
}

#[tokio::test]
async fn pipe_mode_merges_output_without_a_tty() {
    let mut handle = spawner()
        .spawn(
            &spec()
                .io(IoMode::Pipes)
                .env("FIXTURE_LINES", "2")
                .env("FIXTURE_CRASH", "4"),
        )
        .await
        .unwrap();
    let (status, text) = run_to_end(&mut handle).await;
    assert!(text.contains("tty: false\nline 1\nline 2\n"), "{text:?}");
    assert!(text.contains("crashing with 4\n"), "{text:?}");
    assert_eq!(status.code(), Some(4));
}

#[tokio::test]
async fn crash_output_survives_exit_on_a_pty() {
    let mut handle = spawner()
        .spawn(&spec().env("FIXTURE_LINES", "50").env("FIXTURE_CRASH", "3"))
        .await
        .unwrap();
    let (status, text) = run_to_end(&mut handle).await;
    assert_eq!(status.code(), Some(3));
    assert!(text.contains("line 50\r\n"), "{text:?}");
    assert!(text.contains("crashing with 3"), "{text:?}");
}

#[tokio::test]
async fn pty_input_reaches_the_foreground_group() {
    let mut handle = spawner()
        .spawn(&spec().env("FIXTURE_ECHO_STDIN", "1"))
        .await
        .unwrap();
    let mut output = handle.take_output().unwrap();
    let mut seen = Vec::new();
    read_until(&mut output, &mut seen, "ready").await;
    handle.write_input(b"hello\n").await.unwrap();
    read_until(&mut output, &mut seen, "echo: hello").await;
    drain(output);
    stop(handle).await;
}

#[tokio::test]
async fn command_runs_with_given_env_and_cwd() {
    let dir = tempfile::tempdir().unwrap();
    let mut handle = spawner()
        .spawn(
            &SpawnSpec::new("echo \"$GREETING from $(pwd -P)\"", dir.path()).env("GREETING", "hi"),
        )
        .await
        .unwrap();
    let (status, text) = run_to_end(&mut handle).await;
    let cwd = dir.path().canonicalize().unwrap();
    assert_eq!(text, format!("hi from {}\r\n", cwd.display()));
    assert!(status.success());
}

#[tokio::test]
async fn serves_http_on_port() {
    for ipv6 in [false, true] {
        let port = std::net::TcpListener::bind("127.0.0.1:0")
            .unwrap()
            .local_addr()
            .unwrap()
            .port();
        let mut spec = spec()
            .env("PORT", port.to_string())
            .env("FIXTURE_NAME", "web")
            .env("FIXTURE_ENV", "PORT,MISSING");
        if ipv6 {
            spec = spec.env("FIXTURE_BIND", "::1");
        }
        let mut handle = spawner().spawn(&spec).await.unwrap();
        let mut output = handle.take_output().unwrap();
        read_until(&mut output, &mut Vec::new(), "ready").await;
        drain(output);

        let host = if ipv6 {
            format!("[::1]:{port}")
        } else {
            format!("127.0.0.1:{port}")
        };
        assert!(get(&host, "/up").ends_with("\r\n\r\nok"));
        let body = get(&host, "/");
        let expected = format!(
            r#"{{"name":"web","pid":{},"env":{{"PORT":"{port}","MISSING":null}}}}"#,
            handle.pid()
        );
        assert!(body.ends_with(&expected), "{body}");
        stop(handle).await;
    }
}

fn get(host: &str, path: &str) -> String {
    let mut stream = TcpStream::connect(host).unwrap();
    write!(stream, "GET {path} HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();
    let mut response = String::new();
    stream.read_to_string(&mut response).unwrap();
    response
}

#[tokio::test]
async fn stop_terminates_the_group() {
    let mut handle = spawner().spawn(&spec()).await.unwrap();
    drain(handle.take_output().unwrap());
    let pid = handle.pid();
    let started = Instant::now();
    let status = handle.stop(Duration::from_secs(5)).await.unwrap();
    assert_eq!(status.signal(), Some(libc::SIGTERM));
    assert!(started.elapsed() < Duration::from_secs(2));
    assert!(!alive(pid));
    assert!(handle.signal_group(Signal::SIGTERM).is_err());
}

#[tokio::test]
async fn stop_escalates_to_kill() {
    let mut handle = spawner()
        .spawn(&spec().env("FIXTURE_IGNORE_TERM", "1"))
        .await
        .unwrap();
    let mut output = handle.take_output().unwrap();
    read_until(&mut output, &mut Vec::new(), "ready").await;
    drain(output);
    let started = Instant::now();
    let status = handle.stop(Duration::from_millis(300)).await.unwrap();
    assert_eq!(status.signal(), Some(libc::SIGKILL));
    assert!(started.elapsed() >= Duration::from_millis(300));
}

#[tokio::test]
async fn sigint_is_delivered_to_the_group() {
    let mut handle = spawner().spawn(&spec()).await.unwrap();
    let mut output = handle.take_output().unwrap();
    read_until(&mut output, &mut Vec::new(), "ready").await;
    drain(output);
    handle.signal_group(Signal::SIGINT).unwrap();
    assert_eq!(handle.wait().await.unwrap().signal(), Some(libc::SIGINT));
}

#[tokio::test]
async fn shim_forwards_signals_to_the_group() {
    let dir = tempfile::tempdir().unwrap();
    let pidfile = dir.path().join("child.pid");
    let mut handle = spawner()
        .spawn(&spec().env("FIXTURE_CHILD_PIDFILE", &pidfile))
        .await
        .unwrap();
    drain(handle.take_output().unwrap());
    let child = read_pidfile(&pidfile).await;
    kill(Pid::from_raw(handle.shim_pid() as i32), Signal::SIGTERM).unwrap();
    assert_eq!(handle.wait().await.unwrap().signal(), Some(libc::SIGTERM));
    wait_dead(child).await;
}

#[tokio::test]
async fn dropping_the_handle_kills_the_whole_group() {
    let dir = tempfile::tempdir().unwrap();
    let pidfile = dir.path().join("child.pid");
    let handle = spawner()
        .spawn(
            &spec()
                .env("FIXTURE_CHILD_PIDFILE", &pidfile)
                .env("FIXTURE_IGNORE_TERM", "1"),
        )
        .await
        .unwrap();
    let (pid, shim) = (handle.pid(), handle.shim_pid());
    let child = read_pidfile(&pidfile).await;
    assert!(alive(child));
    drop(handle);
    wait_dead(child).await;
    wait_dead(pid).await;
    wait_dead(shim).await;
}

#[tokio::test]
async fn closing_the_lifeline_alone_kills_the_group() {
    let dir = tempfile::tempdir().unwrap();
    let pidfile = dir.path().join("child.pid");
    let mut handle = spawner()
        .spawn(&spec().env("FIXTURE_CHILD_PIDFILE", &pidfile))
        .await
        .unwrap();
    drain(handle.take_output().unwrap());
    let child = read_pidfile(&pidfile).await;
    handle.close_lifeline();
    let status = tokio::time::timeout(Duration::from_secs(10), handle.wait())
        .await
        .unwrap()
        .unwrap();
    assert!(!status.success());
    wait_dead(handle.pid()).await;
    wait_dead(child).await;
}

#[tokio::test]
async fn self_daemonising_child_escapes_the_group() {
    let dir = tempfile::tempdir().unwrap();
    let pidfile = dir.path().join("daemon.pid");
    let handle = spawner()
        .spawn(&spec().env("FIXTURE_DAEMON_PIDFILE", &pidfile))
        .await
        .unwrap();
    let pid = handle.pid();
    let daemon = read_pidfile(&pidfile).await;
    drop(handle);
    wait_dead(pid).await;
    let escaped = alive(daemon);
    let _ = kill(Pid::from_raw(daemon as i32), Signal::SIGKILL);
    assert!(
        escaped,
        "expected the setsid grandchild to survive the group kill"
    );
}

#[tokio::test]
async fn failing_command_reports_its_exit_code() {
    let mut handle = spawner()
        .spawn(&SpawnSpec::new("exit 7", "/"))
        .await
        .unwrap();
    assert_eq!(run_to_end(&mut handle).await.0.code(), Some(7));
    let mut handle = spawner()
        .spawn(&SpawnSpec::new("no-such-command-xyz", "/"))
        .await
        .unwrap();
    let (status, text) = run_to_end(&mut handle).await;
    assert_eq!(status.code(), Some(127));
    assert!(text.contains("no-such-command-xyz"), "{text:?}");
}

/// Plays the daemon for `sigkilled_daemon_takes_its_groups_down`; a no-op otherwise.
#[tokio::test]
async fn fake_daemon() {
    let Some(dir) = std::env::var_os("CARTE_TEST_FAKE_DAEMON") else {
        return;
    };
    let dir = PathBuf::from(dir);
    let mut handles = Vec::new();
    for (i, io) in [IoMode::default(), IoMode::Pipes].into_iter().enumerate() {
        let pidfile = dir.join(format!("child{i}.pid"));
        let mut handle = spawner()
            .spawn(
                &spec()
                    .io(io)
                    .env("FIXTURE_CHILD_PIDFILE", &pidfile)
                    .env("FIXTURE_IGNORE_TERM", "1"),
            )
            .await
            .unwrap();
        drain(handle.take_output().unwrap());
        std::fs::write(dir.join(format!("leader{i}.pid")), handle.pid().to_string()).unwrap();
        handles.push(handle);
    }
    std::future::pending::<()>().await;
}

#[tokio::test]
async fn sigkilled_daemon_takes_its_groups_down() {
    let dir = tempfile::tempdir().unwrap();
    fixture();
    let mut daemon = std::process::Command::new(std::env::current_exe().unwrap())
        .args(["--exact", "fake_daemon", "--nocapture"])
        .env("CARTE_TEST_FAKE_DAEMON", dir.path())
        .stdout(std::process::Stdio::null())
        .spawn()
        .unwrap();
    let mut pids = Vec::new();
    for i in 0..2 {
        pids.push(read_pidfile(&dir.path().join(format!("leader{i}.pid"))).await);
        pids.push(read_pidfile(&dir.path().join(format!("child{i}.pid"))).await);
    }
    assert!(pids.iter().all(|&pid| alive(pid)));
    daemon.kill().unwrap();
    daemon.wait().unwrap();
    for pid in pids {
        wait_dead(pid).await;
    }
}