haematite-cli 0.21.7

use super::{
    Door, Lys, NOW, ORIGIN, Site, TestResult, Value, answer, json, request, signed_in, site,
    written,
};
use crate::serve::frame;
use crate::serve::protocol::{PROTOCOL, RECEIPT_CAPABILITY};
use std::num::NonZeroU32;
use std::os::unix::net::UnixListener;
use std::path::Path;
use std::sync::mpsc;

const FRAME: NonZeroU32 = NonZeroU32::MIN.saturating_add(65_535);

/// A store double on `socket` for one follow: it answers hello, then the
/// pass with `pass`, then the follow with `first` and at once each of
/// `then`, under the follow's id, as a store pushes them. Each request it
/// read is sent, as JSON, on the returned channel.
fn follow_store(
    socket: &Path,
    pass: Value,
    first: Value,
    then: Vec<Value>,
) -> Result<mpsc::Receiver<Value>, Box<dyn std::error::Error>> {
    let listener = UnixListener::bind(socket)?;
    let (sent, read) = mpsc::channel();
    std::thread::spawn(move || -> std::io::Result<()> {
        let (mut stream, _) = listener.accept()?;
        let hello = json!({"result": {
            "protocol": PROTOCOL, "capabilities": [RECEIPT_CAPABILITY],
            "max_page": 10, "max_frame": FRAME.get(),
        }});
        let mut id = Value::Null;
        for answer in [hello, pass, first] {
            let Some(bytes) = frame::read(&mut stream, FRAME)? else {
                return Ok(());
            };
            let request: Value = serde_json::from_slice(&bytes)?;
            id = request["id"].clone();
            if sent.send(request).is_err() {
                return Ok(());
            }
            let mut reply = answer;
            reply["id"] = id.clone();
            frame::write(&mut stream, &serde_json::to_vec(&reply)?, FRAME)?;
        }
        for answer in then {
            let mut reply = answer;
            reply["id"] = id.clone();
            frame::write(&mut stream, &serde_json::to_vec(&reply)?, FRAME)?;
        }
        Ok(())
    });
    Ok(read)
}

fn head_at(sequence: &str) -> Value {
    json!({"branch": "main", "version": {"generation": "7", "sequence": sequence}})
}

/// A follow on a door whose one store is the double on `socket`.
fn follow_site(files: &Path, socket: &Path) -> Site {
    let mut site = site(files);
    site.stores.insert("wards".to_owned(), socket.to_owned());
    site
}

/// Fails while a head reaches the browser late, out of order or not as the
/// store sent it, or the stream outlives the store's refusal: `retry` once,
/// an event `head` per head frame, then the event `refused` that ends it;
/// the door says hello, presents the session's pass, then follows.
#[test]
fn a_follow_streams_an_event_per_head_then_the_refusal_that_ends_it() -> TestResult {
    let files = tempfile::tempdir()?;
    let socket = files.path().join("s.sock");
    let refusal = json!({"code": "pass_expired", "message": "the Lys pass was refused"});
    let read = follow_store(
        &socket,
        json!({"result": {"holder": "clinician", "expires": crate::serve::pass::now() + 600}}),
        json!({"result": head_at("1")}),
        vec![json!({"result": head_at("2")}), json!({"error": refusal})],
    )?;
    let site = follow_site(files.path(), &socket);
    let door = Door::new(Lys);
    let id = signed_in(&door, &site)?;
    let streamed = written(answer(
        &door,
        &site,
        &request(&format!(
            "GET /api/haem/wards/follow?branch=main HTTP/1.1\r\nOrigin: {ORIGIN}\r\nCookie: haem_session={id}\r\n\r\n"
        ))?,
        NOW,
    ))?;
    let (head, body) = streamed.split_once("\r\n\r\n").ok_or("no head")?;
    assert!(head.starts_with("HTTP/1.1 200 OK\r\n"), "{head}");
    assert!(
        head.contains("\r\nContent-Type: text/event-stream\r\n"),
        "{head}"
    );
    assert!(head.contains("\r\nCache-Control: no-store\r\n"), "{head}");
    assert!(!head.contains("Content-Length"), "{head}");
    assert!(!head.contains("Content-Encoding"), "{head}");
    assert_eq!(
        body,
        format!(
            "retry: 1000\nevent: head\ndata: {}\n\nevent: head\ndata: {}\n\nevent: refused\ndata: {refusal}\n\n",
            head_at("1"),
            head_at("2")
        )
    );
    let asked: Vec<Value> = read.iter().collect();
    assert_eq!(asked.len(), 3, "{asked:?}");
    assert_eq!(asked[0]["operation"]["method"], "hello");
    assert_eq!(
        asked[1]["operation"],
        json!({"method": "session.pass", "params": {"pass": "pass-for-c1"}})
    );
    assert_eq!(
        asked[2]["operation"],
        json!({"method": "head.follow", "params": {"branch": "main"}})
    );
    Ok(())
}

/// Fails while a follow the store refused is streamed, which a browser would
/// ask again every second: it is answered as JSON with a status, which ends
/// an `EventSource`, and no follow is sent after a refused pass.
#[test]
fn a_follow_refused_by_the_store_is_answered_and_never_streamed() -> TestResult {
    let files = tempfile::tempdir()?;
    let socket = files.path().join("s.sock");
    let read = follow_store(
        &socket,
        json!({"error": {"code": "pass_expired", "message": "the Lys pass was refused"}}),
        // Answered only if the door sent the follow anyway.
        json!({"result": head_at("1")}),
        Vec::new(),
    )?;
    let site = follow_site(files.path(), &socket);
    let door = Door::new(Lys);
    let id = signed_in(&door, &site)?;
    let refused = written(answer(
        &door,
        &site,
        &request(&format!(
            "GET /api/haem/wards/follow?branch=main HTTP/1.1\r\nCookie: haem_session={id}\r\n\r\n"
        ))?,
        NOW,
    ))?;
    assert!(refused.starts_with("HTTP/1.1 400 "), "{refused}");
    assert!(refused.contains("\"pass_expired\""), "{refused}");
    assert!(!refused.contains("event:"), "{refused}");
    assert_eq!(
        read.iter().count(),
        2,
        "a follow was sent after a refused pass"
    );
    Ok(())
}

/// Fails while a follow reaches a store without a session, from another
/// origin, for a store this door does not serve, or with no branch: each is
/// refused by name before any store is asked.
#[test]
fn a_follow_without_a_session_or_for_an_unknown_store_is_refused() -> TestResult {
    let files = tempfile::tempdir()?;
    let site = site(files.path());
    let door = Door::new(Lys);
    let id = signed_in(&door, &site)?;
    for (text, status, code) in [
        (
            "GET /api/haem/wards/follow?branch=main HTTP/1.1\r\n\r\n".to_owned(),
            "401",
            "session_unknown",
        ),
        (
            format!(
                "GET /api/haem/nowhere/follow?branch=main HTTP/1.1\r\nCookie: haem_session={id}\r\n\r\n"
            ),
            "404",
            "store_unknown",
        ),
        (
            format!(
                "GET /api/haem/wards/follow?branch=main HTTP/1.1\r\nOrigin: https://evil.example\r\nCookie: haem_session={id}\r\n\r\n"
            ),
            "403",
            "origin_refused",
        ),
        (
            format!("GET /api/haem/wards/follow HTTP/1.1\r\nCookie: haem_session={id}\r\n\r\n"),
            "400",
            "invalid_params",
        ),
    ] {
        let refused = written(answer(&door, &site, &request(&text)?, NOW))?;
        assert!(
            refused.starts_with(&format!("HTTP/1.1 {status} ")),
            "{text}: {refused}"
        );
        assert!(
            refused.contains(&format!("\"{code}\"")),
            "{text}: {refused}"
        );
    }
    Ok(())
}