apollo-http-client 0.3.0

//! Integration tests for `unix://` URI routing through [`apollo_http_client::HttpClient`].

#[macro_use]
mod common;

#[cfg(unix)]
mod unix_tests {
    use apollo_http_client::{HttpClientError, unix_uri};
    use apollo_opentelemetry_test::{TelemetryContext, assert_metric, assert_span};
    use axum::Router;
    use axum::routing::get;
    use bytes::Bytes;
    use http_body_util::BodyExt as _;
    use tower::ServiceExt as _;

    use super::common::*;

    /// A `unix://` request through `HttpClient` reaches the socket and records
    /// `http.client.request.duration` with `server.address` set to the decoded
    /// path and no `server.port`.
    #[tokio::test]
    async fn unix_request_routes_through_unix_connector() {
        let ctx = integration_context();
        let server =
            spawn_unix_server(Router::new().route("/", get(|| async { "hello unix" }))).await;

        let uri = unix_uri(server.path())
            .unwrap()
            .path_and_query("/")
            .build()
            .unwrap();
        let req = http::Request::builder()
            .method(http::Method::GET)
            .uri(uri)
            .body(empty_body())
            .unwrap();

        let resp = new_client(&default_config())
            .oneshot(req)
            .await
            .expect("request ok");
        assert_eq!(resp.status(), http::StatusCode::OK);
        let body = resp.into_body().collect().await.unwrap().to_bytes();
        assert_eq!(&body[..], b"hello unix");

        assert_metric!(
            ctx,
            "http.client.request.duration",
            "server.address" = server.path(),
            !"server.port",
        );

        // The span follows the same shape — decoded path in `server.address`,
        // no `server.port`.
        assert_span!(ctx, "GET", "server.address" = server.path(), !"server.port",);
    }

    /// One `HttpClient` instance dispatches `http://` and `unix://` requests
    /// through their respective transports.
    #[tokio::test]
    async fn http_and_unix_coexist_on_one_client() {
        let unix_server =
            spawn_unix_server(Router::new().route("/", get(|| async { "from unix" }))).await;
        let tcp_addr = spawn_server(Router::new().route("/", get(|| async { "from tcp" }))).await;

        let client = new_client(&default_config());

        let unix_request_uri = unix_uri(unix_server.path())
            .unwrap()
            .path_and_query("/")
            .build()
            .unwrap();
        let unix_req = http::Request::builder()
            .method(http::Method::GET)
            .uri(unix_request_uri)
            .body(empty_body())
            .unwrap();
        let unix_resp = client.clone().oneshot(unix_req).await.expect("unix ok");
        assert_eq!(
            unix_resp.into_body().collect().await.unwrap().to_bytes(),
            Bytes::from_static(b"from unix"),
        );

        let tcp_req = http::Request::builder()
            .method(http::Method::GET)
            .uri(format!("http://{tcp_addr}/"))
            .body(empty_body())
            .unwrap();
        let tcp_resp = client.clone().oneshot(tcp_req).await.expect("tcp ok");
        assert_eq!(
            tcp_resp.into_body().collect().await.unwrap().to_bytes(),
            Bytes::from_static(b"from tcp"),
        );
    }

    /// With `protocol: http2`, a `unix://` request runs over prior-knowledge HTTP/2
    /// on the socket; `resp.version()` pins which sender path was taken.
    #[tokio::test]
    async fn http2_unix_request_negotiates_h2() {
        let ctx = integration_context();
        let server = spawn_unix_server(Router::new().route("/", get(|| async { "h2" }))).await;

        let client = new_client(&parse_config("protocol: http2"));
        let uri = unix_uri(server.path())
            .unwrap()
            .path_and_query("/")
            .build()
            .unwrap();
        let req = http::Request::builder()
            .method(http::Method::GET)
            .uri(uri)
            .body(empty_body())
            .unwrap();

        let resp = client.oneshot(req).await.expect("request ok");
        assert_eq!(resp.version(), http::Version::HTTP_2);
        let body = resp.into_body().collect().await.unwrap().to_bytes();
        assert_eq!(&body[..], b"h2");

        assert_metric!(
            ctx,
            "http.client.request.duration",
            "network.protocol.version" = "2",
            "server.address" = server.path(),
            !"server.port",
        );
    }

    /// With the `telemetry.metrics.url_scheme` flag on, a `unix://` request
    /// records `url.scheme = "unix"` on its metrics — the opt-in attribute
    /// flows through the same code path as TCP, just with a different value.
    #[tokio::test]
    async fn unix_request_records_url_scheme_when_enabled() {
        let ctx = integration_context();
        let server = spawn_unix_server(Router::new().route("/", get(|| async { "ok" }))).await;

        let config = parse_config("telemetry:\n  metrics:\n    url_scheme: true\n");

        let uri = unix_uri(server.path())
            .unwrap()
            .path_and_query("/")
            .build()
            .unwrap();
        let req = http::Request::builder()
            .method(http::Method::GET)
            .uri(uri)
            .body(empty_body())
            .unwrap();

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

        assert_metric!(
            ctx,
            "http.client.request.duration",
            "url.scheme" = "unix",
            "server.address" = server.path(),
        );
    }

    /// `protocol: alpn` falls back to HTTP/1.1 over UDS: there is no TLS to
    /// negotiate ALPN through.
    #[tokio::test]
    async fn alpn_unix_falls_back_to_http1() {
        let ctx = integration_context();
        let server = spawn_unix_server(Router::new().route("/", get(|| async { "h1" }))).await;

        let client = new_client(&parse_config("protocol: alpn"));
        let uri = unix_uri(server.path())
            .unwrap()
            .path_and_query("/")
            .build()
            .unwrap();
        let req = http::Request::builder()
            .method(http::Method::GET)
            .uri(uri)
            .body(empty_body())
            .unwrap();

        let resp = client.oneshot(req).await.expect("request ok");
        assert_eq!(resp.version(), http::Version::HTTP_11);
        let body = resp.into_body().collect().await.unwrap().to_bytes();
        assert_eq!(&body[..], b"h1");

        assert_metric!(
            ctx,
            "http.client.request.duration",
            "network.protocol.version" = "1.1",
            "server.address" = server.path(),
            !"server.port",
        );
    }

    /// A `unix://` URI with a non-hex authority fails dispatch with
    /// `InvalidUnixSocketPath`; the metric records the placeholder
    /// `<invalid socket path>` rather than the raw hex.
    #[tokio::test]
    async fn unix_invalid_hex_authority_returns_invalid_unix_socket_path() {
        let ctx = TelemetryContext::new();

        // `http::Uri` accepts this as a reg-name authority, but `hex::decode`
        // rejects the non-hex characters.
        let uri = http::Uri::builder()
            .scheme("unix")
            .authority("not-hex-zz")
            .path_and_query("/")
            .build()
            .expect("uri builds");
        let req = http::Request::builder()
            .method(http::Method::GET)
            .uri(uri)
            .body(empty_body())
            .unwrap();

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

        assert_metric!(
            ctx,
            "http.client.request.duration",
            "server.address" = "<invalid socket path>"
        );
    }

    /// Proxy configuration applies only to TCP/TLS. Point the proxy at a port
    /// nothing listens on; the `unix://` request still succeeds because it
    /// bypasses the proxy path entirely.
    #[tokio::test]
    async fn unix_request_bypasses_proxy_configuration() {
        let server =
            spawn_unix_server(Router::new().route("/", get(|| async { "no proxy here" }))).await;

        let config = parse_config("proxy:\n  url: http://127.0.0.1:1\n");

        let uri = unix_uri(server.path())
            .unwrap()
            .path_and_query("/")
            .build()
            .unwrap();
        let req = http::Request::builder()
            .method(http::Method::GET)
            .uri(uri)
            .body(empty_body())
            .unwrap();

        let resp = new_client(&config).oneshot(req).await.expect("request ok");
        assert_eq!(resp.status(), http::StatusCode::OK);
        let body = resp.into_body().collect().await.unwrap().to_bytes();
        assert_eq!(&body[..], b"no proxy here");
    }

    /// TCP and Unix connection pools on the same `HttpClient` don't share state.
    /// One request to each transport leaves two distinct `idle` entries on
    /// `http.client.open_connections` keyed by `server.address`.
    #[tokio::test]
    async fn unix_and_tcp_pools_are_independent() {
        let ctx = integration_context();
        let unix_server = spawn_unix_server(Router::new().route("/", get(|| async { "u" }))).await;
        let tcp_addr = spawn_server(Router::new().route("/", get(|| async { "t" }))).await;

        let client = new_client(&default_config());
        let unix_request_uri = unix_uri(unix_server.path())
            .unwrap()
            .path_and_query("/")
            .build()
            .unwrap();
        let unix_req = http::Request::builder()
            .method(http::Method::GET)
            .uri(unix_request_uri)
            .body(empty_body())
            .unwrap();
        let _ = client
            .clone()
            .oneshot(unix_req)
            .await
            .expect("unix ok")
            .into_body()
            .collect()
            .await
            .unwrap();

        let tcp_req = http::Request::builder()
            .method(http::Method::GET)
            .uri(format!("http://{tcp_addr}/"))
            .body(empty_body())
            .unwrap();
        let _ = client
            .clone()
            .oneshot(tcp_req)
            .await
            .expect("tcp ok")
            .into_body()
            .collect()
            .await
            .unwrap();

        assert_metric!(
            ctx,
            "http.client.open_connections",
            value: 1,
            "server.address" = unix_server.path(),
            "http.connection.state" = "idle"
        );
        assert_metric!(
            ctx,
            "http.client.open_connections",
            value: 1,
            "server.address" = "127.0.0.1",
            "http.connection.state" = "idle"
        );
    }
}

#[cfg(not(unix))]
mod non_unix_tests {
    use apollo_http_client::HttpClientError;
    use apollo_opentelemetry_test::{TelemetryContext, assert_metric};
    use http_body_util::BodyExt as _;
    use tower::ServiceExt as _;

    use super::common::*;

    /// On non-Unix, `HttpClient::new` builds normally; a subsequent `unix://`
    /// request fails at dispatch with `UnsupportedScheme`. The metric records
    /// the `<invalid socket path>` placeholder so the raw hex authority can't
    /// leak through `server.address`.
    #[tokio::test]
    async fn unix_request_returns_unsupported_scheme() {
        let ctx = TelemetryContext::new();
        let client = new_client(&default_config());

        // Build the `unix://` URI by hand: `unix_uri` is `#[cfg(unix)]`, so it isn't
        // exported here. The authority is the hex-encoded socket path.
        let uri = http::Uri::builder()
            .scheme("unix")
            .authority(hex::encode("/tmp/test.sock"))
            .path_and_query("/")
            .build()
            .expect("uri builds");
        let req = http::Request::builder()
            .method(http::Method::GET)
            .uri(uri)
            .body(empty_body())
            .unwrap();

        let err = client.oneshot(req).await.expect_err("should fail");
        assert!(
            matches!(&err, HttpClientError::UnsupportedScheme { scheme } if scheme == "unix"),
            "expected UnsupportedScheme {{ scheme: \"unix\" }}, got {err:?}",
        );

        assert_metric!(
            ctx,
            "http.client.request.duration",
            "server.address" = "<invalid socket path>",
            !"server.port",
        );
    }
}