haematite-cli 0.21.8

use std::num::NonZeroU32;
use std::os::unix::net::UnixListener;
use std::sync::mpsc;
use std::thread;
use std::time::Duration;

use serde_json::{Value, json};

use super::{Carried, carry};
use crate::serve::frame;
use crate::serve::protocol::{PROTOCOL, RECEIPT_CAPABILITY};

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

const LIMIT: NonZeroU32 = NonZeroU32::MIN.saturating_add(65_535);
const WAIT: Duration = Duration::from_secs(30);

/// A service double on `socket`: it answers one connection's frames in
/// order with `answers` and sends each request it read, as JSON, on the
/// returned channel.
fn service(
    socket: &std::path::Path,
    answers: Vec<Value>,
) -> Result<mpsc::Receiver<Value>, Box<dyn std::error::Error>> {
    let listener = UnixListener::bind(socket)?;
    let (sent, read) = mpsc::channel();
    thread::spawn(move || -> std::io::Result<()> {
        let (mut stream, _) = listener.accept()?;
        for answer in answers {
            let Some(bytes) = frame::read(&mut stream, LIMIT)? else {
                return Ok(());
            };
            let request: Value = serde_json::from_slice(&bytes)?;
            let id = request["id"].clone();
            if sent.send(request).is_err() {
                return Ok(());
            }
            let mut reply = answer;
            reply["id"] = id;
            frame::write(&mut stream, &serde_json::to_vec(&reply)?, LIMIT)?;
        }
        Ok(())
    });
    Ok(read)
}

/// A store's hello under BUSYSHARDNODEADLINE-001 R12: its version, the typed
/// receipt capability it agrees and its limits.
fn hello() -> Value {
    json!({"result": {
        "protocol": PROTOCOL, "capabilities": [RECEIPT_CAPABILITY],
        "max_page": 10, "max_frame": LIMIT.get(),
    }})
}

/// R2: the door presents the session's pass with session.pass on the
/// connection before it carries the verb, so nothing reaches a store
/// undecided.
#[test]
fn a_verb_is_carried_after_the_session_pass_is_presented() -> TestResult {
    let directory = tempfile::tempdir()?;
    let socket = directory.path().join("s.sock");
    let read = service(
        &socket,
        vec![
            hello(),
            json!({"result": {"holder": "clinician", "expires": 2000}}),
            json!({"result": {"entries": [], "next_after": null}}),
        ],
    )?;
    let carried = carry(
        &socket,
        WAIT,
        "the-pass",
        "branches.list",
        Some(&serde_json::value::to_raw_value(&json!({"limit": 10}))?),
    )?;
    let Carried::Answered(answer) = carried else {
        return Err(format!("the verb was not answered: {carried:?}").into());
    };
    assert_eq!(
        serde_json::from_str::<Value>(answer.get())?,
        json!({"entries": [], "next_after": null})
    );
    let asked: Vec<Value> = (0..3)
        .map(|_| read.recv_timeout(WAIT))
        .collect::<Result<_, _>>()?;
    assert_eq!(asked[0]["operation"]["method"], "hello");
    assert_eq!(
        asked[1]["operation"],
        json!({"method": "session.pass", "params": {"pass": "the-pass"}})
    );
    assert_eq!(asked[2]["operation"]["method"], "branches.list");
    Ok(())
}

/// A pass the store refuses ends the request there: the verb is never sent.
#[test]
fn a_refused_pass_is_answered_and_the_verb_is_never_sent() -> TestResult {
    let directory = tempfile::tempdir()?;
    let socket = directory.path().join("s.sock");
    let read = service(
        &socket,
        vec![
            hello(),
            json!({"error": {"code": "pass_expired", "message": "the Lys pass was refused"}}),
            // Answered only if the door sent the verb anyway.
            json!({"result": {"entries": [], "next_after": null}}),
        ],
    )?;
    let carried = carry(&socket, WAIT, "old-pass", "branches.list", None)?;
    let Carried::Refused(refusal) = carried else {
        return Err(format!("an expired pass was carried: {carried:?}").into());
    };
    assert_eq!(refusal.code, "pass_expired");
    assert_eq!(
        read.iter().count(),
        2,
        "the verb was sent after a refused pass"
    );
    Ok(())
}

/// The browser never presents a pass of its own: session.pass is the
/// door's, refused before any connection.
#[test]
fn the_browser_cannot_present_a_pass_of_its_own() -> TestResult {
    let directory = tempfile::tempdir()?;
    let socket = directory.path().join("absent.sock");
    let carried = carry(&socket, WAIT, "the-pass", "session.pass", None)?;
    let Carried::Refused(refusal) = carried else {
        return Err(format!("session.pass was carried: {carried:?}").into());
    };
    assert_eq!(refusal.code, "invalid_operation");
    Ok(())
}

/// Studio asks hello as a verb; it touches nothing, so the store's own hello
/// answers it and no pass is sent.
#[test]
fn hello_is_answered_with_the_stores_own_hello() -> TestResult {
    let directory = tempfile::tempdir()?;
    let socket = directory.path().join("s.sock");
    // A second answer, so a pass sent after hello would be read and counted.
    let read = service(
        &socket,
        vec![
            hello(),
            json!({"result": {"holder": "clinician", "expires": 2000}}),
        ],
    )?;
    let carried = carry(&socket, WAIT, "the-pass", "hello", None)?;
    let Carried::Answered(answer) = carried else {
        return Err(format!("hello was not answered: {carried:?}").into());
    };
    assert_eq!(
        serde_json::from_str::<Value>(answer.get())?["protocol"],
        "haem/1"
    );
    assert_eq!(read.iter().count(), 1, "more than hello was sent");
    Ok(())
}