aioduct 0.2.4

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

use std::convert::Infallible;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};

use bytes::Bytes;
use http_body_util::Full;
use hyper::Response;

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

use aioduct_test_server::h1::{echo_headers, h1_server, h1_server_with};
use aioduct_test_server::raw::raw_server;

use http_body_util::BodyExt;

#[tokio::test]
async fn test_get_request() {
    let (addr, _counter) = h1_server().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);
    let body = resp.text().await.unwrap();
    assert_eq!(body, "hello aioduct");
}

#[tokio::test]
async fn test_post_request() {
    let (addr, _counter) = h1_server().await;
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();

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

    assert_eq!(resp.status(), http::StatusCode::OK);
}

#[tokio::test]
async fn test_connection_reuse() {
    let (addr, _counter) = h1_server().await;
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let url = format!("http://{addr}/");

    let resp1 = client.get(&url).unwrap().send().await.unwrap();
    assert_eq!(resp1.status(), http::StatusCode::OK);
    let _ = resp1.text().await.unwrap();

    let resp2 = client.get(&url).unwrap().send().await.unwrap();
    assert_eq!(resp2.status(), http::StatusCode::OK);
    let body = resp2.text().await.unwrap();
    assert_eq!(body, "hello aioduct");
}

#[tokio::test]
async fn test_host_header_and_path() {
    let (addr, _counter) = h1_server_with(echo_headers).await;
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();

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

    assert_eq!(resp.status(), http::StatusCode::OK);
    let body = resp.text().await.unwrap();
    assert!(
        body.contains(&format!("host={addr}")),
        "expected Host header to be set, got: {body}"
    );
    assert!(
        body.contains("path=/some/path"),
        "expected path-only URI, got: {body}"
    );
}

#[tokio::test]
async fn test_custom_header() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let custom = req
            .headers()
            .get("x-custom")
            .map(|v| v.to_str().unwrap_or(""))
            .unwrap_or("missing");
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(custom.to_string()))))
    })
    .await;
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();

    let resp = client
        .get(&format!("http://{addr}/"))
        .unwrap()
        .header_str("x-custom", "test-value")
        .unwrap()
        .send()
        .await
        .unwrap();

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

#[tokio::test]
async fn test_invalid_url() {
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    assert!(client.get("not a url").is_err());
}

#[tokio::test]
async fn test_missing_scheme() {
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    assert!(client.get("127.0.0.1/path").is_err());
}
#[tokio::test]
async fn test_query_params() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let query = req.uri().query().unwrap_or("none").to_owned();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(query))))
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .get(&format!("http://{addr}/search"))
        .unwrap()
        .query(&[("q", "hello world"), ("page", "1")])
        .send()
        .await
        .unwrap();

    let body = resp.text().await.unwrap();
    assert_eq!(body, "q=hello%20world&page=1");
}

#[tokio::test]
async fn test_default_user_agent() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let ua = req
            .headers()
            .get("user-agent")
            .map(|v| v.to_str().unwrap_or("").to_owned())
            .unwrap_or_default();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(ua))))
    })
    .await;

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

    let body = resp.text().await.unwrap();
    assert!(
        body.starts_with("aioduct/"),
        "expected default User-Agent, got: {body}"
    );
}

#[tokio::test]
async fn test_custom_default_headers() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let custom = req
            .headers()
            .get("x-default")
            .map(|v| v.to_str().unwrap_or("").to_owned())
            .unwrap_or_default();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(custom))))
    })
    .await;

    let mut headers = http::HeaderMap::new();
    headers.insert("x-default", "from-client".parse().unwrap());
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
        .default_headers(headers)
        .build()
        .unwrap();

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

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

#[tokio::test]
async fn test_request_headers_override_defaults() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let ua = req
            .headers()
            .get("user-agent")
            .map(|v| v.to_str().unwrap_or("").to_owned())
            .unwrap_or_default();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(ua))))
    })
    .await;

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

    let body = resp.text().await.unwrap();
    assert_eq!(body, "custom-agent/1.0");
}

#[tokio::test]
async fn test_put_request() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let method = req.method().to_string();
        let body = req.into_body().collect().await.unwrap().to_bytes();
        let resp_body = format!("method={method} body={}", String::from_utf8_lossy(&body));
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(resp_body))))
    })
    .await;

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

    let body = resp.text().await.unwrap();
    assert!(body.contains("method=PUT"), "expected PUT, got: {body}");
    assert!(
        body.contains("body=update data"),
        "expected body, got: {body}"
    );
}

#[tokio::test]
async fn test_patch_request() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let method = req.method().to_string();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(method))))
    })
    .await;

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

    assert_eq!(resp.text().await.unwrap(), "PATCH");
}

#[tokio::test]
async fn test_delete_request() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let method = req.method().to_string();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(method))))
    })
    .await;

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

    assert_eq!(resp.text().await.unwrap(), "DELETE");
}

#[tokio::test]
async fn test_head_request() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let method = req.method().to_string();
        Ok::<_, Infallible>(
            Response::builder()
                .header("x-method", method)
                .header("content-length", "1000")
                .body(Full::new(Bytes::new()))
                .unwrap(),
        )
    })
    .await;

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

    assert_eq!(resp.status(), http::StatusCode::OK);
    assert_eq!(
        resp.headers().get("x-method").unwrap().to_str().unwrap(),
        "HEAD"
    );
    assert_eq!(resp.content_length(), Some(1000));
}

#[tokio::test]
async fn test_query_params_with_existing_query() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let query = req.uri().query().unwrap_or("").to_string();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(query))))
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .get(&format!("http://{addr}/?existing=1"))
        .unwrap()
        .query(&[("extra", "2")])
        .send()
        .await
        .unwrap();

    let body = resp.text().await.unwrap();
    assert!(
        body.contains("existing=1"),
        "expected existing, got: {body}"
    );
    assert!(body.contains("extra=2"), "expected extra, got: {body}");
}

#[tokio::test]
async fn test_no_default_headers() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let ua = req
            .headers()
            .get("user-agent")
            .map(|v| v.to_str().unwrap_or("").to_owned())
            .unwrap_or_else(|| "none".to_owned());
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(ua))))
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
        .no_default_headers()
        .build()
        .unwrap();

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

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

#[tokio::test]
async fn test_custom_method() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let method = req.method().to_string();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(method))))
    })
    .await;

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

    assert_eq!(resp.text().await.unwrap(), "OPTIONS");
}

#[tokio::test]
async fn test_multiple_headers_same_name() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let values: Vec<String> = req
            .headers()
            .get_all("x-multi")
            .iter()
            .map(|v| v.to_str().unwrap().to_string())
            .collect();
        let body = values.join(",");
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(body))))
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let mut headers = http::HeaderMap::new();
    headers.append("x-multi", "value1".parse().unwrap());
    headers.append("x-multi", "value2".parse().unwrap());

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

    let body = resp.text().await.unwrap();
    assert!(body.contains("value1"), "expected value1, got: {body}");
    assert!(body.contains("value2"), "expected value2, got: {body}");
}

#[tokio::test]
async fn auto_headers_no_accept_by_default() {
    let (addr, _counter) = h1_server_with(|req| async move {
        assert_eq!(req.method(), "GET");
        let accept = req
            .headers()
            .get("accept")
            .map(|v| v.to_str().unwrap().to_owned());
        let body = match accept {
            Some(v) => format!("accept={v}"),
            None => "accept=none".to_string(),
        };
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(body))))
    })
    .await;

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

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

#[tokio::test]
async fn donot_set_content_length_0_if_have_no_body() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let headers = req.headers();
        assert!(
            headers.get("content-length").is_none(),
            "GET should not set content-length"
        );
        assert!(
            headers.get("content-type").is_none(),
            "GET should not set content-type"
        );
        assert!(
            headers.get("transfer-encoding").is_none(),
            "GET should not set transfer-encoding"
        );
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::new())))
    })
    .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);
}

#[tokio::test]
async fn custom_user_agent_via_builder() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let ua = req
            .headers()
            .get("user-agent")
            .map(|v| v.to_str().unwrap().to_owned())
            .unwrap_or_default();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(ua))))
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
        .user_agent("aioduct-test-agent")
        .build()
        .unwrap();

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

    let body = resp.text().await.unwrap();
    assert_eq!(body, "aioduct-test-agent");
}

#[tokio::test]
async fn response_text_and_content_length() {
    let (addr, _counter) = h1_server_with(|_req| async {
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from("Hello"))))
    })
    .await;

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

    assert_eq!(resp.content_length(), Some(5));
    let text = resp.text().await.unwrap();
    assert_eq!(text, "Hello");
}

#[tokio::test]
async fn response_bytes_and_content_length() {
    let (addr, _counter) = h1_server_with(|_req| async {
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from("Hello"))))
    })
    .await;

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

    assert_eq!(resp.content_length(), Some(5));
    let bytes = resp.bytes().await.unwrap();
    assert_eq!(&bytes[..], b"Hello");
}

#[cfg(feature = "json")]
#[tokio::test]
async fn response_json_string() {
    let (addr, _counter) = h1_server_with(|_req| async {
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from("\"Hello\""))))
    })
    .await;

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

    let text: String = resp.json().await.unwrap();
    assert_eq!(text, "Hello");
}

#[cfg(feature = "json")]
#[tokio::test]
async fn json_content_type_default() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let ct = req
            .headers()
            .get("content-type")
            .map(|v| v.to_str().unwrap().to_owned())
            .unwrap_or_default();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(ct))))
    })
    .await;

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

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

#[cfg(feature = "json")]
#[tokio::test]
async fn json_content_type_not_overridden_if_set() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let ct = req
            .headers()
            .get("content-type")
            .map(|v| v.to_str().unwrap().to_owned())
            .unwrap_or_default();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(ct))))
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .post(&format!("http://{addr}/"))
        .unwrap()
        .header(
            http::header::CONTENT_TYPE,
            http::HeaderValue::from_static("application/vnd.api+json"),
        )
        .json(&serde_json::json!({"body": "json"}))
        .unwrap()
        .send()
        .await
        .unwrap();

    let body = resp.text().await.unwrap();
    assert_eq!(body, "application/vnd.api+json");
}

#[tokio::test]
async fn body_pipe_response_to_post() {
    let (addr, _counter) = h1_server_with(|req| async move {
        if req.uri().path() == "/get" {
            Ok::<_, Infallible>(Response::new(Full::new(Bytes::from("pipe me"))))
        } else {
            assert_eq!(req.uri().path(), "/pipe");
            let full = req.into_body().collect().await.unwrap().to_bytes();
            assert_eq!(&full[..], b"pipe me");
            Ok(Response::new(Full::new(Bytes::from("piped"))))
        }
    })
    .await;

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

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

    assert_eq!(res1.status(), http::StatusCode::OK);
    assert_eq!(res1.content_length(), Some(7));

    let body_bytes = res1.bytes().await.unwrap();

    let res2 = client
        .post(&format!("http://{addr}/pipe"))
        .unwrap()
        .body(body_bytes.to_vec())
        .send()
        .await
        .unwrap();

    assert_eq!(res2.status(), http::StatusCode::OK);
    assert_eq!(res2.text().await.unwrap(), "piped");
}

#[tokio::test]
async fn raw_server_custom_response() {
    let addr =
        raw_server(|_req| async { b"HTTP/1.1 200 OK\r\nContent-Length: 3\r\n\r\nraw".to_vec() })
            .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);
    assert_eq!(resp.text().await.unwrap(), "raw");
}

#[tokio::test]
async fn text_part() {
    let form = aioduct::Multipart::new().text("foo", "bar");
    let expected_body = format!(
        "--{0}\r\nContent-Disposition: form-data; name=\"foo\"\r\n\r\nbar\r\n--{0}--\r\n",
        form.boundary()
    );
    let ct = form.content_type();

    let (addr, _counter) = h1_server_with(move |req| {
        let ct = ct.clone();
        let expected_body = expected_body.clone();
        async move {
            assert_eq!(req.method(), "POST");
            assert_eq!(req.headers()["content-type"], ct);
            assert_eq!(
                req.headers()["content-length"],
                expected_body.len().to_string()
            );
            let full = req.into_body().collect().await.unwrap().to_bytes();
            assert_eq!(full, expected_body.as_bytes());
            Ok::<_, Infallible>(Response::new(Full::new(Bytes::new())))
        }
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .post(&format!("http://{addr}/multipart/1"))
        .unwrap()
        .multipart(form)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), http::StatusCode::OK);
}

#[tokio::test]
async fn multipart_custom_boundary_and_subtype() {
    let form = aioduct::Multipart::new()
        .with_boundary("WebKitFormBoundaryABC123")
        .unwrap()
        .subtype("mixed")
        .unwrap()
        .text("foo", "bar");
    let expected_ct = form.content_type();
    let expected_body = format!(
        "--{0}\r\nContent-Disposition: form-data; name=\"foo\"\r\n\r\nbar\r\n--{0}--\r\n",
        form.boundary()
    );
    assert_eq!(
        expected_ct,
        "multipart/mixed; boundary=\"WebKitFormBoundaryABC123\""
    );

    let (addr, _counter) = h1_server_with(move |req| {
        let ct = expected_ct.clone();
        let expected_body = expected_body.clone();
        async move {
            assert_eq!(req.headers()["content-type"], ct);
            let full = req.into_body().collect().await.unwrap().to_bytes();
            assert_eq!(full, expected_body.as_bytes());
            Ok::<_, Infallible>(Response::new(Full::new(Bytes::new())))
        }
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .post(&format!("http://{addr}/multipart/custom"))
        .unwrap()
        .multipart(form)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), http::StatusCode::OK);
}

#[tokio::test]
async fn stream_part() {
    let stream_data = "part1 part2";
    let stream_body: aioduct::body::RequestBodySend =
        http_body_util::Full::new(Bytes::from(stream_data))
            .map_err(|never| match never {})
            .boxed_unsync();

    let form = aioduct::Multipart::new()
        .text("foo", "bar")
        .part(aioduct::multipart::Part::stream("part_stream", stream_body));

    let expected_body = format!(
        "--{0}\r\nContent-Disposition: form-data; name=\"foo\"\r\n\r\nbar\r\n--{0}\r\nContent-Disposition: form-data; name=\"part_stream\"\r\n\r\n{1}\r\n--{0}--\r\n",
        form.boundary(),
        stream_data,
    );
    let ct = form.content_type();

    let (addr, _counter) = h1_server_with(move |req| {
        let ct = ct.clone();
        let expected_body = expected_body.clone();
        async move {
            assert_eq!(req.method(), "POST");
            assert_eq!(req.headers()["content-type"], ct);
            assert_eq!(req.headers()["transfer-encoding"], "chunked");
            let full = req.into_body().collect().await.unwrap().to_bytes();
            assert_eq!(full, expected_body.as_bytes());
            Ok::<_, Infallible>(Response::new(Full::new(Bytes::new())))
        }
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .post(&format!("http://{addr}/multipart/stream"))
        .unwrap()
        .multipart(form)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), http::StatusCode::OK);
}

#[tokio::test]
async fn file_part() {
    let file_contents = "file contents here";
    let form = aioduct::Multipart::new().file(
        "upload",
        "test.txt",
        "application/octet-stream",
        file_contents.as_bytes().to_vec(),
    );

    let expected_body = format!(
        "--{0}\r\nContent-Disposition: form-data; name=\"upload\"; filename=\"test.txt\"\r\nContent-Type: application/octet-stream\r\n\r\n{1}\r\n--{0}--\r\n",
        form.boundary(),
        file_contents,
    );
    let ct = form.content_type();

    let (addr, _counter) = h1_server_with(move |req| {
        let ct = ct.clone();
        let expected_body = expected_body.clone();
        async move {
            assert_eq!(req.method(), "POST");
            assert_eq!(req.headers()["content-type"], ct);
            assert_eq!(
                req.headers()["content-length"],
                expected_body.len().to_string()
            );
            let full = req.into_body().collect().await.unwrap().to_bytes();
            assert_eq!(full, expected_body.as_bytes());
            Ok::<_, Infallible>(Response::new(Full::new(Bytes::new())))
        }
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();
    let resp = client
        .post(&format!("http://{addr}/multipart/file"))
        .unwrap()
        .multipart(form)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), http::StatusCode::OK);
}

#[tokio::test]
async fn raw_server_chunked_response() {
    let addr = raw_server(|_req| async {
        b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n6\r\n world\r\n0\r\n\r\n"
            .to_vec()
    })
    .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);
    assert_eq!(resp.text().await.unwrap(), "hello world");
}

#[tokio::test]
async fn custom_method_propfind() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let method = req.method().to_string();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(method))))
    })
    .await;

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

    assert_eq!(resp.text().await.unwrap(), "PROPFIND");
}

#[tokio::test]
async fn no_default_headers_removes_user_agent() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let ua = req
            .headers()
            .get("user-agent")
            .map(|v| v.to_str().unwrap_or("").to_owned())
            .unwrap_or_else(|| "none".to_owned());
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(ua))))
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
        .no_default_headers()
        .build()
        .unwrap();

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

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

#[tokio::test]
async fn response_headers_accessible_after_bytes() {
    let (addr, _counter) = h1_server().await;
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();

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

    // Headers should be accessible before consuming the body
    assert!(
        !resp.headers().is_empty(),
        "response headers should not be empty"
    );
    assert!(
        resp.headers().contains_key("content-length"),
        "content-length header should be present"
    );
    assert_eq!(resp.content_length(), Some(13));

    // Consume body as bytes
    let body = resp.bytes().await.unwrap();
    assert_eq!(&body[..], b"hello aioduct");
}

#[tokio::test]
async fn response_extensions_round_trip() {
    let (addr, _counter) = h1_server().await;
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::new();

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

    resp.extensions_mut().insert("key");
    let stored = resp.extensions().get::<&str>().copied();
    assert_eq!(stored, Some("key"), "stored value should be retrieved");

    // Body can still be consumed
    let body = resp.text().await.unwrap();
    assert_eq!(body, "hello aioduct");
}

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

    let handler = |_req: hyper::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 = hyper::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();

    // Metadata accessible before consuming the body
    let remote = resp.remote_addr();
    assert!(remote.is_some(), "remote_addr should be set");
    assert_eq!(remote.unwrap().port(), addr.port());
    assert_eq!(resp.version(), http::Version::HTTP_11);

    // Consume body (triggers decompression)
    let text = resp.text().await.unwrap();
    assert_eq!(text, "hello compressed world");
}

#[tokio::test]
async fn response_metadata_on_cached_hit() {
    let attempt = Arc::new(AtomicU32::new(0));
    let attempt_clone = attempt.clone();

    let (addr, _counter) = h1_server_with(move |_req| {
        let attempt = attempt_clone.clone();
        async move {
            attempt.fetch_add(1, Ordering::SeqCst);
            Ok::<_, Infallible>(
                hyper::Response::builder()
                    .header("cache-control", "max-age=3600")
                    .body(Full::new(Bytes::from("cached data")))
                    .unwrap(),
            )
        }
    })
    .await;

    let cache = aioduct::HttpCache::new();
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
        .cache(cache)
        .build()
        .unwrap();

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

    // First request: hits the server, stores in cache
    let resp = client.get(&url).unwrap().send().await.unwrap();
    assert_eq!(resp.version(), http::Version::HTTP_11);
    assert_eq!(resp.text().await.unwrap(), "cached data");
    assert_eq!(attempt.load(Ordering::SeqCst), 1);

    // Second request: cache hit
    let resp = client.get(&url).unwrap().send().await.unwrap();
    assert!(
        resp.remote_addr().is_none(),
        "cached hit should have no remote_addr"
    );
    assert_eq!(
        resp.version(),
        http::Version::HTTP_11,
        "cached hit should preserve version from original response"
    );
    assert_eq!(resp.text().await.unwrap(), "cached data");
    assert_eq!(
        attempt.load(Ordering::SeqCst),
        1,
        "cache should prevent second server hit"
    );
}

// ── Base URL resolution ─────────────────────────────────────────────────────

/// A client base_url lets relative request paths resolve against it. The
/// server echoes back the request target path so we can assert resolution.
#[tokio::test]
async fn base_url_resolves_relative_request_path() {
    let (addr, _counter) = h1_server_with(|req| async move {
        let path = req.uri().path().to_string();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(path))))
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
        .base_url(&format!("http://{addr}/v1/"))
        .unwrap()
        .build()
        .unwrap();

    // Relative reference resolves under the base path.
    let resp = client.get("users").unwrap().send().await.unwrap();
    assert_eq!(resp.status(), http::StatusCode::OK);
    assert_eq!(resp.text().await.unwrap(), "/v1/users");

    // Absolute path replaces the base path.
    let resp = client.get("/health").unwrap().send().await.unwrap();
    assert_eq!(resp.text().await.unwrap(), "/health");
}

/// An absolute request URL overrides the configured base entirely, routing to
/// a different server.
#[tokio::test]
async fn base_url_absolute_request_overrides_base() {
    let (base_addr, base_counter) = h1_server_with(|_req| async move {
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from("base"))))
    })
    .await;
    let (other_addr, _other_counter) = h1_server_with(|_req| async move {
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from("other"))))
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
        .base_url(&format!("http://{base_addr}/v1/"))
        .unwrap()
        .build()
        .unwrap();

    let resp = client
        .get(&format!("http://{other_addr}/x"))
        .unwrap()
        .send()
        .await
        .unwrap();
    assert_eq!(resp.text().await.unwrap(), "other");
    // The base server was never contacted for this request.
    assert_eq!(base_counter.requests(), 0);
}

/// An invalid base URL is rejected at build time.
#[tokio::test]
async fn base_url_invalid_is_rejected() {
    let result = HttpEngineSend::<TokioRuntime, TcpConnector>::builder().base_url("not a url");
    assert!(result.is_err());
}

/// The generic `HttpClient` trait path also honors base_url, matching the
/// concrete engine methods.
#[tokio::test]
async fn base_url_applies_through_httpclient_trait() {
    use aioduct::{HttpClient, RequestBuilderExt};

    let (addr, _counter) = h1_server_with(|req| async move {
        let path = req.uri().path().to_string();
        Ok::<_, Infallible>(Response::new(Full::new(Bytes::from(path))))
    })
    .await;

    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
        .base_url(&format!("http://{addr}/v1/"))
        .unwrap()
        .build()
        .unwrap();

    let resp = HttpClient::request(&client, http::Method::GET, "users")
        .unwrap()
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), http::StatusCode::OK);
    assert_eq!(resp.text().await.unwrap(), "/v1/users");
}

/// With a base_url set, an absolute non-http(s) request override is rejected
/// rather than being dispatched as cleartext HTTP.
#[tokio::test]
async fn base_url_rejects_non_http_absolute_override() {
    let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
        .base_url("https://api.example.com/")
        .unwrap()
        .build()
        .unwrap();

    assert!(client.get("ftp://example.com/path").is_err());
    assert!(client.get("file:///etc/passwd").is_err());
}