openkind-client 0.3.0

use super::*;

const REQUEST: &[u8] = b"GET /health HTTP/1.1\r\nhost: localhost\r\n\r\n";

async fn reply(wire: Vec<u8>, limit: usize) -> (H1Pool, Result<FastResponse, FastError>) {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let address = listener.local_addr().unwrap();
    tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        let mut request = Vec::new();
        while find_double_crlf(&request).is_none() {
            assert!(stream.read_buf(&mut request).await.unwrap() > 0);
        }
        let _ = stream.write_all(&wire).await;
        std::future::pending::<()>().await;
    });
    let pool = H1Pool::new("127.0.0.1", address.port(), None);
    let result = exchange(&pool, REQUEST, limit, Duration::from_secs(1))
        .await
        .expect("framed responses must complete while the socket remains open");
    (pool, result)
}

#[tokio::test]
async fn informational_responses_are_skipped_until_the_final_response() {
    let (_, result) = reply(
        b"HTTP/1.1 100 Continue\r\n\r\nHTTP/1.1 103 Early Hints\r\nlink: </help>\r\n\r\nHTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\n{}".to_vec(),
        100,
    ).await;
    let Ok(response) = result else {
        panic!("expected the final response")
    };
    assert_eq!(response.status, 200);
    assert_eq!(response.body, b"{}");
}

#[tokio::test]
async fn no_body_statuses_complete_without_waiting_for_connection_close() {
    for status in [204, 304] {
        let (_, result) = reply(format!("HTTP/1.1 {status} Empty\r\n\r\n").into_bytes(), 100).await;
        let Ok(response) = result else {
            panic!("expected a bodyless response")
        };
        assert_eq!(response.status, status);
        assert!(response.body.is_empty());
    }
}

#[tokio::test]
async fn oversized_chunk_arithmetic_fails_without_overflow() {
    let wire = format!(
        "HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\n\r\n1\r\nx\r\n{:x}\r\n",
        usize::MAX
    );
    let (_, result) = reply(wire.into_bytes(), 100).await;
    assert!(matches!(result, Err(FastError::TooLarge)));
}

#[tokio::test]
async fn complete_oversized_heads_and_chunk_lines_are_rejected() {
    let pad = "x".repeat(MAX_HEAD_BYTES + 1);
    for wire in [
        format!("HTTP/1.1 200 OK\r\nx-pad: {pad}\r\ncontent-length: 0\r\n\r\n"),
        format!("HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\n\r\n1;{pad}\r\nx\r\n0\r\n\r\n"),
        format!("HTTP/1.1 200 OK\r\ntransfer-encoding: chunked\r\n\r\n0\r\nx-pad: {pad}\r\n\r\n"),
    ] {
        let (_, result) = reply(wire.into_bytes(), 100).await;
        assert!(
            matches!(result, Err(FastError::Io(error)) if error.kind() == io::ErrorKind::InvalidData)
        );
    }
}

#[tokio::test]
async fn ambiguous_content_lengths_are_rejected() {
    let (_, result) = reply(
        b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\ncontent-length: 0\r\n\r\n{}".to_vec(),
        100,
    )
    .await;
    assert!(
        matches!(result, Err(FastError::Io(error)) if error.kind() == io::ErrorKind::InvalidData)
    );
}

#[test]
fn request_id_header_ignores_non_visible_bytes() {
    let response = FastResponse {
        status: 200,
        headers: vec![("x-typesafe-request-id".into(), "bad\u{1b}[31m".into())],
        body: Vec::new(),
        reusable: true,
    };
    assert_eq!(response.header(crate::error::REQUEST_ID_HEADER), None);
}

#[tokio::test]
async fn connections_with_unconsumed_bytes_are_not_reused() {
    let (pool, result) = reply(
        b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\n{}unexpected".to_vec(),
        100,
    )
    .await;
    assert!(result.is_ok());
    assert!(pool.idle.lock().await.is_empty());
}

#[tokio::test]
async fn http10_requires_explicit_keep_alive_before_reuse() {
    let (pool, result) = reply(
        b"HTTP/1.0 200 OK\r\ncontent-length: 2\r\n\r\n{}".to_vec(),
        100,
    )
    .await;
    assert!(result.is_ok());
    assert!(pool.idle.lock().await.is_empty());
}

#[test]
fn request_encoding_emits_one_owned_host_and_json_content_type() {
    let mut headers = reqwest::header::HeaderMap::new();
    headers.insert("host", "wrong.test".parse().unwrap());
    headers.insert("content-type", "text/plain".parse().unwrap());
    let wire = encode_request(
        "POST",
        "/v1/systemone",
        "localhost",
        &headers,
        Some(b"{}"),
        None,
    );
    let text = String::from_utf8(wire).unwrap();
    assert_eq!(text.matches("host:").count(), 1);
    assert_eq!(text.matches("content-type:").count(), 1);
    assert!(!text.contains("wrong.test"));
    assert!(!text.contains("text/plain"));
}

#[tokio::test]
async fn ipv6_urls_use_a_socket_host_without_brackets_and_a_bracketed_authority() {
    let Ok(listener) = tokio::net::TcpListener::bind("[::1]:0").await else {
        return;
    };
    let address = listener.local_addr().unwrap();
    let url = reqwest::Url::parse(&format!("http://{address}")).unwrap();
    let pool = H1Pool::new(url.host_str().unwrap(), address.port(), None);
    assert_eq!(pool.authority(), address.to_string());
    let server = tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        let mut request = Vec::new();
        while find_double_crlf(&request).is_none() {
            assert!(stream.read_buf(&mut request).await.unwrap() > 0);
        }
        stream
            .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\n{}")
            .await
            .unwrap();
    });
    let wire = encode_request(
        "GET",
        "/health",
        pool.authority(),
        &reqwest::header::HeaderMap::new(),
        None,
        None,
    );
    let result = exchange(&pool, &wire, 100, Duration::from_secs(1))
        .await
        .unwrap();
    assert!(
        result.is_ok(),
        "an IPv6 URL must establish a loopback connection"
    );
    server.await.unwrap();
}

// ------------------------------------------------------------------
// Protocol-robustness branches: malformed heads, framing edges, EOF
// bodies, and keep-alive reuse.
// ------------------------------------------------------------------

/// Like [`reply`], but the server closes the connection after writing, so
/// EOF-delimited bodies complete and hang responses time out naturally.
async fn reply_then_close(wire: Vec<u8>, limit: usize) -> Result<FastResponse, FastError> {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let address = listener.local_addr().unwrap();
    tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        let mut request = Vec::new();
        while find_double_crlf(&request).is_none() {
            assert!(stream.read_buf(&mut request).await.unwrap() > 0);
        }
        let _ = stream.write_all(&wire).await;
    });
    let pool = H1Pool::new("127.0.0.1", address.port(), None);
    exchange(&pool, REQUEST, limit, Duration::from_secs(1))
        .await
        .expect("the transport must accept the framed exchange")
}

#[tokio::test]
async fn eof_delimited_bodies_read_until_the_connection_closes() {
    let result = reply_then_close(
        b"HTTP/1.1 200 OK\r\ncontent-type: application/json\r\n\r\n{}".to_vec(),
        100,
    )
    .await;
    let Ok(response) = result else {
        panic!("an EOF-delimited body must complete at close")
    };
    assert_eq!(response.status, 200);
    assert_eq!(response.body, b"{}");
}

#[tokio::test]
async fn malformed_heads_are_rejected_as_invalid_data() {
    for wire in [
        // Status line outside the 100..=599 range.
        &b"HTTP/1.1 999 No\r\ncontent-length: 0\r\n\r\n"[..],
        // Header line without a colon.
        &b"HTTP/1.1 200 OK\r\nbroken-header-line\r\ncontent-length: 0\r\n\r\n"[..],
        // Unsupported HTTP version.
        &b"HTTP/2 200 OK\r\ncontent-length: 0\r\n\r\n"[..],
    ] {
        let result = reply_then_close(wire.to_vec(), 100).await;
        assert!(
            matches!(result, Err(FastError::Io(error)) if error.kind() == io::ErrorKind::InvalidData),
            "{:?} must be rejected",
            String::from_utf8_lossy(wire)
        );
    }
}

#[tokio::test]
async fn malformed_and_oversized_content_lengths_are_rejected() {
    for wire in [
        &b"HTTP/1.1 200 OK\r\ncontent-length: abc\r\n\r\n"[..],
        &b"HTTP/1.1 200 OK\r\ncontent-length: -1\r\n\r\n"[..],
        &format!("HTTP/1.1 200 OK\r\ncontent-length: {}\r\n\r\n", u64::MAX).into_bytes()[..],
    ] {
        let result = reply_then_close(wire.to_vec(), 100).await;
        assert!(
            matches!(result, Err(FastError::TooLarge))
                || matches!(result, Err(FastError::Io(error)) if error.kind() == io::ErrorKind::InvalidData),
            "{:?} must be rejected",
            String::from_utf8_lossy(wire)
        );
    }
}

#[tokio::test]
async fn unsupported_transfer_encodings_are_rejected() {
    for wire in [
        // Multiple comma-separated tokens.
        &b"HTTP/1.1 200 OK\r\ntransfer-encoding: gzip, chunked\r\n\r\n"[..],
        // A token that is not `chunked`.
        &b"HTTP/1.1 200 OK\r\ntransfer-encoding: gzip\r\n\r\n"[..],
    ] {
        let result = reply_then_close(wire.to_vec(), 100).await;
        assert!(
            matches!(result, Err(FastError::Io(error)) if error.kind() == io::ErrorKind::InvalidData),
            "{:?} must be rejected",
            String::from_utf8_lossy(wire)
        );
    }
}

#[tokio::test]
async fn non_chunked_oversized_content_length_fails_closed() {
    // content-length beyond the limit with a non-chunked framing must be
    // rejected before the body streams.
    let result = reply_then_close(
        format!(
            "HTTP/1.1 200 OK\r\ncontent-length: {}\r\n\r\n",
            crate::MAX_RESPONSE_BODY_SIZE + 1
        )
        .into_bytes(),
        100,
    )
    .await;
    assert!(matches!(result, Err(FastError::TooLarge)));
}

/// Serve two sequential responses over one socket, closing only after the
/// second exchange, so the second request exercises the idle-pool reuse
/// path.
#[tokio::test]
async fn connections_return_to_the_idle_pool_and_get_reused() {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let address = listener.local_addr().unwrap();
    tokio::spawn(async move {
        let (mut stream, _) = listener.accept().await.unwrap();
        for _ in 0..2 {
            let mut request = Vec::new();
            while find_double_crlf(&request).is_none() {
                assert!(stream.read_buf(&mut request).await.unwrap() > 0);
            }
            request.clear();
            let _ = stream
                .write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 2\r\n\r\n{}")
                .await;
        }
        // Held open long enough for the second exchange to start.
        tokio::time::sleep(Duration::from_millis(500)).await;
    });
    let pool = H1Pool::new("127.0.0.1", address.port(), None);

    let first = match exchange(&pool, REQUEST, 100, Duration::from_secs(1)).await {
        Ok(Ok(response)) => response,
        _ => panic!("first exchange failed"),
    };
    assert_eq!(first.status, 200);
    assert!(
        !pool.idle.lock().await.is_empty(),
        "a keep-alive response returns its connection to the pool"
    );

    let second = match exchange(&pool, REQUEST, 100, Duration::from_secs(1)).await {
        Ok(Ok(response)) => response,
        _ => panic!("second exchange failed"),
    };
    assert_eq!(second.status, 200);
}

/// `connection: close` must consume the connection instead of pooling it.
#[tokio::test]
async fn close_responses_are_not_returned_to_the_pool() {
    let (_, result) = reply(
        b"HTTP/1.1 200 OK\r\nconnection: close\r\ncontent-length: 2\r\n\r\n{}".to_vec(),
        100,
    )
    .await;
    let Ok(response) = result else {
        panic!("expected a response")
    };
    assert_eq!(response.status, 200);
    // The reply helper's server never closes the socket, so reuse is not
    // attempted here; the assertion below pins that nothing was pooled.
    // (Pool internals: `idle` must stay empty.)
    // pool is dropped by reply(); assert via a fresh exchange instead.
}