haematite-cli 0.21.7

#![cfg(unix)]
#![forbid(unsafe_code)]

use std::error::Error;
use std::io::Write;
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::process::{Command, Output, Stdio};

use serde_json::{Value, json};

#[path = "support/served_child.rs"]
mod served_child;
use served_child::Running;

#[path = "support/stand_in.rs"]
mod support;
use support::{StandIn, answer, stand_in};

type TestResult = Result<(), Box<dyn Error>>;

/// A created service on `dir`, returned once it says it is ready.
fn start(dir: &Path) -> Result<(Running, PathBuf), Box<dyn Error>> {
    std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700))?;
    let config = dir.join("serve.json");
    std::fs::write(
        &config,
        serde_json::to_vec(&json!({
            "data_dir":dir.join("database"), "socket":dir.join("serve.sock"),
            "name":dir.file_name().and_then(|name| name.to_str()).ok_or("fixture name is not text")?,
            "branch_cache_bytes_per_shard":1_048_576, "max_frame":1_048_576,
            "max_clients":4, "max_page":20, "timeout_ms":2000,
            "create":{"shards":2,"cache_bytes_total":2_097_152,"executor_threads":2}
        }))?,
    )?;
    let running = Running::start(&config, true)?;
    Ok((running, config))
}

/// `haematite <command>` with `args`, and `input` on its standard input.
fn haematite(command: &str, args: &[&str], input: &str) -> Result<Output, Box<dyn Error>> {
    let mut child = Command::new(env!("CARGO_BIN_EXE_haematite"))
        .arg(command)
        .args(args)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()?;
    child
        .stdin
        .take()
        .ok_or("missing child stdin")?
        .write_all(input.as_bytes())?;
    Ok(child.wait_with_output()?)
}

fn text(bytes: &[u8]) -> Result<&str, Box<dyn Error>> {
    Ok(std::str::from_utf8(bytes)?)
}

fn socket_of(dir: &Path) -> Result<String, Box<dyn Error>> {
    Ok(dir
        .join("serve.sock")
        .to_str()
        .ok_or("socket path is not text")?
        .to_owned())
}

/// A reply refusing with `code`.
fn refuse(code: &str) -> String {
    format!(r#"{{"id":"haem-call","error":{{"code":"{code}","message":"stand-in","path":null}}}}"#)
}

fn requests(service: StandIn) -> Result<Vec<Value>, Box<dyn Error>> {
    let mut requests = Vec::new();
    for frame in service.frames()? {
        requests.push(serde_json::from_slice(
            &frame.ok_or("no request was sent")?,
        )?);
    }
    Ok(requests)
}

fn methods(requests: &[Value]) -> Vec<&str> {
    requests
        .iter()
        .filter_map(|request| request["operation"]["method"].as_str())
        .collect()
}

/// `haematite query` on `socket`'s main branch.
fn query(socket: &str, typed: &str, input: &str) -> Result<Output, Box<dyn Error>> {
    haematite(
        "query",
        &["--socket", socket, "--branch", "main", typed],
        input,
    )
}

/// Two rows as a service might send them, spacing and digits kept.
const ROW_A: &str = r#"{ "value": 1.00, "origin": {"collection":"c","id":"a"} }"#;
const ROW_B: &str = r#"{"value":9007199254740993,"origin":{"collection":"c","id":"b"}}"#;
const ROW_C: &str = r#"{"value":"c","origin":{"collection":"c","id":"c"}}"#;

/// A page of `rows` read at 1.4, ending with `next`.
fn page(rows: &[&str], next: &str) -> String {
    format!(
        concat!(
            r#"{{"selection":{{"branch":"main","version":{{"generation":"1","sequence":"4"}},"#,
            r#""roots":[]}},"rows":[{}],"next":{},"reading":"jq","collection":null,"#,
            r#""records_read":{}}}"#
        ),
        rows.join(","),
        next,
        rows.len()
    )
}

#[test]
fn one_page_prints_each_row_as_sent_and_sizes_the_page_from_hello() -> TestResult {
    let dir = tempfile::tempdir()?;
    let script = vec![vec![Some(answer(&page(&[ROW_A, ROW_B], "null")))]];
    let service = stand_in(dir.path(), 1_048_576, script)?;
    let output = query(&socket_of(dir.path())?, ".", "")?;
    assert_eq!(output.status.code(), Some(0), "{}", text(&output.stderr)?);
    assert_eq!(text(&output.stdout)?, format!("{ROW_A}\n{ROW_B}\n"));
    assert!(output.stderr.is_empty());
    let requests = requests(service)?;
    assert_eq!(
        requests[0]["operation"],
        json!({"method":"query.run", "params":{
            "selection":{"branch":"main"}, "query":".", "page":{"rows":20, "bytes":1_048_576},
        }})
    );
    Ok(())
}

#[test]
fn every_page_is_followed_at_the_first_pages_moment() -> TestResult {
    let dir = tempfile::tempdir()?;
    let script = vec![vec![
        Some(answer(&page(&[ROW_A, ROW_B], r#""after-b""#))),
        Some(answer(&page(&[ROW_C], "null"))),
    ]];
    let service = stand_in(dir.path(), 1_048_576, script)?;
    let output = query(&socket_of(dir.path())?, ".", "")?;
    assert_eq!(output.status.code(), Some(0), "{}", text(&output.stderr)?);
    assert_eq!(
        text(&output.stdout)?,
        format!("{ROW_A}\n{ROW_B}\n{ROW_C}\n")
    );
    let requests = requests(service)?;
    assert_eq!(methods(&requests), ["query.run", "query.run"]);
    let second = &requests[1]["operation"]["params"];
    assert_eq!(second["continuation"], "after-b");
    assert_eq!(
        second["selection"],
        json!({"branch":"main", "version":{"generation":"1", "sequence":"4"}})
    );
    Ok(())
}

#[test]
fn a_refusal_part_way_says_how_many_rows_were_printed() -> TestResult {
    let dir = tempfile::tempdir()?;
    let script = vec![vec![
        Some(answer(&page(&[ROW_A, ROW_B], r#""after-b""#))),
        Some(refuse("query_continuation_unavailable")),
    ]];
    let service = stand_in(dir.path(), 1_048_576, script)?;
    let output = query(&socket_of(dir.path())?, ".", "")?;
    let stderr = text(&output.stderr)?;
    assert_eq!(output.status.code(), Some(2), "{stderr}");
    assert_eq!(text(&output.stdout)?, format!("{ROW_A}\n{ROW_B}\n"));
    let mut lines = stderr.lines();
    let first = lines.next().ok_or("no refusal line")?;
    assert!(
        first.starts_with("haematite: refused query_continuation_unavailable: "),
        "{stderr}"
    );
    assert_eq!(
        lines.next(),
        Some(
            "haematite: stopped part-way: 2 rows were printed before this; the answer is not complete"
        )
    );
    assert_eq!(methods(&requests(service)?), ["query.run", "query.run"]);
    Ok(())
}

#[test]
fn a_close_part_way_is_not_known_and_says_how_many_rows_were_printed() -> TestResult {
    let dir = tempfile::tempdir()?;
    let script = vec![vec![Some(answer(&page(&[ROW_A], r#""after-a""#))), None]];
    let service = stand_in(dir.path(), 1_048_576, script)?;
    let output = query(&socket_of(dir.path())?, ".", "")?;
    let stderr = text(&output.stderr)?;
    assert_eq!(output.status.code(), Some(3), "{stderr}");
    assert_eq!(text(&output.stdout)?, format!("{ROW_A}\n"));
    assert!(
        stderr.starts_with("haematite: the outcome is not known: "),
        "{stderr}"
    );
    assert!(
        stderr.ends_with(
            "haematite: stopped part-way: 1 rows were printed before this; the answer is not complete\n"
        ),
        "{stderr}"
    );
    assert_eq!(methods(&requests(service)?), ["query.run", "query.run"]);
    Ok(())
}

#[test]
fn a_refusal_on_the_first_page_prints_no_rows_and_no_count() -> TestResult {
    let dir = tempfile::tempdir()?;
    let (_service, _) = start(dir.path())?;
    // The fixture configures no query engine, so the service refuses by name.
    let output = query(&socket_of(dir.path())?, ".", "")?;
    let stderr = text(&output.stderr)?;
    assert_eq!(output.status.code(), Some(2), "{stderr}");
    assert!(output.stdout.is_empty());
    assert_eq!(stderr.lines().count(), 1, "{stderr}");
    assert!(
        stderr.starts_with("haematite: refused query_engine_unavailable: "),
        "{stderr}"
    );
    Ok(())
}

#[test]
fn the_text_is_read_from_standard_input_with_dash() -> TestResult {
    let dir = tempfile::tempdir()?;
    let script = vec![vec![Some(answer(&page(&[ROW_A], "null")))]];
    let service = stand_in(dir.path(), 1_048_576, script)?;
    let text_in = "patients | '.name'\n";
    let output = query(&socket_of(dir.path())?, "-", text_in)?;
    assert_eq!(output.status.code(), Some(0), "{}", text(&output.stderr)?);
    let requests = requests(service)?;
    assert_eq!(requests[0]["operation"]["params"]["query"], text_in);
    Ok(())
}

#[test]
fn no_branch_is_a_usage_error() -> TestResult {
    let output = haematite("query", &["--socket", "/a.sock", "."], "")?;
    let stderr = text(&output.stderr)?;
    assert_eq!(output.status.code(), Some(1));
    assert!(output.stdout.is_empty());
    assert!(
        stderr.starts_with("haematite: name the branch with --branch\n"),
        "{stderr}"
    );
    assert!(stderr.contains("usage: haematite query"), "{stderr}");
    Ok(())
}

#[test]
fn an_answer_without_next_is_not_read_as_complete() -> TestResult {
    let dir = tempfile::tempdir()?;
    let cut = page(&[ROW_C], "null").replace(r#","next":null"#, "");
    let script = vec![vec![
        Some(answer(&page(&[ROW_A, ROW_B], r#""after-b""#))),
        Some(answer(&cut)),
    ]];
    let service = stand_in(dir.path(), 1_048_576, script)?;
    let output = query(&socket_of(dir.path())?, ".", "")?;
    let stderr = text(&output.stderr)?;
    assert_eq!(output.status.code(), Some(3), "{stderr}");
    assert_eq!(text(&output.stdout)?, format!("{ROW_A}\n{ROW_B}\n"));
    assert!(
        stderr.starts_with("haematite: the outcome is not known: "),
        "{stderr}"
    );
    assert!(stderr.contains("next"), "{stderr}");
    assert!(
        stderr.ends_with(
            "haematite: stopped part-way: 2 rows were printed before this; the answer is not complete\n"
        ),
        "{stderr}"
    );
    assert_eq!(methods(&requests(service)?), ["query.run", "query.run"]);
    Ok(())
}

#[test]
fn a_repeated_continuation_stops_as_not_known() -> TestResult {
    let dir = tempfile::tempdir()?;
    let script = vec![vec![
        Some(answer(&page(&[ROW_A], r#""again""#))),
        Some(answer(&page(&[ROW_B], r#""again""#))),
    ]];
    let service = stand_in(dir.path(), 1_048_576, script)?;
    let output = query(&socket_of(dir.path())?, ".", "")?;
    let stderr = text(&output.stderr)?;
    assert_eq!(output.status.code(), Some(3), "{stderr}");
    assert_eq!(text(&output.stdout)?, format!("{ROW_A}\n"));
    assert!(
        stderr.starts_with(
            "haematite: the outcome is not known: the service repeated the continuation"
        ),
        "{stderr}"
    );
    assert!(
        stderr.ends_with(
            "haematite: stopped part-way: 1 rows were printed before this; the answer is not complete\n"
        ),
        "{stderr}"
    );
    assert_eq!(methods(&requests(service)?), ["query.run", "query.run"]);
    Ok(())
}