aioduct 0.2.5

#![cfg(feature = "tokio")]

use std::convert::Infallible;
#[cfg(any(
    feature = "gzip",
    feature = "deflate",
    feature = "brotli",
    feature = "zstd"
))]
use std::sync::atomic::{AtomicU32, Ordering};
#[cfg(any(
    feature = "gzip",
    feature = "deflate",
    feature = "brotli",
    feature = "zstd"
))]
use std::sync::{Arc, Mutex};
#[cfg(any(
    feature = "gzip",
    feature = "deflate",
    feature = "brotli",
    feature = "zstd"
))]
use std::time::Duration;

use bytes::Bytes;
use http_body_util::Full;
#[cfg(any(
    feature = "gzip",
    feature = "deflate",
    feature = "brotli",
    feature = "zstd"
))]
use hyper::Request;
use hyper::Response;

use aioduct::HttpEngineSend;
use aioduct::runtime::TokioRuntime;
use aioduct::runtime::tokio_rt::TcpConnector;

use aioduct_test_server::h1::h1_server_with;
#[cfg(any(
    feature = "gzip",
    feature = "deflate",
    feature = "brotli",
    feature = "zstd"
))]
use aioduct_test_server::raw::raw_streaming_server;

#[cfg(feature = "gzip")]
#[tokio::test]
async fn test_gzip_decompression() {
    use flate2::Compression;
    use flate2::write::GzEncoder;
    use std::io::Write;

    let handler = |_req: Request<hyper::body::Incoming>| async {
        let mut encoder = GzEncoder::new(Vec::new(), Compression::fast());
        encoder.write_all(b"hello compressed world").unwrap();
        let compressed = encoder.finish().unwrap();

        let resp = Response::builder()
            .header("content-encoding", "gzip")
            .body(Full::new(Bytes::from(compressed)))
            .unwrap();
        Ok::<_, Infallible>(resp)
    };
    let (addr, _counter) = h1_server_with(handler).await;
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();

    assert!(!resp.headers().contains_key("content-encoding"));
    let text = resp.text().await.unwrap();
    assert_eq!(text, "hello compressed world");
}
#[cfg(feature = "gzip")]
#[tokio::test]
async fn test_gzip_accept_encoding_header() {
    let handler = |req: Request<hyper::body::Incoming>| async move {
        let accept = req
            .headers()
            .get("accept-encoding")
            .map(|v| v.to_str().unwrap().to_string())
            .unwrap_or_default();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(accept))))
    };
    let (addr, _counter) = h1_server_with(handler).await;
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let text = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();

    assert!(text.contains("gzip"));
}
#[cfg(feature = "gzip")]
#[tokio::test]
async fn test_no_decompression_passthrough() {
    use flate2::Compression;
    use flate2::write::GzEncoder;
    use std::io::Write;

    let handler = |_req: Request<hyper::body::Incoming>| async {
        let mut encoder = GzEncoder::new(Vec::new(), Compression::fast());
        encoder.write_all(b"raw gzip data").unwrap();
        let compressed = encoder.finish().unwrap();

        let resp = Response::builder()
            .header("content-encoding", "gzip")
            .body(Full::new(Bytes::from(compressed)))
            .unwrap();
        Ok::<_, Infallible>(resp)
    };
    let (addr, _counter) = h1_server_with(handler).await;
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
        .no_decompression()
        .build()
        .unwrap();
    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();

    assert!(resp.headers().contains_key("content-encoding"));
    let bytes = resp.bytes().await.unwrap();
    // Should be raw gzip, not decompressed
    assert_ne!(bytes.as_ref(), b"raw gzip data");
}
#[cfg(feature = "deflate")]
#[tokio::test]
async fn test_deflate_decompression() {
    use flate2::Compression;
    use flate2::write::ZlibEncoder;
    use std::io::Write;

    let handler = |_req: Request<hyper::body::Incoming>| async {
        let mut encoder = ZlibEncoder::new(Vec::new(), Compression::fast());
        encoder.write_all(b"deflate test payload").unwrap();
        let compressed = encoder.finish().unwrap();

        let resp = Response::builder()
            .header("content-encoding", "deflate")
            .body(Full::new(Bytes::from(compressed)))
            .unwrap();
        Ok::<_, Infallible>(resp)
    };
    let (addr, _counter) = h1_server_with(handler).await;
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let text = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap()
        .text()
        .await
        .unwrap();

    assert_eq!(text, "deflate test payload");
}
#[tokio::test]
async fn test_get_no_content_headers() {
    let (addr, _counter) = h1_server_with(|req| async move {
        assert_eq!(req.method(), "GET");
        assert!(
            req.headers().get("content-length").is_none(),
            "GET should not have content-length"
        );
        assert!(
            req.headers().get("content-type").is_none(),
            "GET should not have content-type"
        );
        assert!(
            req.headers().get("transfer-encoding").is_none(),
            "GET should not have transfer-encoding"
        );
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from("ok"))))
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), http::StatusCode::OK);
}
#[cfg(feature = "gzip")]
#[tokio::test]
async fn test_gzip_empty_body_head_request() {
    let (addr, _counter) = h1_server_with(|req| async move {
        assert_eq!(req.method(), "HEAD");
        Ok::<_, Infallible>(
            Response::builder()
                .header("content-encoding", "gzip")
                .body(Full::new(Bytes::new()))
                .unwrap(),
        )
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .head(&format!("http://{addr}/gzip"))
        .unwrap()
        .send()
        .await
        .unwrap();

    let body = resp.text().await.unwrap();
    assert_eq!(body, "");
}
#[cfg(feature = "gzip")]
#[tokio::test]
async fn test_custom_accept_encoding_preserved() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let accept_encoding = req
            .headers()
            .get("accept-encoding")
            .map(|v| v.to_str().unwrap().to_owned())
            .unwrap_or_default();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(accept_encoding))))
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .header(
            http::header::ACCEPT_ENCODING,
            http::header::HeaderValue::from_static("identity"),
        )
        .send()
        .await
        .unwrap();

    let body = resp.text().await.unwrap();
    assert_eq!(body, "identity");
}

#[cfg(any(
    feature = "gzip",
    feature = "deflate",
    feature = "brotli",
    feature = "zstd"
))]
#[path = "decompression/codec_coverage.rs"]
mod codec_coverage;

#[cfg(any(
    feature = "gzip",
    feature = "deflate",
    feature = "brotli",
    feature = "zstd"
))]
#[path = "decompression/encoding_headers.rs"]
mod encoding_headers;

// ── Body malformed/boundary tests ─────────────────────────────────────

#[cfg(feature = "gzip")]
#[tokio::test]
async fn malformed_gzip_body_returns_error() {
    use tokio::io::AsyncWriteExt;

    let addr = raw_streaming_server(move |_req, mut stream| async move {
        let body = b"this is not valid gzip compressed data";
        let header = format!(
            "HTTP/1.1 200 OK\r\nContent-Encoding: gzip\r\nContent-Length: {}\r\n\r\n",
            body.len()
        );
        stream.write_all(header.as_bytes()).await.unwrap();
        stream.write_all(body).await.unwrap();
        stream.flush().await.unwrap();
        stream.shutdown().await.unwrap();
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();

    let result = resp.text().await;
    assert!(
        result.is_err(),
        "malformed gzip body should cause decompression error, got: {:?}",
        result.ok()
    );
}

/// Server sends Content-Length: 5 but writes far more bytes on the wire.
/// After the client reads the 5-byte body, the remaining bytes corrupt the
/// connection, so the next request either fails or opens a new connection.
#[tokio::test]
async fn content_length_mismatch_too_long_poisons_reuse() {
    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::time::Duration;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::TcpListener;

    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let conn_count = Arc::new(AtomicUsize::new(0));

    tokio::spawn({
        let conn_count = conn_count.clone();
        async move {
            loop {
                let (mut stream, _) = match listener.accept().await {
                    Ok(v) => v,
                    Err(_) => continue,
                };
                conn_count.fetch_add(1, Ordering::SeqCst);
                tokio::spawn(async move {
                    let mut buf = [0u8; 4096];
                    let n = match stream.read(&mut buf).await {
                        Ok(0) | Err(_) => return,
                        Ok(n) => n,
                    };
                    if !buf[..n].windows(4).any(|w| w == b"\r\n\r\n") {
                        return;
                    }

                    // Content-Length claims 5 bytes but we write ~105.
                    let extra = "x".repeat(100);
                    let response = format!(
                        "HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: keep-alive\r\n\r\nhello{extra}"
                    );
                    let _ = stream.write_all(response.as_bytes()).await;
                    let _ = stream.flush().await;

                    // Leave the connection open; the 100 extra bytes will
                    // corrupt any subsequent request the client tries to
                    // pipeline on the same TCP stream.
                    let _ = tokio::time::timeout(Duration::from_millis(300), stream.read(&mut buf))
                        .await;
                });
            }
        }
    });

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
        .pool_idle_timeout(Duration::from_secs(60))
        .timeout(Duration::from_secs(5))
        .build()
        .unwrap();

    let url = format!("http://{addr}/");

    // First request: reads exactly 5 bytes (Content-Length) — succeeds.
    let resp = client.get(&url).unwrap().send().await.unwrap();
    let body = resp.text().await.unwrap();
    assert_eq!(body, "hello");

    // Give the pool a moment to return the (now-corrupted) connection.
    tokio::time::sleep(Duration::from_millis(50)).await;

    // Second request: either fails (dirty connection) or succeeds but
    // must use a fresh connection.
    let before = conn_count.load(Ordering::SeqCst);
    let result = client.get(&url).unwrap().send().await;
    match result {
        Ok(resp) => {
            let _ = resp.text().await.unwrap();
            let after = conn_count.load(Ordering::SeqCst);
            assert!(
                after > before,
                "expected a new connection after Content-Length \
                 mismatch (before={before}, after={after})"
            );
        }
        Err(_) => {
            // Corrupted connection produced an error — acceptable.
        }
    }
}

#[cfg(feature = "gzip")]
#[tokio::test]
async fn content_length_removed_after_decompression() {
    use flate2::Compression;
    use flate2::write::GzEncoder;
    use std::io::Write;

    let content = "content length should disappear";
    let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
    encoder.write_all(content.as_bytes()).unwrap();
    let compressed = encoder.finish().unwrap();

    let (addr, _counter) = h1_server_with(move |_req| {
        let compressed = compressed.clone();
        async move {
            Ok::<_, Infallible>(
                Response::builder()
                    .header("content-encoding", "gzip")
                    .header("content-length", compressed.len().to_string())
                    .body(Full::new(Bytes::from(compressed)))
                    .unwrap(),
            )
        }
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();

    // Content-Length must be removed because the decompressed body size
    // no longer matches the original value.
    assert!(
        resp.headers().get("content-length").is_none(),
        "Content-Length should be stripped after decompression"
    );

    let text = resp.text().await.unwrap();
    assert_eq!(text, content);
}

#[cfg(feature = "gzip")]
#[tokio::test]
async fn decompressed_body_empty_is_ok() {
    use flate2::Compression;
    use flate2::write::GzEncoder;

    // A gzip stream with no payload — just header + footer.
    let encoder = GzEncoder::new(Vec::new(), Compression::default());
    let compressed = encoder.finish().unwrap();
    // Should still be a valid gzip file (header + trailer, zero payload).

    let (addr, _counter) = h1_server_with(move |_req| {
        let compressed = compressed.clone();
        async move {
            Ok::<_, Infallible>(
                Response::builder()
                    .header("content-encoding", "gzip")
                    .body(Full::new(Bytes::from(compressed)))
                    .unwrap(),
            )
        }
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();

    let text = resp.text().await.unwrap();
    assert_eq!(
        text, "",
        "empty gzip body should decompress to empty string"
    );
}

// ── Round-trip and trailer pass-through tests ──────────────────────────

/// 64KB gzip round-trip: compress, serve, decompress, verify.
#[cfg(feature = "gzip")]
#[tokio::test]
async fn decompress_gzip_round_trip_with_large_body() {
    use flate2::Compression;
    use flate2::write::GzEncoder;
    use std::io::Write;

    let content = "A".repeat(65536); // 64 KB
    let mut encoder = GzEncoder::new(Vec::new(), Compression::fast());
    encoder.write_all(content.as_bytes()).unwrap();
    let compressed = encoder.finish().unwrap();

    let (addr, _counter) = h1_server_with(move |_req| {
        let compressed = compressed.clone();
        async move {
            Ok::<_, Infallible>(
                Response::builder()
                    .header("content-encoding", "gzip")
                    .header("content-length", compressed.len().to_string())
                    .body(Full::new(Bytes::from(compressed)))
                    .unwrap(),
            )
        }
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();

    let text = resp.text().await.unwrap();
    assert_eq!(text, content);
}

/// Send gzip-compressed chunked response followed by a trailer header.
/// Verify the decompressed body is correct AND that `TrailersReceived`
/// fires with the expected trailer headers.
#[cfg(feature = "gzip")]
#[tokio::test]
async fn trailer_frame_passes_through_decompress() {
    use flate2::Compression;
    use flate2::write::GzEncoder;
    use std::io::Write;
    use tokio::io::AsyncWriteExt;

    use aioduct::observer::{ConnectionEvent, RequestEvent, RequestObserver, RequestPhase};

    let content = "hello trailer decompress";
    let mut encoder = GzEncoder::new(Vec::new(), Compression::fast());
    encoder.write_all(content.as_bytes()).unwrap();
    let compressed = encoder.finish().unwrap();

    let events: Arc<Mutex<Vec<RequestPhase>>> = Arc::new(Mutex::new(Vec::new()));
    let events_clone = events.clone();

    struct TrailerObserver(Arc<Mutex<Vec<RequestPhase>>>);
    impl RequestObserver for TrailerObserver {
        fn on_event(&self, event: &RequestEvent) {
            self.0.lock().unwrap().push(event.phase.clone());
        }
        fn on_connection_event(&self, _event: &ConnectionEvent) {}
    }

    let addr = raw_streaming_server(move |_req, mut stream| {
        let compressed = compressed.clone();
        async move {
            let chunk_header = format!("{:x}\r\n", compressed.len());
            let response_header = "HTTP/1.1 200 OK\r\n\
                 Content-Encoding: gzip\r\n\
                 Transfer-Encoding: chunked\r\n\
                 Trailer: x-response-time\r\n\
                 \r\n";
            stream.write_all(response_header.as_bytes()).await.unwrap();
            stream.write_all(chunk_header.as_bytes()).await.unwrap();
            stream.write_all(&compressed).await.unwrap();
            stream.write_all(b"\r\n").await.unwrap();
            // Terminating chunk + trailer
            stream.write_all(b"0\r\n").await.unwrap();
            stream.write_all(b"x-response-time: 42\r\n").await.unwrap();
            stream.write_all(b"\r\n").await.unwrap();
            stream.flush().await.unwrap();
        }
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
        .timeout(Duration::from_secs(5))
        .request_observer(TrailerObserver(events_clone))
        .build()
        .unwrap();

    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();

    // Use into_bytes_stream so the observer fires TrailersReceived.
    let mut stream = resp.into_bytes_stream();
    let mut body = Vec::new();
    while let Some(chunk) = stream.next().await {
        body.extend_from_slice(&chunk.unwrap());
    }

    assert_eq!(
        String::from_utf8(body).unwrap(),
        content,
        "decompressed body must match original"
    );

    let captured = events.lock().unwrap();
    let has_trailers = captured.iter().any(|p| {
        matches!(p, RequestPhase::TrailersReceived { headers }
            if headers.iter().any(|(k, v)| k == "x-response-time" && v == "42"))
    });
    assert!(
        has_trailers,
        "expected TrailersReceived with x-response-time: 42, got: {captured:?}"
    );
}

/// Round-trip brotli: compress, serve with Content-Encoding: br, decompress.
#[cfg(feature = "brotli")]
#[tokio::test]
async fn decompress_brotli_round_trip() {
    use std::io::Write;

    let content = "hello brotli round trip test payload with sufficient length to compress well";
    let mut compressed = Vec::new();
    {
        let mut writer = brotli::CompressorWriter::new(&mut compressed, 4096, 6, 22);
        writer.write_all(content.as_bytes()).unwrap();
        drop(writer);
    }

    let (addr, _counter) = h1_server_with(move |_req| {
        let compressed = compressed.clone();
        async move {
            Ok::<_, Infallible>(
                Response::builder()
                    .header("content-encoding", "br")
                    .body(Full::new(Bytes::from(compressed)))
                    .unwrap(),
            )
        }
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();

    let text = resp.text().await.unwrap();
    assert_eq!(text, content);
}

/// Round-trip zstd: compress, serve with Content-Encoding: zstd, decompress.
#[cfg(feature = "zstd")]
#[tokio::test]
async fn decompress_zstd_round_trip() {
    let content = "hello zstd round trip test payload with sufficient length to compress well";
    let compressed = zstd::encode_all(content.as_bytes(), 3).unwrap();

    let (addr, _counter) = h1_server_with(move |_req| {
        let compressed = compressed.clone();
        async move {
            Ok::<_, Infallible>(
                Response::builder()
                    .header("content-encoding", "zstd")
                    .body(Full::new(Bytes::from(compressed)))
                    .unwrap(),
            )
        }
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();

    let text = resp.text().await.unwrap();
    assert_eq!(text, content);
}

// ── Decompression Edge-Case Tests ──────────────────────────────────────

/// Corrupt gzip body (valid header/footer, zeroed payload) must produce a
/// decode error. The error is at the application (decompression) level, not
/// transport corruption, so the pooled connection must remain usable: a second
/// request on the same client must succeed.
#[cfg(feature = "gzip")]
#[tokio::test]
async fn corrupt_gzip_body_propagates_decode_error() {
    use flate2::Compression;
    use flate2::write::GzEncoder;
    use std::io::Write;

    // Build a valid gzip body, then corrupt its middle bytes.
    let valid_content = "second request valid content";
    let valid = {
        let mut e = GzEncoder::new(Vec::new(), Compression::default());
        e.write_all(valid_content.as_bytes()).unwrap();
        e.finish().unwrap()
    };

    let corrupt_original =
        "first request content that will be corrupted in the middle of the gzip stream";
    let mut corrupt = {
        let mut e = GzEncoder::new(Vec::new(), Compression::default());
        e.write_all(corrupt_original.as_bytes()).unwrap();
        e.finish().unwrap()
    };

    // Keep the gzip header and footer intact; zero-fill the middle section so
    // the decoder recognizes the format but fails during stream decompression.
    let start = corrupt.len() / 3;
    let end = (corrupt.len() * 2) / 3;
    for byte in &mut corrupt[start..end] {
        *byte = 0;
    }

    let request_count = Arc::new(AtomicU32::new(0));

    let (addr, _counter) = h1_server_with({
        let request_count = request_count.clone();
        let valid = valid.clone();
        let corrupt = corrupt.clone();
        move |_req: Request<hyper::body::Incoming>| {
            let count = request_count.fetch_add(1, Ordering::SeqCst);
            let body = if count == 0 {
                corrupt.clone()
            } else {
                valid.clone()
            };
            async move {
                Ok::<_, Infallible>(
                    Response::builder()
                        .header("content-encoding", "gzip")
                        .body(Full::new(Bytes::from(body)))
                        .unwrap(),
                )
            }
        }
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();

    // Request 1: corrupt body → must error.
    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();
    let result = resp.text().await;
    assert!(
        result.is_err(),
        "corrupt gzip body must produce a decode error, got: {:?}",
        result.ok()
    );

    // Request 2: same client, valid body → must succeed.
    // The pool connection was NOT evicted by the application-level error.
    let resp2 = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();
    let text2 = resp2.text().await.unwrap();
    assert_eq!(text2, valid_content);
}

/// Content-Encoding declares brotli ("br") but the body is actually gzip.
/// With both decompressors enabled, the brotli decoder receives gzip bytes
/// and must return an error — not silently pass through or return wrong data.
#[cfg(all(feature = "gzip", feature = "brotli"))]
#[tokio::test]
async fn content_encoding_brotli_with_gzip_body_errors() {
    use flate2::Compression;
    use flate2::write::GzEncoder;
    use std::io::Write;

    let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
    encoder.write_all(b"gzip body served as brotli").unwrap();
    let gzip_body = encoder.finish().unwrap();

    let (addr, _counter) = h1_server_with(move |_req| {
        let gzip_body = gzip_body.clone();
        async move {
            Ok::<_, Infallible>(
                Response::builder()
                    .header("content-encoding", "br")
                    .body(Full::new(Bytes::from(gzip_body)))
                    .unwrap(),
            )
        }
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();

    let result = resp.text().await;
    assert!(
        result.is_err(),
        "brotli Content-Encoding with gzip body must error, got: {:?}",
        result.ok()
    );
}

/// Decompression bomb: 100 MB of zeros → ~100 KB gzip.
///
/// The `max_decoded_size(1_000_000)` limit (1 MB) should cause the body read
/// to error before allocating the full 100 MB.
#[cfg(feature = "gzip")]
#[tokio::test]
async fn gzip_bomb_rejected_by_max_decoded_size() {
    use flate2::Compression;
    use flate2::write::GzEncoder;
    use std::io::Write;

    let big = vec![0u8; 100_000_000]; // 100 MB zeros
    let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
    encoder.write_all(&big).unwrap();
    let compressed = encoder.finish().unwrap();

    let (addr, _counter) = h1_server_with(move |_req| {
        let compressed = compressed.clone();
        async move {
            Ok::<_, Infallible>(
                Response::builder()
                    .header("content-encoding", "gzip")
                    .header("content-length", compressed.len().to_string())
                    .body(Full::new(Bytes::from(compressed)))
                    .unwrap(),
            )
        }
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
        .max_decoded_size(Some(1_000_000)) // 1 MB limit
        .timeout(Duration::from_secs(10))
        .build()
        .unwrap();

    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .send()
        .await
        .unwrap();

    let result = resp.text().await;
    assert!(
        result.is_err(),
        "decompression bomb must be rejected by max_decoded_size limit"
    );
    let err_msg = result.unwrap_err().to_string();
    assert!(
        err_msg.contains("exceeds max size"),
        "error should mention max size, got: {err_msg}"
    );
}

#[cfg(feature = "gzip")]
#[path = "decompression/no_decompression.rs"]
mod no_decompression;