semora-cli 0.3.0

mod common;

use std::io::{BufRead, BufReader};
use std::process::Stdio;
use std::sync::mpsc;
use std::time::Duration;

use common::*;

/// Spawns `semora watch` and returns its stdout lines through a channel.
fn spawn_watch(cli: &Cli, args: &[&str]) -> (std::process::Child, mpsc::Receiver<String>) {
    let mut cmd = cli.command(args);
    cmd.stdout(Stdio::piped()).stderr(Stdio::inherit());
    let mut child = cmd.spawn().unwrap();
    let stdout = child.stdout.take().unwrap();
    let (tx, rx) = mpsc::channel();
    std::thread::spawn(move || {
        for line in BufReader::new(stdout).lines().map_while(Result::ok) {
            if tx.send(line).is_err() {
                break;
            }
        }
    });
    (child, rx)
}

async fn next_push(rx: &mpsc::Receiver<String>, secs: u64) -> Option<serde_json::Value> {
    let deadline = std::time::Instant::now() + Duration::from_secs(secs);
    loop {
        let left = deadline.saturating_duration_since(std::time::Instant::now());
        if left.is_zero() {
            return None;
        }
        // Poll without blocking the runtime that serves the test server.
        match rx.try_recv() {
            Ok(line) => {
                if let Ok(v) = serde_json::from_str::<serde_json::Value>(&line)
                    && v.get("rev").is_some()
                {
                    return Some(v);
                }
            }
            Err(mpsc::TryRecvError::Empty) => tokio::time::sleep(Duration::from_millis(50)).await,
            Err(mpsc::TryRecvError::Disconnected) => return None,
        }
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn watch_repushes_a_folder_once_per_burst() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    cli.write("site/index.md", "# One");
    cli.write("site/other.md", "# Other");
    let (mut child, rx) = spawn_watch(&cli, &["watch", "site", "--json"]);

    let first = next_push(&rx, 15).await.expect("initial push");
    assert_eq!(first["rev"], 1);
    // The push itself must not trigger another push.
    assert!(
        next_push(&rx, 2).await.is_none(),
        "no push without a change"
    );

    // A burst of edits becomes one push.
    for n in 0..4 {
        cli.write("site/index.md", format!("# Edit {n}"));
        tokio::time::sleep(Duration::from_millis(80)).await;
    }
    cli.write("site/other.md", "# Other changed");
    let second = next_push(&rx, 15).await.expect("push after edits");
    assert_eq!(second["rev"], 2);
    assert_eq!(second["id"], first["id"]);
    assert!(
        next_push(&rx, 2).await.is_none(),
        "the burst was coalesced into one push"
    );

    // Hidden files are ignored.
    cli.write("site/.notes", "scratch");
    assert!(next_push(&rx, 2).await.is_none());

    child.kill().unwrap();
    let _ = child.wait();
}

#[tokio::test(flavor = "multi_thread")]
async fn watch_a_single_file() {
    let server = Server::start().await;
    let cli = Cli::new(&server);
    cli.write("doc.md", "# Start");
    cli.write("unrelated.md", "# Unrelated");
    let (mut child, rx) = spawn_watch(&cli, &["watch", "doc.md", "--json"]);
    let first = next_push(&rx, 15).await.expect("initial push");
    cli.write("unrelated.md", "# Changed but not watched");
    assert!(
        next_push(&rx, 2).await.is_none(),
        "other files in the folder do not count"
    );
    cli.write("doc.md", "# Changed");
    let second = next_push(&rx, 15).await.expect("push after edit");
    assert_eq!(second["id"], first["id"]);
    assert_eq!(second["rev"], 2);
    child.kill().unwrap();
    let _ = child.wait();
}