boxmaker/crates/loopd/tests/http.rs

//! Tests for the HTTP client, against the fake server. Do not edit.

mod support;

use loopd::http::{Connection, Head, HttpError, MAX_HEAD, Request, read_capped};
use std::time::Duration;
use support::{FakeServer, Reply};

const GET_PROPS: Request<'static> = Request {
    method: "GET",
    path: "/props?model=m",
    body: None,
};

fn get(server: &FakeServer) -> (Connection, Result<Head, HttpError>) {
    let mut conn = Connection::open(&server.socket).unwrap();
    conn.set_read_timeout(Duration::from_secs(5)).unwrap();
    conn.send(&GET_PROPS).unwrap();
    let head = conn.read_head();
    (conn, head)
}

fn get_body(server: &FakeServer) -> Result<Vec<u8>, HttpError> {
    let (mut conn, head) = get(server);
    let head = head?;
    let mut body = conn.body(&head)?;
    read_capped(&mut body, 4 * 1024 * 1024)
}

#[test]
fn reads_a_recorded_chunked_response() {
    let server = FakeServer::start();
    server.route("/props", vec![Reply::fixture("props")]);
    let (mut conn, head) = get(&server);
    let head = head.unwrap();
    assert_eq!(head.status, 200);
    assert_eq!(head.header("transfer-encoding"), Some("chunked"));
    assert_eq!(
        head.header("content-type"),
        Some("application/json; charset=utf-8")
    );
    // The real server sends this header with an empty value, and `Connection` twice.
    assert_eq!(head.header("access-control-allow-origin"), Some(""));
    assert_eq!(
        head.headers
            .iter()
            .filter(|(n, _)| n == "connection")
            .count(),
        2
    );
    assert_eq!(head.header("x-missing"), None);

    let bytes = read_capped(&mut conn.body(&head).unwrap(), 4 * 1024 * 1024).unwrap();
    let props: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
    let template = props["chat_template"].as_str().unwrap();
    // The hash recorded in docs/inference-contract.md. If the de-chunking were off by a byte,
    // this would not match.
    assert_eq!(
        proto::sha256(template.as_bytes()).unwrap().to_hex(),
        "f55f52930aa8bf44ab5cb85f99370fcc3c56e9a85640b812086d5330bce5d86b"
    );
}

#[test]
fn the_result_does_not_depend_on_how_the_bytes_arrive() {
    let whole = {
        let server = FakeServer::start();
        server.route("/props", vec![Reply::fixture("props")]);
        get_body(&server).unwrap()
    };
    for piece in [1, 2, 3, 7, 64, 1000] {
        let server = FakeServer::start();
        server.route("/props", vec![Reply::fixture("props").trickle(piece, 0)]);
        assert_eq!(get_body(&server).unwrap(), whole, "in pieces of {piece}");
    }
}

#[test]
fn reads_a_body_with_a_content_length() {
    let server = FakeServer::start();
    server.route("/props", vec![Reply::fixture("unknown_model")]);
    let (mut conn, head) = get(&server);
    let head = head.unwrap();
    assert_eq!(head.status, 400);
    assert_eq!(head.header("content-length"), Some("97"));
    let bytes = read_capped(&mut conn.body(&head).unwrap(), 4096).unwrap();
    let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
    assert_eq!(body["error"]["message"], "model 'no-such-model' not found");
}

#[test]
fn reads_a_body_that_runs_to_the_close() {
    let server = FakeServer::start();
    server.route(
        "/props",
        vec![Reply::raw("HTTP/1.0 200 OK\r\nX-A: b\r\n\r\nto the end")],
    );
    assert_eq!(get_body(&server).unwrap(), b"to the end");
}

#[test]
fn an_empty_body_is_empty_in_every_framing() {
    for raw in [
        "HTTP/1.1 204 No Content\r\nContent-Length: 0\r\n\r\n",
        "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n0\r\n\r\n",
        "HTTP/1.1 200 OK\r\n\r\n",
    ] {
        let server = FakeServer::start();
        server.route("/props", vec![Reply::raw(raw)]);
        assert_eq!(get_body(&server).unwrap(), b"", "{raw:?}");
    }
}

#[test]
fn chunk_extensions_and_trailers_are_skipped() {
    let raw = "HTTP/1.1 200 OK\r\nTransfer-Encoding: Chunked\r\n\r\n\
               5;ext=1\r\nhello\r\n1\r\n,\r\n6\r\n world\r\n0\r\nX-Trailer: t\r\n\r\n";
    let server = FakeServer::start();
    server.route("/props", vec![Reply::raw(raw)]);
    assert_eq!(get_body(&server).unwrap(), b"hello, world");
}

#[test]
fn sends_a_well_formed_request() {
    let server = FakeServer::start();
    server.route("/tokenize", vec![Reply::fixture("tokenize")]);
    let mut conn = Connection::open(&server.socket).unwrap();
    let body = br#"{"content":"x"}"#;
    conn.send(&Request {
        method: "POST",
        path: "/tokenize",
        body: Some(body),
    })
    .unwrap();
    conn.read_head().unwrap();
    let sent = &server.requests()[0];
    assert_eq!(
        (sent.method.as_str(), sent.target.as_str()),
        ("POST", "/tokenize")
    );
    assert_eq!(sent.body, body);
    let has = |line: &str| sent.headers.iter().any(|h| h.eq_ignore_ascii_case(line));
    assert!(has("Connection: close"), "{:?}", sent.headers);
    assert!(has("Content-Type: application/json"), "{:?}", sent.headers);
    assert!(has("Content-Length: 15"), "{:?}", sent.headers);
    assert!(
        sent.headers
            .iter()
            .any(|h| h.to_ascii_lowercase().starts_with("host:"))
    );

    let server = FakeServer::start();
    server.route("/props", vec![Reply::fixture("props")]);
    let _ = get(&server);
    let sent = &server.requests()[0];
    assert_eq!(sent.target, "/props?model=m");
    assert!(
        !sent
            .headers
            .iter()
            .any(|h| h.to_ascii_lowercase().starts_with("content-length"))
    );
}

#[test]
fn a_timeout_before_the_first_byte_can_be_waited_out() {
    let server = FakeServer::start();
    server.route("/props", vec![Reply::fixture("props").head_delay(300)]);
    let mut conn = Connection::open(&server.socket).unwrap();
    conn.set_read_timeout(Duration::from_millis(50)).unwrap();
    conn.send(&GET_PROPS).unwrap();
    let mut timeouts = 0;
    let head = loop {
        match conn.read_head() {
            Ok(head) => break head,
            Err(HttpError::Timeout) => {
                assert!(!conn.received_any(), "nothing has arrived yet");
                timeouts += 1;
                assert!(timeouts < 100);
            }
            Err(e) => panic!("{e}"),
        }
    };
    assert!(
        timeouts >= 3,
        "only {timeouts} timeouts in 300 ms at 50 ms each"
    );
    assert!(conn.received_any());
    assert_eq!(head.status, 200);
}

#[test]
fn a_timeout_in_the_middle_of_the_head_keeps_what_arrived() {
    let server = FakeServer::start();
    server.route("/props", vec![Reply::fixture("props").trickle(10, 120)]);
    let mut conn = Connection::open(&server.socket).unwrap();
    conn.set_read_timeout(Duration::from_millis(40)).unwrap();
    conn.send(&GET_PROPS).unwrap();
    let mut saw_partial = false;
    let head = loop {
        match conn.read_head() {
            Ok(head) => break head,
            Err(HttpError::Timeout) => saw_partial |= conn.received_any(),
            Err(e) => panic!("{e}"),
        }
    };
    assert!(
        saw_partial,
        "at least one timeout must have happened after some bytes arrived"
    );
    assert_eq!(head.status, 200);
    assert_eq!(head.header("server"), Some("llama.cpp"));
}

#[test]
fn a_connection_closed_before_the_head_is_complete_is_closed() {
    for cut in [0, 5, 40] {
        let server = FakeServer::start();
        server.route("/props", vec![Reply::fixture("props").cut_after(cut)]);
        let (_, head) = get(&server);
        assert!(
            matches!(head, Err(HttpError::Closed)),
            "cut after {cut} bytes: {head:?}"
        );
    }
}

#[test]
fn a_body_cut_short_is_closed_in_both_framings() {
    for name in ["props", "unknown_model"] {
        let full = support::fixture_bytes("http", &format!("{name}.http")).len();
        let server = FakeServer::start();
        server.route("/props", vec![Reply::fixture(name).cut_after(full - 10)]);
        let got = get_body(&server);
        assert!(matches!(got, Err(HttpError::Closed)), "{name}: {got:?}");
    }
}

#[test]
fn malformed_heads_are_rejected() {
    for raw in [
        "HTTP/2 200 OK\r\n\r\n",
        "HELLO\r\n\r\n",
        "HTTP/1.1 abc OK\r\n\r\n",
        "HTTP/1.1 99 Low\r\n\r\n",
        "HTTP/1.1 200 OK\r\nno colon here\r\n\r\n",
        "HTTP/1.1 200 OK\r\n: no name\r\n\r\n",
    ] {
        let server = FakeServer::start();
        server.route("/props", vec![Reply::raw(raw)]);
        let (_, head) = get(&server);
        assert!(
            matches!(head, Err(HttpError::Malformed(_))),
            "{raw:?}: {head:?}"
        );
    }
}

#[test]
fn malformed_framing_is_an_error_not_a_panic() {
    for raw in [
        "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\nzz\r\nhello\r\n0\r\n\r\n",
        "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhelloXX0\r\n\r\n",
        "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\nffffffffffffffffffffffff\r\nx\r\n",
        "HTTP/1.1 200 OK\r\nContent-Length: many\r\n\r\nhello",
        "HTTP/1.1 200 OK\r\nContent-Length: -5\r\n\r\nhello",
    ] {
        let server = FakeServer::start();
        server.route("/props", vec![Reply::raw(raw)]);
        assert!(get_body(&server).is_err(), "{raw:?}");
    }
}

#[test]
fn everything_read_from_the_peer_is_bounded() {
    // A head that never ends.
    let server = FakeServer::start();
    let endless = format!("HTTP/1.1 200 OK\r\nX-Pad: {}\r\n", "a".repeat(2 * MAX_HEAD));
    server.route("/props", vec![Reply::raw(endless)]);
    let (_, head) = get(&server);
    assert!(matches!(head, Err(HttpError::TooLarge(_))), "{head:?}");

    // A chunk-size line that never ends.
    let server = FakeServer::start();
    let raw = format!(
        "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n{}",
        "1".repeat(10_000)
    );
    server.route("/props", vec![Reply::raw(raw)]);
    assert!(get_body(&server).is_err());

    // A body larger than the caller allows.
    let server = FakeServer::start();
    server.route("/props", vec![Reply::fixture("props")]);
    let (mut conn, head) = get(&server);
    let head = head.unwrap();
    let got = read_capped(&mut conn.body(&head).unwrap(), 1000);
    assert!(matches!(got, Err(HttpError::TooLarge(_))), "{got:?}");
}

#[test]
fn a_missing_socket_is_a_connect_error() {
    let missing = std::env::temp_dir().join("loopd-no-such-socket.sock");
    assert!(matches!(
        Connection::open(&missing),
        Err(HttpError::Connect(_))
    ));
    let e: Box<dyn std::error::Error> = Box::new(HttpError::Timeout);
    assert!(!e.to_string().is_empty());
}

/// A streamed body is read while it arrives. A `read` that already has data must not block
/// waiting for more, or every event is delivered one buffer late and the last ones only when
/// the stream ends.
#[test]
fn streamed_data_is_delivered_as_it_arrives() {
    let server = FakeServer::start();
    let reply = Reply::fixture("turn1");
    let piece = reply.offset_after_events(1);
    // About one event per piece, 300 ms apart: the whole stream takes over a second.
    server.route("/v1/chat/completions", vec![reply.trickle(piece, 300)]);
    let mut conn = Connection::open(&server.socket).unwrap();
    conn.set_read_timeout(Duration::from_secs(5)).unwrap();
    conn.send(&Request {
        method: "POST",
        path: "/v1/chat/completions",
        body: Some(b"{}"),
    })
    .unwrap();
    let head = conn.read_head().unwrap();
    let started = std::time::Instant::now();
    let mut body = conn.body(&head).unwrap();
    let mut arrivals = Vec::new();
    let mut buf = [0u8; 8192];
    loop {
        let n = std::io::Read::read(&mut body, &mut buf).unwrap();
        if n == 0 {
            break;
        }
        arrivals.push(started.elapsed());
    }
    let first = arrivals.first().copied().unwrap_or_default();
    let last = arrivals.last().copied().unwrap_or_default();
    assert!(
        first < Duration::from_millis(200),
        "the first data came only after {first:?}"
    );
    assert!(
        last - first >= Duration::from_millis(600),
        "everything arrived within {:?} of the first read: the reader is buffering the stream",
        last - first
    );
}