base-http1 0.3.0

// Released under the MIT License.
// Copyright, 2026, by Samuel Williams.

use crate::{
    observer,
    support::{Input, Segments, failing_input, input, is_invalid, is_io},
};
use base_http::{
    Fields, InterimResponses, Method, Protocol, Request, RequestMetadata, ResponseMetadata,
    Version,
    body::{BodyFrame, BodyReader, BufferedBody, ReadableBody},
};
use base_http1::{Error, Options, RequestWriter, ResponseReader};
use base_scheduler::{
    Cancellation, IoBufSequence, OperationResult, SocketSendFlags, SocketStreamWriter,
};
use futures_lite::future::block_on;
use std::{
    io::{self, IoSlice},
    sync::{Arc, Mutex},
};

fn unsupported<T>() -> OperationResult<T> {
    Err(io::Error::from(io::ErrorKind::Unsupported).into())
}

fn broken<T>() -> OperationResult<T> {
    Err(io::Error::from(io::ErrorKind::BrokenPipe).into())
}

/// Bytes accepted by the transport, and the number of upcoming writes and flushes that fail.
#[derive(Default)]
struct Wire {
    bytes: Vec<u8>,
    write_failures: usize,
    flush_failures: usize,
}

#[derive(Clone, Default)]
struct Output(Arc<Mutex<Wire>>);

impl Output {
    fn bytes(&self) -> String {
        String::from_utf8_lossy(&self.0.lock().unwrap().bytes).into_owned()
    }
    fn fail_writes(&self, count: usize) {
        self.0.lock().unwrap().write_failures = count;
    }
    fn fail_flushes(&self, count: usize) {
        self.0.lock().unwrap().flush_failures = count;
    }
}

impl SocketStreamWriter for Output {
    type Buffer = Vec<u8>;
    async fn write(&mut self, bytes: &[u8], cancellation: &Cancellation) -> OperationResult<usize> {
        cancellation.check()?;
        let mut wire = self.0.lock().unwrap();
        if wire.write_failures > 0 {
            wire.write_failures -= 1;
            return broken();
        }
        wire.bytes.extend_from_slice(bytes);
        Ok(bytes.len())
    }
    async fn write_vectored(
        &mut self,
        _: &[IoSlice<'_>],
        _: &Cancellation,
    ) -> OperationResult<usize> {
        unsupported()
    }
    async fn submit(
        &mut self,
        buffer: Vec<u8>,
        _: &Cancellation,
    ) -> Result<(), (base_scheduler::OperationError, Vec<u8>)> {
        Err((io::Error::from(io::ErrorKind::Unsupported).into(), buffer))
    }
    async fn flush(&mut self, cancellation: &Cancellation) -> OperationResult<()> {
        cancellation.check()?;
        let mut wire = self.0.lock().unwrap();
        if wire.flush_failures > 0 {
            wire.flush_failures -= 1;
            return broken();
        }
        Ok(())
    }
    async fn write_vectored_with_flags(
        &mut self,
        _: &[IoSlice<'_>],
        _: SocketSendFlags,
        _: &Cancellation,
    ) -> OperationResult<usize> {
        unsupported()
    }
    async fn write_owned_vectored<B: IoBufSequence>(
        &mut self,
        buffers: B,
        _: SocketSendFlags,
        _: &Cancellation,
    ) -> (OperationResult<usize>, B) {
        (unsupported(), buffers)
    }
    async fn close(&mut self) -> io::Result<()> {
        Ok(())
    }
}

fn request_writer() -> (RequestWriter<Output>, Output) {
    let output = Output::default();
    (RequestWriter::new(output.clone()), output)
}

fn request(method: Method, body: &[u8]) -> Request<BufferedBody> {
    let mut request = Request::new(method, "/", BufferedBody::from(body.to_vec()));
    request.authority = Some("example".into());
    request
}

fn fields(pairs: &[(&str, &[u8])]) -> Fields {
    let mut fields = Fields::new();
    for (name, value) in pairs {
        fields.add(name, value).unwrap();
    }
    fields
}

fn large_head() -> Request<BufferedBody> {
    let mut request = request(Method::GET, b"");
    request
        .fields
        .add("X-Large", "a".repeat(32 * 1024))
        .unwrap();
    request
}

fn not_active(result: Result<(), Error>) -> bool {
    matches!(result, Err(Error::Invalid("request output is not active")))
}

fn data(bytes: &[u8]) -> BodyFrame<Vec<u8>> {
    BodyFrame::Data {
        data: bytes.to_vec(),
        end: false,
    }
}

#[test]
fn invalid_request_heads_are_rejected_before_any_output() {
    let protocol = || Some(Protocol::new("websocket").unwrap());
    let cases: Vec<(Request<BufferedBody>, &str)> = vec![
        (
            {
                let mut request = request(Method::GET, b"");
                request.version = Some(Version::Http2);
                request
            },
            "unsupported request version",
        ),
        (
            {
                let mut request = request(Method::GET, b"");
                request.fields = fields(&[("Host", b"example"), ("Host", b"example")]);
                request
            },
            "duplicate singleton field",
        ),
        (
            {
                let mut request = request(Method::GET, b"");
                request.authority = None;
                request.fields = fields(&[("Host", b"\xff")]);
                request
            },
            "invalid host",
        ),
        (
            {
                let mut request = request(Method::GET, b"");
                request.authority = None;
                request
            },
            "missing host",
        ),
        (
            {
                let mut request = request(Method::GET, b"");
                request.version = Some(Version::Http10);
                request.protocol = protocol();
                request
            },
            "upgrade requires HTTP/1.1",
        ),
        (
            {
                let mut request = request(Method::GET, b"");
                request.protocol = protocol();
                request.fields = fields(&[("Upgrade", b"websocket"), ("Upgrade", b"websocket")]);
                request
            },
            "duplicate singleton field",
        ),
        (
            {
                let mut request = request(Method::GET, b"");
                request.protocol = protocol();
                request.fields = fields(&[("Upgrade", b"h2c")]);
                request
            },
            "conflicting upgrade protocol",
        ),
        (
            {
                let mut request = request(Method::GET, b"");
                request.fields = fields(&[("Connection", b"upgrade")]);
                request
            },
            "missing upgrade protocol",
        ),
        (
            {
                let mut request = request(Method::CONNECT, b"");
                request.authority = Some("example:443".into());
                request
            },
            "CONNECT cannot have a path",
        ),
        (
            {
                let mut request = request(Method::CONNECT, b"");
                request.path = None;
                request.authority = None;
                request.version = Some(Version::Http10);
                request
            },
            "CONNECT requires authority",
        ),
        (
            {
                let mut request = request(Method::POST, b"abc");
                request.fields = fields(&[("Transfer-Encoding", b"gzip")]);
                request
            },
            "unsupported transfer coding",
        ),
        (
            {
                let mut request = request(Method::POST, b"abc");
                request.content_length = Some(3);
                request.fields = fields(&[("Transfer-Encoding", b"chunked")]);
                request
            },
            "content-length with transfer-encoding",
        ),
        (
            {
                let mut request = request(Method::POST, b"abc");
                request.version = Some(Version::Http10);
                request.fields = fields(&[("Transfer-Encoding", b"chunked")]);
                request
            },
            "HTTP/1.0 transfer-encoding",
        ),
    ];
    block_on(async {
        for (mut request, reason) in cases {
            let (mut writer, output) = request_writer();
            let result = writer
                .write_request(&mut request, &Cancellation::never())
                .await;
            assert!(
                matches!(result, Err(Error::Invalid(actual)) if actual == reason),
                "{reason}: {result:?}"
            );
            assert!(output.bytes().is_empty());
            assert!(writer.is_reusable());
        }
        let (mut writer, output) = request_writer();
        let mut request = request(Method::POST, b"abc");
        request.fields = fields(&[("Content-Length", b"5")]);
        assert!(matches!(
            writer
                .write_request(&mut request, &Cancellation::never())
                .await,
            Err(Error::LengthMismatch)
        ));
        assert!(output.bytes().is_empty());
        let mut metadata = RequestMetadata::new(Method::POST);
        metadata.path = Some("/".into());
        metadata.version = Some(Version::Http10);
        assert!(matches!(
            writer
                .begin_request(&metadata, &Cancellation::never())
                .await,
            Err(Error::Invalid("HTTP/1.0 request requires a body length"))
        ));
        assert!(output.bytes().is_empty());
        assert!(writer.is_reusable());
    });
}

// A body of known length describes the data a request sends, so a typed declaration it contradicts, such as one kept after replacing the body, is ignored.
#[test]
fn the_body_length_overrides_a_contradicted_typed_declaration() {
    block_on(async {
        let (mut writer, output) = request_writer();
        let mut post = request(Method::POST, b"abc");
        post.content_length = Some(5);
        writer
            .write_request(&mut post, &Cancellation::never())
            .await
            .unwrap();
        assert_eq!(
            output.bytes(),
            "POST / HTTP/1.1\r\nHost: example\r\nContent-Length: 3\r\n\r\nabc"
        );

        // A raw field the body agrees with is sent as written.
        let (mut writer, output) = request_writer();
        let mut post = request(Method::POST, b"abc");
        post.content_length = Some(5);
        post.fields = fields(&[("Content-Length", b"3")]);
        writer
            .write_request(&mut post, &Cancellation::never())
            .await
            .unwrap();
        assert_eq!(
            output.bytes(),
            "POST / HTTP/1.1\r\nContent-Length: 3\r\nHost: example\r\n\r\nabc"
        );

        // A raw Transfer-Encoding refuses any declared length, whether or not the body agrees with it.
        for declared in [3, 5] {
            let (mut writer, output) = request_writer();
            let mut post = request(Method::POST, b"abc");
            post.content_length = Some(declared);
            post.fields = fields(&[("Transfer-Encoding", b"chunked")]);
            assert!(matches!(
                writer
                    .write_request(&mut post, &Cancellation::never())
                    .await,
                Err(Error::Invalid("content-length with transfer-encoding"))
            ));
            assert!(output.bytes().is_empty());
        }

        // Without one, a raw Transfer-Encoding chunks a body of known length and keeps its trailers.
        let (body, mut producer) = BodyReader::with_length(64, 4, 2);
        producer
            .write(b"hi".to_vec(), &Cancellation::never())
            .await
            .unwrap();
        producer
            .write_frame(trailer(), &Cancellation::never())
            .await
            .unwrap();
        let mut post = Request::new(Method::POST, "/", body);
        post.authority = Some("example".into());
        post.fields = fields(&[("Transfer-Encoding", b"chunked")]);
        let (mut writer, output) = request_writer();
        writer
            .write_request(&mut post, &Cancellation::never())
            .await
            .unwrap();
        assert_eq!(
            output.bytes(),
            "POST / HTTP/1.1\r\nTransfer-Encoding: chunked\r\nHost: example\r\n\r\n2\r\nhi\r\n0\r\nx-digest: 1\r\n\r\n"
        );
    });
}

fn trailer() -> BodyFrame<Vec<u8>> {
    BodyFrame::Headers {
        fields: fields(&[("x-digest", b"1")]),
        end: true,
    }
}

// HTTP/1.0 has no chunked framing, so a request body of known length is sent with it, and any trailers it produces are dropped.
#[test]
fn http_10_requests_send_known_length_bodies_and_drop_their_trailers() {
    block_on(async {
        let never = Cancellation::never();
        let (body, mut producer) = BodyReader::with_length(64, 4, 2);
        producer.write(b"hi".to_vec(), &never).await.unwrap();
        producer.write_frame(trailer(), &never).await.unwrap();
        assert!(body.may_have_trailers());
        let mut post = Request::new(Method::POST, "/", body);
        post.authority = Some("example".into());
        post.version = Some(Version::Http10);
        let (mut writer, output) = request_writer();
        assert!(!writer.write_request(&mut post, &never).await.unwrap());
        assert_eq!(
            output.bytes(),
            "POST / HTTP/1.0\r\nHost: example\r\nContent-Length: 2\r\n\r\nhi"
        );

        // A Trailer field is only a hint, so it neither selects framing nor is refused.
        let mut announced = request(Method::POST, b"hi");
        announced.version = Some(Version::Http10);
        announced.fields = fields(&[("Trailer", b"x-digest")]);
        let (mut writer, output) = request_writer();
        writer.write_request(&mut announced, &never).await.unwrap();
        assert_eq!(
            output.bytes(),
            "POST / HTTP/1.0\r\nTrailer: x-digest\r\nHost: example\r\nContent-Length: 2\r\n\r\nhi"
        );

        // Nor on HTTP/1.1, where a body without trailers keeps fixed-length framing.
        let mut announced = request(Method::POST, b"hi");
        announced.fields = fields(&[("Trailer", b"x-digest")]);
        let (mut writer, output) = request_writer();
        writer.write_request(&mut announced, &never).await.unwrap();
        assert_eq!(
            output.bytes(),
            "POST / HTTP/1.1\r\nTrailer: x-digest\r\nHost: example\r\nContent-Length: 2\r\n\r\nhi"
        );

        // Imperative output drops them too.
        let mut metadata = RequestMetadata::new(Method::POST);
        metadata.path = Some("/".into());
        metadata.version = Some(Version::Http10);
        metadata.content_length = Some(2);
        let (mut writer, output) = request_writer();
        writer.begin_request(&metadata, &never).await.unwrap();
        writer.write_frame(data(b"hi"), &never).await.unwrap();
        writer.write_frame(trailer(), &never).await.unwrap();
        assert_eq!(
            output.bytes(),
            "POST / HTTP/1.0\r\nContent-Length: 2\r\n\r\nhi"
        );
        assert!(!writer.is_reusable());
    });
}

// Request trailers are rare and a client cannot assume the server accepts chunked uploads, so an HTTP/1.1 request of known length is sent with Content-Length and drops its trailers, while one of unknown length is chunked and keeps them.
#[test]
fn http_11_requests_of_known_length_drop_their_trailers() {
    block_on(async {
        let never = Cancellation::never();
        let (body, mut producer) = BodyReader::with_length(64, 4, 2);
        producer.write(b"hi".to_vec(), &never).await.unwrap();
        producer.write_frame(trailer(), &never).await.unwrap();
        let mut post = Request::new(Method::POST, "/", body);
        post.authority = Some("example".into());
        let (mut writer, output) = request_writer();
        assert!(writer.write_request(&mut post, &never).await.unwrap());
        assert_eq!(
            output.bytes(),
            "POST / HTTP/1.1\r\nHost: example\r\nContent-Length: 2\r\n\r\nhi"
        );

        // Imperative output with a declared length, announcing the trailer it then writes.
        let mut metadata = RequestMetadata::new(Method::POST);
        metadata.path = Some("/".into());
        metadata.authority = Some("example".into());
        metadata.content_length = Some(2);
        metadata.fields = fields(&[("Trailer", b"x-digest")]);
        let (mut writer, output) = request_writer();
        writer.begin_request(&metadata, &never).await.unwrap();
        writer.write_frame(data(b"hi"), &never).await.unwrap();
        writer.write_frame(trailer(), &never).await.unwrap();
        assert_eq!(
            output.bytes(),
            "POST / HTTP/1.1\r\nTrailer: x-digest\r\nHost: example\r\nContent-Length: 2\r\n\r\nhi"
        );
        // The lane completed normally and accepts the next request.
        assert!(writer.is_reusable());
        let mut get = request(Method::GET, b"");
        writer.write_request(&mut get, &never).await.unwrap();
        assert!(
            output
                .bytes()
                .ends_with("hiGET / HTTP/1.1\r\nHost: example\r\n\r\n")
        );

        // Dropped trailers are still validated.
        writer.begin_request(&metadata, &never).await.unwrap();
        writer.write_frame(data(b"hi"), &never).await.unwrap();
        let forbidden = BodyFrame::<Vec<u8>>::Headers {
            fields: fields(&[("content-length", b"2")]),
            end: true,
        };
        assert!(writer.write_frame(forbidden, &never).await.is_err());
        assert!(!writer.is_reusable());

        // Without a length the request is chunked and keeps its trailers.
        metadata.content_length = None;
        let (mut writer, output) = request_writer();
        writer.begin_request(&metadata, &never).await.unwrap();
        writer.write_frame(data(b"hi"), &never).await.unwrap();
        writer.write_frame(trailer(), &never).await.unwrap();
        assert!(
            output
                .bytes()
                .ends_with("Transfer-Encoding: chunked\r\n\r\n2\r\nhi\r\n0\r\nx-digest: 1\r\n\r\n")
        );
    });
}

// An HTTP/1.0 request must declare its length, so a body of unknown length is refused before any output, whether or not it may have trailers.
#[test]
fn http_10_requests_refuse_bodies_of_unknown_length() {
    block_on(async {
        let never = Cancellation::never();
        let (body, mut producer) = BodyReader::bounded(64, 4);
        producer.write(b"hi".to_vec(), &never).await.unwrap();
        let mut post = Request::new(Method::POST, "/", body);
        post.authority = Some("example".into());
        post.version = Some(Version::Http10);
        let (mut writer, output) = request_writer();
        assert!(matches!(
            writer.write_request(&mut post, &never).await,
            Err(Error::Invalid("HTTP/1.0 request requires a body length"))
        ));
        assert!(output.bytes().is_empty());
        assert!(writer.is_reusable());
        // Writing the request closed its body.
        assert!(producer.write(b"!".to_vec(), &never).await.is_err());
    });
}

#[test]
fn generated_heads_complete_host_upgrade_connection_and_length_fields() {
    let protocol = || Some(Protocol::new("websocket").unwrap());
    let cases: Vec<(Request<BufferedBody>, bool, &str)> = vec![
        (
            {
                let mut request = request(Method::GET, b"");
                request.fields = fields(&[("Host", b"EXAMPLE")]);
                request
            },
            true,
            "GET / HTTP/1.1\r\nHost: EXAMPLE\r\n\r\n",
        ),
        (
            {
                let mut request = request(Method::GET, b"");
                request.protocol = protocol();
                request
            },
            true,
            "GET / HTTP/1.1\r\nHost: example\r\nUpgrade: websocket\r\nConnection: upgrade\r\n\r\n",
        ),
        (
            {
                let mut request = request(Method::GET, b"");
                request.protocol = protocol();
                request.fields = fields(&[("Connection", b"Upgrade"), ("Upgrade", b"WebSocket")]);
                request
            },
            true,
            "GET / HTTP/1.1\r\nConnection: Upgrade\r\nUpgrade: WebSocket\r\nHost: example\r\n\r\n",
        ),
        (
            {
                let mut request = request(Method::GET, b"");
                request.version = Some(Version::Http10);
                request.persistent = Some(true);
                request
            },
            true,
            "GET / HTTP/1.0\r\nHost: example\r\nConnection: keep-alive\r\n\r\n",
        ),
        (
            {
                let mut request = request(Method::GET, b"");
                request.version = Some(Version::Http10);
                request.fields = fields(&[("Connection", b"keep-alive")]);
                request
            },
            true,
            "GET / HTTP/1.0\r\nConnection: keep-alive\r\nHost: example\r\n\r\n",
        ),
        (
            {
                let mut request = request(Method::GET, b"");
                request.version = Some(Version::Http10);
                request
            },
            false,
            "GET / HTTP/1.0\r\nHost: example\r\n\r\n",
        ),
        (
            request(Method::POST, b""),
            true,
            "POST / HTTP/1.1\r\nHost: example\r\nContent-Length: 0\r\n\r\n",
        ),
    ];
    block_on(async {
        for (mut request, persistent, expected) in cases {
            let (mut writer, output) = request_writer();
            assert_eq!(
                writer
                    .write_request(&mut request, &Cancellation::never())
                    .await
                    .unwrap(),
                persistent,
                "{expected}"
            );
            assert_eq!(writer.is_reusable(), persistent);
            assert_eq!(output.bytes(), expected);
        }
    });
}

#[test]
fn request_output_failures_and_cancellation_retire_the_lane() {
    block_on(async {
        let cancelled = Cancellation::new();
        cancelled.cancel();
        let (mut writer, output) = request_writer();
        assert!(matches!(
            writer
                .write_request(&mut request(Method::GET, b""), &cancelled)
                .await,
            Err(Error::Cancelled)
        ));
        assert!(writer.is_reusable());
        let mut close = request(Method::GET, b"");
        close.persistent = Some(false);
        assert!(
            !writer
                .write_request(&mut close, &Cancellation::never())
                .await
                .unwrap()
        );
        assert!(matches!(
            writer
                .write_request(&mut request(Method::GET, b""), &Cancellation::never())
                .await,
            Err(Error::Invalid("output is not reusable"))
        ));
        assert_eq!(
            output.bytes(),
            "GET / HTTP/1.1\r\nHost: example\r\nConnection: close\r\n\r\n"
        );

        // A large head is written directly rather than batched; a small one is sent by the flush.
        let (mut writer, output) = request_writer();
        output.fail_writes(1);
        assert!(matches!(
            writer
                .write_request(&mut large_head(), &Cancellation::never())
                .await,
            Err(Error::Io(_))
        ));
        assert!(!writer.is_reusable());
        let (mut writer, output) = request_writer();
        output.fail_flushes(1);
        assert!(matches!(
            writer
                .write_request(&mut request(Method::GET, b""), &Cancellation::never())
                .await,
            Err(Error::Io(_))
        ));
        assert!(!writer.is_reusable());
    });
}

#[test]
fn failed_request_heads_never_accept_body_frames() {
    block_on(async {
        let metadata = request(Method::POST, b"").metadata;
        let cancelled = Cancellation::new();
        cancelled.cancel();
        let (mut writer, output) = request_writer();
        assert!(matches!(
            writer.begin_request(&metadata, &cancelled).await,
            Err(Error::Cancelled)
        ));
        assert!(writer.is_reusable());
        assert!(not_active(
            writer
                .write_frame(data(b"abc"), &Cancellation::never())
                .await
        ));
        assert!(output.bytes().is_empty());

        // Each transport failure is transient, so frames accepted after the failed head would reach the wire.
        let (mut writer, output) = request_writer();
        output.fail_writes(1);
        assert!(matches!(
            writer
                .begin_request(&large_head().metadata, &Cancellation::never())
                .await,
            Err(Error::Io(_))
        ));
        assert!(not_active(
            writer
                .write_frame(data(b"abc"), &Cancellation::never())
                .await
        ));
        assert!(output.bytes().is_empty());
        let (mut writer, output) = request_writer();
        output.fail_flushes(1);
        assert!(matches!(
            writer
                .begin_request(&metadata, &Cancellation::never())
                .await,
            Err(Error::Io(_))
        ));
        assert!(!writer.is_reusable());
        assert!(not_active(
            writer
                .write_frame(data(b"abc"), &Cancellation::never())
                .await
        ));
        assert!(not_active(
            writer
                .write_frame(BodyFrame::<Vec<u8>>::End, &Cancellation::never())
                .await
        ));
        assert_eq!(
            output.bytes(),
            "POST / HTTP/1.1\r\nHost: example\r\nTransfer-Encoding: chunked\r\n\r\n"
        );
    });
}

#[test]
fn imperative_request_frames_require_an_active_flushed_body() {
    block_on(async {
        let metadata = request(Method::POST, b"").metadata;
        let (mut writer, output) = request_writer();
        assert!(not_active(
            writer
                .write_frame(data(b"abc"), &Cancellation::never())
                .await
        ));
        writer
            .begin_request(&metadata, &Cancellation::never())
            .await
            .unwrap();
        let cancelled = Cancellation::new();
        cancelled.cancel();
        assert!(matches!(
            writer.write_frame(data(b"abc"), &cancelled).await,
            Err(Error::Cancelled)
        ));
        writer
            .write_frame(data(b"abc"), &Cancellation::never())
            .await
            .unwrap();
        writer.flush(&Cancellation::never()).await.unwrap();
        assert!(!writer.is_reusable());
        output.fail_flushes(1);
        assert!(matches!(
            writer
                .write_frame(data(b"de"), &Cancellation::never())
                .await,
            Err(Error::Io(_))
        ));
        assert!(not_active(
            writer.write_frame(data(b"f"), &Cancellation::never()).await
        ));
        assert!(!writer.is_reusable());
        assert_eq!(
            output.bytes(),
            "POST / HTTP/1.1\r\nHost: example\r\nTransfer-Encoding: chunked\r\n\r\n3\r\nabc\r\n2\r\nde\r\n"
        );

        // A failed explicit flush ends the active body and the lane.
        let (mut writer, output) = request_writer();
        writer
            .begin_request(&metadata, &Cancellation::never())
            .await
            .unwrap();
        output.fail_flushes(1);
        assert!(matches!(
            writer.flush(&Cancellation::never()).await,
            Err(Error::Io(_))
        ));
        assert!(not_active(
            writer
                .write_frame(BodyFrame::<Vec<u8>>::End, &Cancellation::never())
                .await
        ));
        assert!(!writer.is_reusable());
    });
}

fn response_reader(bytes: &'static [u8]) -> ResponseReader<Input> {
    ResponseReader::new(input(bytes, usize::MAX), Options::default()).unwrap()
}

async fn read_error(bytes: &'static [u8], request: &RequestMetadata) -> Error {
    let mut reader = response_reader(bytes);
    let error = match reader
        .read_response(request, &mut observer(), &Cancellation::never())
        .await
    {
        Ok(response) => panic!("accepted {bytes:?}: {:?}", response.metadata),
        Err(error) => error,
    };
    assert!(!reader.is_reusable());
    error
}

#[test]
fn response_readers_validate_their_limits() {
    let options = Options {
        max_fields: 0,
        ..Options::default()
    };
    assert!(matches!(
        ResponseReader::new(input(b"", usize::MAX), options),
        Err(Error::Invalid("invalid HTTP/1 limits"))
    ));
}

#[test]
fn invalid_response_connection_options_and_interim_failures_are_errors() {
    block_on(async {
        let get = RequestMetadata::new(Method::GET);
        for bytes in [
            b"HTTP/1.1 103 Early Hints\r\nConnection: a@b\r\n\r\n".as_slice(),
            b"HTTP/1.1 200 OK\r\nConnection: a@b\r\nContent-Length: 0\r\n\r\n",
        ] {
            assert!(matches!(
                read_error(bytes, &get).await,
                Error::Invalid("invalid connection option")
            ));
        }
        let mut reader = response_reader(b"HTTP/1.1 100 Continue\r\n\r\nHTTP/1.1 200 OK\r\n\r\n");
        let mut interim = InterimResponses::new(|_: ResponseMetadata| {
            std::future::ready(Err::<(), _>(io::Error::other("interim handler failed")))
        });
        let error = reader
            .read_response(&get, &mut interim, &Cancellation::never())
            .await
            .map(|_| ())
            .unwrap_err();
        let Error::Body(source) = error else {
            panic!("unexpected {error:?}");
        };
        assert_eq!(source.to_string(), "interim handler failed");
        assert!(!reader.is_reusable());
    });
}

#[test]
fn switching_protocols_responses_must_select_an_offered_token() {
    block_on(async {
        let mut offer = RequestMetadata::new(Method::GET);
        offer.protocol = Some(Protocol::new("websocket").unwrap());
        for (bytes, reason) in [
            (
                b"HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\n\r\n".as_slice(),
                "invalid upgrade response",
            ),
            (
                b"HTTP/1.1 101 Switching Protocols\r\nConnection: upgrade\r\nUpgrade: websocket\r\nContent-Length: 0\r\n\r\n",
                "invalid upgrade response",
            ),
            (
                b"HTTP/1.1 101 Switching Protocols\r\nConnection: upgrade\r\nUpgrade: websocket\r\nUpgrade: websocket\r\n\r\n",
                "duplicate singleton field",
            ),
            (
                b"HTTP/1.1 101 Switching Protocols\r\nConnection: upgrade\r\n\r\n",
                "missing selected protocol",
            ),
            (
                b"HTTP/1.1 101 Switching Protocols\r\nConnection: upgrade\r\nUpgrade: \xff\r\n\r\n",
                "invalid selected protocol",
            ),
        ] {
            let error = read_error(bytes, &offer).await;
            assert!(
                matches!(error, Error::Invalid(actual) if actual == reason),
                "{reason}: {error:?}"
            );
        }

        // Offers may also be made with raw request fields, including forms `Protocol` cannot represent.
        let mut offer = RequestMetadata::new(Method::GET);
        offer.fields = fields(&[("Connection", b"upgrade"), ("Upgrade", b"h2c, h2c/1")]);
        let mut reader = response_reader(
            b"HTTP/1.1 101 Switching Protocols\r\nConnection: upgrade\r\nUpgrade: H2C\r\n\r\nraw",
        );
        let response = reader
            .read_response(&offer, &mut observer(), &Cancellation::never())
            .await
            .unwrap();
        assert_eq!(response.protocol.as_ref().unwrap().as_str(), "H2C");
        drop(response);
        assert!(!reader.is_reusable());
        assert!(matches!(
            read_error(
                b"HTTP/1.1 101 Switching Protocols\r\nConnection: upgrade\r\nUpgrade: h2c/1\r\n\r\n",
                &offer
            )
            .await,
            Error::Invalid("unsupported selected protocol")
        ));
    });
}

fn get() -> RequestMetadata {
    RequestMetadata::new(Method::GET)
}

fn read_status(
    reader: &mut ResponseReader<Input>,
    request: &RequestMetadata,
) -> Result<u16, Error> {
    block_on(async {
        let never = Cancellation::never();
        let mut response = reader
            .read_response(request, &mut observer(), &never)
            .await?;
        while !response.body.read(&never).await?.is_end() {}
        Ok(response.status.as_u16())
    })
}

#[test]
fn an_http_10_request_persists_only_when_it_asks_to() {
    let mut keep_alive_field = get();
    keep_alive_field.version = Some(Version::Http10);
    keep_alive_field
        .fields
        .add("Connection", "keep-alive")
        .unwrap();
    let mut keep_alive_preference = get();
    keep_alive_preference.version = Some(Version::Http10);
    keep_alive_preference.persistent = Some(true);
    let mut plain = get();
    plain.version = Some(Version::Http10);
    for (request, reusable) in [
        (keep_alive_field, true),
        (keep_alive_preference, true),
        (plain, false),
    ] {
        let mut reader = ResponseReader::new(
            input(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n", usize::MAX),
            Options::default(),
        )
        .unwrap();
        assert_eq!(read_status(&mut reader, &request).unwrap(), 200);
        assert_eq!(reader.is_reusable(), reusable);
    }
}

#[test]
fn a_response_head_beyond_the_head_limit_is_rejected() {
    let options = Options {
        max_head_bytes: 16,
        ..Options::default()
    };
    let mut reader = ResponseReader::new(
        input(b"HTTP/1.1 200 OK\r\nServer: example\r\n\r\n", 4),
        options,
    )
    .unwrap();
    let result = read_status(&mut reader, &get());
    assert!(matches!(result, Err(Error::Limit("message head"))));
    assert!(!reader.is_reusable());
}

#[test]
fn ambiguous_or_malformed_response_controls_are_rejected() {
    for (bytes, reason) in [
        (
            &b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nTransfer-Encoding: chunked\r\n\r\n0\r\n\r\n"[..],
            "duplicate singleton field",
        ),
        (
            b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: keep alive\r\n\r\n",
            "invalid connection option",
        ),
    ] {
        let bytes: &'static [u8] = bytes;
        let mut reader = ResponseReader::new(input(bytes, usize::MAX), Options::default()).unwrap();
        assert!(is_invalid(read_status(&mut reader, &get()), reason), "{bytes:?}");
    }
}

#[test]
fn input_failures_are_reported_for_heads_and_tunnels() {
    let mut reader =
        ResponseReader::new(failing_input(b"HTTP/1.1 200 OK\r\n"), Options::default()).unwrap();
    assert!(is_io(
        read_status(&mut reader, &get()),
        io::ErrorKind::ConnectionReset
    ));

    let mut connect = RequestMetadata::new(Method::CONNECT);
    connect.authority = Some("example:443".into());
    let mut reader = ResponseReader::new(
        failing_input(b"HTTP/1.1 200 OK\r\n\r\nraw"),
        Options::default(),
    )
    .unwrap();
    block_on(async {
        let never = Cancellation::never();
        let mut response = reader
            .read_response(&connect, &mut observer(), &never)
            .await
            .unwrap();
        let BodyFrame::Data { data, .. } = response.body.read(&never).await.unwrap() else {
            panic!("expected raw data");
        };
        assert_eq!(data, b"raw");
        assert!(is_io(
            response.body.read(&never).await,
            io::ErrorKind::ConnectionReset
        ));
    });
}

async fn complete<R: base_endpoint::stream::ReadableStream>(reader: &mut ResponseReader<R>) {
    let get = RequestMetadata::new(Method::GET);
    let mut response = reader
        .read_response(&get, &mut observer(), &Cancellation::never())
        .await
        .unwrap();
    while !response
        .body
        .read(&Cancellation::never())
        .await
        .unwrap()
        .is_end()
    {}
}

#[test]
fn idle_probes_keep_waiting_input_and_reject_anything_else() {
    block_on(async {
        // A reader that has not completed an exchange is not read, even with input waiting.
        let segments = Segments::default();
        let mut reader = ResponseReader::new(segments.clone(), Options::default()).unwrap();
        segments.push(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok");
        assert!(reader.probe_viable());
        complete(&mut reader).await;

        // Nothing arrives while idle, so the connection stays usable for the next response.
        assert!(reader.probe_viable());
        assert!(reader.is_reusable());

        // Input arriving while idle, or left over after a response, ends reuse.
        segments.push(b"HTTP/1.1 408 Request Timeout\r\n\r\n");
        assert!(!reader.probe_viable());
        assert!(!reader.is_reusable());
        let mut reader = response_reader(b"HTTP/1.1 204 No Content\r\n\r\nX");
        complete(&mut reader).await;
        assert!(!reader.probe_viable());

        // A peer close or a failed read ends reuse.
        let mut reader = response_reader(b"HTTP/1.1 204 No Content\r\n\r\n");
        complete(&mut reader).await;
        assert!(!reader.probe_viable());
        let mut reader = ResponseReader::new(
            failing_input(b"HTTP/1.1 204 No Content\r\n\r\n"),
            Options::default(),
        )
        .unwrap();
        complete(&mut reader).await;
        assert!(!reader.probe_viable());

        // Inputs that already prohibit reuse are not viable.
        let mut reader = response_reader(b"HTTP/1.1 204 No Content\r\nConnection: close\r\n\r\n");
        complete(&mut reader).await;
        assert!(!reader.probe_viable());
        let get = RequestMetadata::new(Method::GET);
        let mut reader = response_reader(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\n");
        let response = reader
            .read_response(&get, &mut observer(), &Cancellation::never())
            .await
            .unwrap();
        drop(response);
        assert!(!reader.probe_viable());
    });
}

// Only chunked framing can carry trailers, so only a chunked response body reports that it may have them; HTTP/1.0 never does, whether fixed-length or close-delimited.
#[test]
fn only_chunked_response_bodies_may_have_trailers() {
    block_on(async {
        let never = Cancellation::never();
        for (bytes, trailers) in [
            (
                &b"HTTP/1.0 200 OK\r\nContent-Length: 2\r\nTrailer: x-digest\r\n\r\nhi"[..],
                false,
            ),
            (b"HTTP/1.0 200 OK\r\nTrailer: x-digest\r\n\r\nhi", false),
            (
                b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nTrailer: x-digest\r\n\r\nhi",
                false,
            ),
            (b"HTTP/1.1 200 OK\r\nTrailer: x-digest\r\n\r\nhi", false),
            (
                b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n2\r\nhi\r\n0\r\nx-digest: 1\r\n\r\n",
                true,
            ),
        ] {
            let mut reader = response_reader(bytes);
            let mut response = reader
                .read_response(&get(), &mut observer(), &never)
                .await
                .unwrap();
            assert_eq!(response.body.may_have_trailers(), trailers, "{bytes:?}");
            let mut headers = false;
            loop {
                let frame = response.body.read(&never).await.unwrap();
                headers |= matches!(frame, BodyFrame::Headers { .. });
                if frame.is_end() {
                    break;
                }
            }
            assert_eq!(headers, trailers, "{bytes:?}");
            // Received trailers stay out of the header section.
            assert!(!response.fields.contains("x-digest"), "{bytes:?}");
        }
    });
}