apollo-http-client 0.2.0

//! Integration tests for proxy support.

#[macro_use]
mod common;

use std::net::SocketAddr;
use std::sync::{Arc, Mutex};

use apollo_opentelemetry_test::{TelemetryContext, assert_metrics_snapshot};
use axum::Router;
use axum::routing::get;
use bytes::Bytes;
use http::Method;
use http_body_util::{BodyExt as _, Empty};
use hyper::body::Incoming;
use hyper::client::conn::http1 as http1_client;
use hyper::server::conn::http1;
use hyper::service::service_fn;
use hyper_util::rt::TokioIo;
use indoc::indoc;
use tower::BoxError;
use tower::ServiceExt as _;

use common::*;

// ---- Proxy server helpers ---------------------------------------------------

type BoxBody = http_body_util::combinators::BoxBody<Bytes, BoxError>;

fn empty_boxbody() -> BoxBody {
    Empty::<Bytes>::new()
        .map_err(|e: std::convert::Infallible| match e {})
        .boxed()
}

/// Spawns a minimal HTTP proxy that handles both plain HTTP forwarding and
/// CONNECT tunnels. Returns the proxy's listen address and an `Arc<Mutex<_>>`
/// that accumulates any `Proxy-Authorization` headers it receives.
async fn spawn_proxy() -> (SocketAddr, Arc<Mutex<Vec<String>>>) {
    let captured = Arc::new(Mutex::new(Vec::<String>::new()));
    let captured_clone = captured.clone();
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();

    tokio::spawn(async move {
        loop {
            let Ok((stream, _)) = listener.accept().await else {
                break;
            };
            let captured = captured_clone.clone();
            tokio::spawn(async move {
                http1::Builder::new()
                    .preserve_header_case(true)
                    .serve_connection(
                        TokioIo::new(stream),
                        service_fn(move |req| handle_proxy_request(req, captured.clone())),
                    )
                    .with_upgrades()
                    .await
                    .ok();
            });
        }
    });

    (addr, captured)
}

async fn handle_proxy_request(
    req: hyper::Request<Incoming>,
    captured_auth: Arc<Mutex<Vec<String>>>,
) -> Result<hyper::Response<BoxBody>, BoxError> {
    use tokio::net::TcpStream;

    // Capture any Proxy-Authorization header.
    if let Some(auth) = req.headers().get(http::header::PROXY_AUTHORIZATION) {
        captured_auth
            .lock()
            .unwrap()
            .push(auth.to_str().unwrap_or("").to_string());
    }

    if req.method() == Method::CONNECT {
        // CONNECT tunnel: open TCP to the target and relay bytes.
        let authority = req.uri().authority().unwrap().to_string();
        tokio::spawn(async move {
            let Ok(mut target) = TcpStream::connect(&authority).await else {
                return;
            };
            let Ok(upgraded) = hyper::upgrade::on(req).await else {
                return;
            };
            tokio::io::copy_bidirectional(&mut TokioIo::new(upgraded), &mut target)
                .await
                .ok();
        });
        Ok(hyper::Response::builder()
            .status(200)
            .body(empty_boxbody())
            .unwrap())
    } else {
        // Plain HTTP: forward to the target using the absolute-form URI.
        let uri = req.uri().clone();
        let host = uri.host().ok_or("missing host in proxy request")?;
        let port = uri.port_u16().unwrap_or(80);

        let tcp = TcpStream::connect(format!("{host}:{port}")).await?;
        let (mut sender, conn) = http1_client::handshake::<_, Incoming>(TokioIo::new(tcp)).await?;
        tokio::spawn(async move {
            conn.await.expect("plain http proxy connection errored");
        });

        // Strip the host/scheme from the URI so the backend receives a
        // relative-form request.
        let (mut parts, body) = req.into_parts();
        let path_and_query = uri.path_and_query().map(|pq| pq.as_str()).unwrap_or("/");
        parts.uri = path_and_query.parse()?;

        let resp = sender
            .send_request(hyper::Request::from_parts(parts, body))
            .await?;
        Ok(resp.map(|b| b.map_err(BoxError::from).boxed()))
    }
}

/// Spawns a self-signed TLS server that speaks HTTP/1.1 and advertises ALPN
/// `"http/1.1"`. Used to test HTTPS requests through a CONNECT proxy.
/// Clients must set `tls.danger_accept_invalid_certs: true`.
async fn spawn_tls_h1_server(router: Router) -> SocketAddr {
    use hyper_util::service::TowerToHyperService;
    use rcgen::{CertificateParams, KeyPair};
    use rustls::ServerConfig;
    use rustls::crypto::aws_lc_rs as aws_lc_rs_crypto;
    use tokio_rustls::TlsAcceptor;

    let key_pair = KeyPair::generate().unwrap();
    let params = CertificateParams::new(vec!["127.0.0.1".to_string()]).unwrap();
    let cert = params.self_signed(&key_pair).unwrap();
    let cert_der = cert.der().clone();
    let key_der = rustls::pki_types::PrivateKeyDer::Pkcs8(key_pair.serialize_der().into());

    let mut server_config =
        ServerConfig::builder_with_provider(Arc::new(aws_lc_rs_crypto::default_provider()))
            .with_safe_default_protocol_versions()
            .unwrap()
            .with_no_client_auth()
            .with_single_cert(vec![cert_der], key_der)
            .unwrap();
    server_config.alpn_protocols = vec![b"http/1.1".to_vec()];
    let acceptor = TlsAcceptor::from(Arc::new(server_config));

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();

    tokio::spawn(async move {
        loop {
            let Ok((stream, _)) = listener.accept().await else {
                break;
            };
            let acceptor = acceptor.clone();
            let router = router.clone();
            tokio::spawn(async move {
                let Ok(tls_stream) = acceptor.accept(stream).await else {
                    return;
                };
                let svc = TowerToHyperService::new(router);
                http1::Builder::new()
                    .serve_connection(TokioIo::new(tls_stream), svc)
                    .await
                    .ok();
            });
        }
    });

    addr
}

// ---- Tests ------------------------------------------------------------------

/// HTTP requests through a proxy are forwarded to the backend and the response
/// body is delivered to the caller.
#[tokio::test]
async fn proxy_routes_http_request_through_proxy() {
    let backend =
        spawn_server(Router::new().route("/", get(|| async { "hello from backend" }))).await;
    let (proxy_addr, _) = spawn_proxy().await;

    let config = parse_config_with_vars(
        indoc! {r#"
            proxy:
              url: ${env.PROXY_URL}
        "#},
        &[("PROXY_URL", &format!("http://{proxy_addr}"))],
    );
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("http://{backend}/"))
        .body(empty_body())
        .unwrap();

    let resp = new_client(&config).oneshot(req).await.expect("request ok");
    let body = resp.into_body().collect().await.unwrap().to_bytes();
    assert_eq!(body, "hello from backend");
}

/// HTTPS requests are tunnelled through the proxy using a CONNECT tunnel.
/// The response body is delivered to the caller through the encrypted channel.
#[tokio::test]
async fn proxy_routes_https_request_via_connect_tunnel() {
    let backend =
        spawn_tls_h1_server(Router::new().route("/", get(|| async { "hello via tunnel" }))).await;
    let (proxy_addr, _) = spawn_proxy().await;

    let config = parse_config_with_vars(
        indoc! {r#"
            proxy:
              url: ${env.PROXY_URL}
            tls:
              danger_accept_invalid_certs: true
        "#},
        &[("PROXY_URL", &format!("http://{proxy_addr}"))],
    );
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("https://{backend}/"))
        .body(empty_body())
        .unwrap();

    let resp = new_client(&config).oneshot(req).await.expect("request ok");
    let body = resp.into_body().collect().await.unwrap().to_bytes();
    assert_eq!(body, "hello via tunnel");
}

/// When the proxy URL includes credentials, the client sends a
/// `Proxy-Authorization: Basic …` header with every request.
#[tokio::test]
async fn proxy_sends_proxy_authorization_header_when_credentials_present() {
    let backend = spawn_server(Router::new().route("/", get(|| async { "ok" }))).await;
    let (proxy_addr, captured_auth) = spawn_proxy().await;
    // base64("alice:s3cr3t") = "YWxpY2U6czNjcjN0"

    let config = parse_config_with_vars(
        indoc! {r#"
            proxy:
              url: ${env.PROXY_URL}
        "#},
        &[("PROXY_URL", &format!("http://alice:s3cr3t@{proxy_addr}"))],
    );
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("http://{backend}/"))
        .body(empty_body())
        .unwrap();

    new_client(&config).oneshot(req).await.expect("request ok");

    let headers = captured_auth.lock().unwrap();
    assert_eq!(
        headers.len(),
        1,
        "proxy should have received one auth header"
    );
    assert_eq!(headers[0], "Basic YWxpY2U6czNjcjN0");
}

/// When the proxy URL includes credentials and the target is HTTPS, the client
/// sends `Proxy-Authorization` on the CONNECT request (not on the tunnelled
/// application request, which the proxy never sees).
#[tokio::test]
async fn proxy_sends_proxy_authorization_header_on_https_connect_request() {
    let backend = spawn_tls_h1_server(Router::new().route("/", get(|| async { "ok" }))).await;
    let (proxy_addr, captured_auth) = spawn_proxy().await;
    // base64("alice:s3cr3t") = "YWxpY2U6czNjcjN0"

    let config = parse_config_with_vars(
        indoc! {r#"
            proxy:
              url: ${env.PROXY_URL}
            tls:
              danger_accept_invalid_certs: true
        "#},
        &[("PROXY_URL", &format!("http://alice:s3cr3t@{proxy_addr}"))],
    );
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("https://{backend}/"))
        .body(empty_body())
        .unwrap();

    new_client(&config).oneshot(req).await.expect("request ok");

    let headers = captured_auth.lock().unwrap();
    assert_eq!(
        headers.len(),
        1,
        "proxy should have received one auth header on CONNECT"
    );
    assert_eq!(headers[0], "Basic YWxpY2U6czNjcjN0");
}

/// The application request flowing through an HTTPS CONNECT tunnel must not
/// carry `Proxy-Authorization`. That header authenticates the client to the
/// proxy on the CONNECT exchange only; including it on the tunnelled request
/// would leak proxy credentials to the backend, which has no business seeing
/// them. The proxy-sender's HTTPS guard skips the per-request header
/// injection — this test catches a regression in that guard.
#[tokio::test]
async fn proxy_https_tunnelled_request_does_not_leak_proxy_authorization_to_backend() {
    use axum::http::HeaderMap;

    let backend_headers: Arc<Mutex<Option<HeaderMap>>> = Arc::new(Mutex::new(None));
    let captured = backend_headers.clone();
    let backend = spawn_tls_h1_server(Router::new().route(
        "/",
        get(move |headers: HeaderMap| {
            let captured = captured.clone();
            async move {
                *captured.lock().unwrap() = Some(headers);
                "ok"
            }
        }),
    ))
    .await;
    let (proxy_addr, _) = spawn_proxy().await;

    let config = parse_config_with_vars(
        indoc! {r#"
            proxy:
              url: ${env.PROXY_URL}
            tls:
              danger_accept_invalid_certs: true
        "#},
        &[("PROXY_URL", &format!("http://alice:s3cr3t@{proxy_addr}"))],
    );
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("https://{backend}/"))
        .body(empty_body())
        .unwrap();

    new_client(&config).oneshot(req).await.expect("request ok");

    let received = backend_headers
        .lock()
        .unwrap()
        .clone()
        .expect("backend must have received the request");
    assert!(
        !received.contains_key(http::header::PROXY_AUTHORIZATION),
        "backend received Proxy-Authorization on tunnelled request: {received:?}"
    );
}

/// Handler that rejects any request with a fixed status. Extracted to a free
/// function because inlining it as a closure runs into HRTB lifetime issues
/// in the `service_fn` bound.
async fn reject_with_status(
    status: http::StatusCode,
    _req: hyper::Request<Incoming>,
) -> Result<hyper::Response<BoxBody>, std::convert::Infallible> {
    Ok(hyper::Response::builder()
        .status(status)
        .body(empty_boxbody())
        .unwrap())
}

/// Spawns a proxy that rejects every CONNECT request with the given status.
/// Used to verify that a non-200 CONNECT response surfaces as
/// `HttpClientError::ProxyTunnel { status }` rather than a generic transport error.
async fn spawn_rejecting_connect_proxy(status: http::StatusCode) -> SocketAddr {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();

    tokio::spawn(async move {
        loop {
            let Ok((stream, _)) = listener.accept().await else {
                break;
            };
            tokio::spawn(async move {
                http1::Builder::new()
                    .serve_connection(
                        TokioIo::new(stream),
                        service_fn(move |req| reject_with_status(status, req)),
                    )
                    .await
                    .ok();
            });
        }
    });

    addr
}

/// When the proxy responds to a CONNECT request with a non-200 status (e.g. 407
/// Proxy Authentication Required), the client surfaces this as a structured
/// `HttpClientError::ProxyTunnel` error carrying the status code, rather than a
/// generic transport error.
#[tokio::test]
async fn proxy_returns_proxy_tunnel_error_when_connect_rejected() {
    let proxy_addr =
        spawn_rejecting_connect_proxy(http::StatusCode::PROXY_AUTHENTICATION_REQUIRED).await;

    let config = parse_config_with_vars(
        indoc! {r#"
            proxy:
              url: ${env.PROXY_URL}
            tls:
              danger_accept_invalid_certs: true
        "#},
        &[("PROXY_URL", &format!("http://{proxy_addr}"))],
    );
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri("https://example.com/")
        .body(empty_body())
        .unwrap();

    let err = new_client(&config)
        .oneshot(req)
        .await
        .expect_err("should fail");
    match err {
        apollo_http_client::HttpClientError::ProxyTunnel { status } => {
            assert_eq!(status, http::StatusCode::PROXY_AUTHENTICATION_REQUIRED);
        }
        other => panic!("expected ProxyTunnel error, got {other}"),
    }
}

/// When the proxy accepts the TCP connection but never responds, the connect
/// timeout fires and the client returns `HttpClientError::ConnectionTimeout`.
/// This is distinct from "connection refused", which surfaces as a transport
/// error from the OS.
#[tokio::test]
async fn proxy_connect_timeout_fires_when_proxy_unresponsive() {
    let proxy_addr = spawn_stall_server().await;

    let config = parse_config_with_vars(
        indoc! {r#"
            proxy:
              url: ${env.PROXY_URL}
            connect_timeout: 100ms
            tls:
              danger_accept_invalid_certs: true
        "#},
        &[("PROXY_URL", &format!("http://{proxy_addr}"))],
    );
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri("https://example.com/")
        .body(empty_body())
        .unwrap();

    let err = new_client(&config)
        .oneshot(req)
        .await
        .expect_err("should fail");
    assert!(
        matches!(err, apollo_http_client::HttpClientError::ConnectionTimeout),
        "expected ConnectionTimeout, got: {err}"
    );
}

/// When CONNECT succeeds but the backend does not speak TLS, the TLS handshake
/// inside the tunnel fails and the client returns `HttpClientError::Transport`.
/// The proxy itself behaves correctly — the failure is between the client and
/// the backend through the established tunnel — so no `ProxyTunnel` error is
/// expected.
#[tokio::test]
async fn proxy_https_tunnel_fails_when_target_not_tls() {
    // The drop-server accepts TCP and immediately closes; the client sees EOF
    // when it expects a TLS ServerHello.
    let backend = spawn_drop_server().await;
    let (proxy_addr, _) = spawn_proxy().await;

    let config = parse_config_with_vars(
        indoc! {r#"
            proxy:
              url: ${env.PROXY_URL}
            tls:
              danger_accept_invalid_certs: true
        "#},
        &[("PROXY_URL", &format!("http://{proxy_addr}"))],
    );
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("https://{backend}/"))
        .body(empty_body())
        .unwrap();

    let err = new_client(&config)
        .oneshot(req)
        .await
        .expect_err("should fail");
    assert!(
        matches!(err, apollo_http_client::HttpClientError::Transport { .. }),
        "expected Transport error from failed tunnel TLS handshake, got: {err}"
    );
}

/// With `protocol: http2`, an HTTPS target is reached through the proxy by
/// opening a CONNECT tunnel, performing TLS, and running the HTTP/2 handshake
/// inside the encrypted channel. The backend advertises only `h2` ALPN, so a
/// successful response proves the H2 path through the proxy works end to end.
#[tokio::test]
async fn proxy_routes_h2_request_via_connect_tunnel() {
    let backend =
        spawn_tls_h2_server(Router::new().route("/", get(|| async { "hello via h2 tunnel" })))
            .await;
    let (proxy_addr, _) = spawn_proxy().await;

    let config = parse_config_with_vars(
        indoc! {r#"
            protocol: http2
            proxy:
              url: ${env.PROXY_URL}
            tls:
              danger_accept_invalid_certs: true
        "#},
        &[("PROXY_URL", &format!("http://{proxy_addr}"))],
    );
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("https://{backend}/"))
        .body(empty_body())
        .unwrap();

    let resp = new_client(&config).oneshot(req).await.expect("request ok");
    let body = resp.into_body().collect().await.unwrap().to_bytes();
    assert_eq!(body, "hello via h2 tunnel");
}

/// With `protocol: alpn`, the client offers both `h2` and `http/1.1` during the
/// TLS handshake inside the CONNECT tunnel. When the server advertises only
/// `h2`, a successful response proves the ALPN negotiation upgrade path
/// through the proxy chose HTTP/2 — falling back to HTTP/1.1 would fail the
/// handshake because the server does not speak it.
#[tokio::test]
async fn proxy_alpn_negotiates_h2_via_connect_tunnel() {
    let backend =
        spawn_tls_h2_server(Router::new().route("/", get(|| async { "hello via alpn-h2 tunnel" })))
            .await;
    let (proxy_addr, _) = spawn_proxy().await;

    let config = parse_config_with_vars(
        indoc! {r#"
            protocol: alpn
            proxy:
              url: ${env.PROXY_URL}
            tls:
              danger_accept_invalid_certs: true
        "#},
        &[("PROXY_URL", &format!("http://{proxy_addr}"))],
    );
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("https://{backend}/"))
        .body(empty_body())
        .unwrap();

    let resp = new_client(&config).oneshot(req).await.expect("request ok");
    let body = resp.into_body().collect().await.unwrap().to_bytes();
    assert_eq!(body, "hello via alpn-h2 tunnel");
}

/// When the CONNECT response succeeds but the backend stalls during the tunnel
/// TLS handshake, the connect timeout fires. This proves `connect_timeout`
/// extends past the CONNECT exchange to cover the TLS handshake inside the
/// tunnel — distinct from the existing test, which only stalls the proxy at
/// the CONNECT request itself.
#[tokio::test]
async fn proxy_connect_timeout_fires_during_tunnel_tls_handshake() {
    // The backend completes TCP but never sends a TLS ServerHello, so the
    // tunnel TLS handshake hangs after CONNECT succeeds.
    let backend = spawn_stall_server().await;
    let (proxy_addr, _) = spawn_proxy().await;

    let config = parse_config_with_vars(
        indoc! {r#"
            proxy:
              url: ${env.PROXY_URL}
            connect_timeout: 200ms
            tls:
              danger_accept_invalid_certs: true
        "#},
        &[("PROXY_URL", &format!("http://{proxy_addr}"))],
    );
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("https://{backend}/"))
        .body(empty_body())
        .unwrap();

    let err = new_client(&config)
        .oneshot(req)
        .await
        .expect_err("should fail");
    assert!(
        matches!(err, apollo_http_client::HttpClientError::ConnectionTimeout),
        "expected ConnectionTimeout from stalled tunnel TLS handshake, got: {err}"
    );
}

/// A successful HTTP request through a proxy records the expected connection-
/// state transitions and uses the **target** server address for `server.address`,
/// while `network.peer.*` exposes the **proxy** as the actual TCP peer.
///
/// All three states (connecting, active, idle) appear in the snapshot. The
/// connecting counter increments while the proxy connector establishes TCP,
/// then drops to zero. The active counter increments during the request and
/// drops to zero when it completes. The idle counter increments when the
/// connection returns to the pool, then drops to zero when the client is
/// dropped at the end of the test. The connection-duration metric is excluded
/// via `integration_context` because its wall-clock values are non-deterministic.
#[tokio::test]
async fn proxy_request_records_connection_state_metrics() {
    let ctx = integration_context();

    let backend = spawn_server(Router::new().route("/", get(|| async { "ok" }))).await;
    let (proxy_addr, _) = spawn_proxy().await;

    let config = parse_config_with_vars(
        indoc! {r#"
            proxy:
              url: ${env.PROXY_URL}
        "#},
        &[("PROXY_URL", &format!("http://{proxy_addr}"))],
    );
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri(format!("http://{backend}/"))
        .body(empty_body())
        .unwrap();

    let resp = new_client(&config).oneshot(req).await.expect("request ok");
    // Consume body so the connection returns to the pool as idle.
    resp.into_body().collect().await.unwrap();

    snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: GET
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.open_connections
      description: Number of open connections in the HTTP client pool
      unit: "{connection}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.connection.state: active
              network.peer.address: 127.0.0.1
              network.peer.port: "<peer_port>"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: connecting
              network.peer.address: 127.0.0.1
              network.peer.port: "<peer_port>"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
          - attributes:
              http.connection.state: idle
              network.peer.address: 127.0.0.1
              network.peer.port: "<peer_port>"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              http.request.method: GET
              http.response.status_code: "200"
              network.protocol.version: "1.1"
              server.address: 127.0.0.1
              server.port: "<port>"
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}

/// When the proxy is unreachable, the in-flight counters return to zero and
/// the request duration records `error.type: _OTHER`. The `connecting` state
/// for `open_connections` is incremented and decremented by the
/// `_connecting` track guard, so its final value must also be zero — this is
/// the test that catches a regression in the guard's drop behaviour.
///
/// `network.peer.*` is set to the proxy address on the failure path too, so
/// operators can tell from telemetry whether a connect failure was proxy-side
/// or target-side.
#[tokio::test]
async fn proxy_unreachable_records_error_and_resets_counters() {
    let ctx = TelemetryContext::new();

    let config = parse_config(indoc! {r#"
        proxy:
          url: http://127.0.0.1:1
        connect_timeout: 1s
    "#});
    let req = http::Request::builder()
        .method(http::Method::GET)
        .uri("http://example.com/")
        .body(empty_body())
        .unwrap();

    let err = new_client(&config)
        .oneshot(req)
        .await
        .expect_err("should fail");
    assert!(
        matches!(
            err,
            apollo_http_client::HttpClientError::Transport { .. }
                | apollo_http_client::HttpClientError::ConnectionTimeout
        ),
        "unexpected error: {err}"
    );

    // No `connection.duration` is asserted here: that metric is recorded by the
    // H1Connection/H2Connection drop guard, which is only constructed once the
    // HTTP handshake completes. A failed TCP connect never reaches that point.
    assert_metrics_snapshot!(ctx, @r#"
    - name: http.client.active_requests
      description: Number of HTTP requests currently in flight
      unit: "{request}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.request.method: GET
              server.address: example.com
              server.port: "80"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.open_connections
      description: Number of open connections in the HTTP client pool
      unit: "{connection}"
      data:
        type: Sum
        data_points:
          - attributes:
              http.connection.state: connecting
              network.peer.address: 127.0.0.1
              network.peer.port: "1"
              server.address: example.com
              server.port: "80"
            value: 0
        is_monotonic: false
        temporality: Cumulative
    - name: http.client.request.duration
      description: Duration of HTTP client requests
      unit: s
      data:
        type: Histogram
        data_points:
          - attributes:
              error.type: _OTHER
              http.request.method: GET
              network.protocol.version: "1.1"
              server.address: example.com
              server.port: "80"
            count: 1
            sum: 0
            min: 0
            max: 0
            bounds:
              - 0.005
              - 0.01
              - 0.025
              - 0.05
              - 0.075
              - 0.1
              - 0.25
              - 0.5
              - 0.75
              - 1
              - 2.5
              - 5
              - 7.5
              - 10
            bucket_counts:
              - 1
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
              - 0
        temporality: Cumulative
    "#);
}