apollo-http-client 0.3.0

//! Shared test infrastructure for `HttpClient` integration tests.
// Not every helper is used in every test binary — suppress dead-code warnings
// for this shared module.
#![allow(dead_code, unused_macros)]
//!
//! Import via `#[macro_use] mod common;` in each integration test file.
//! Helper functions are brought into scope with `use common::*;`.
//!
//! ## Non-determinism mitigation
//!
//! - **`server.port`** — OS-assigned port is normalized to `"<port>"` by the
//!   `snapshot!` macro via an insta filter.
//! - **`http.client.connection.duration`** — wall-clock `sum`/`min`/`max` vary per
//!   run; the metric is dropped via a metric view in [`integration_context`] and
//!   tested separately in the H1 connection-state tests.

use std::net::SocketAddr;

use apollo_http_client::{HttpBody, HttpClient, HttpClientConfig};
use apollo_opentelemetry::metrics::Clock;
use apollo_opentelemetry_test::TelemetryContext;
use axum::Router;
use bytes::Bytes;
use http_body_util::{BodyExt as _, Full};
use indoc::indoc;
use opentelemetry_sdk::metrics::{Aggregation, Instrument, StreamBuilder};

/// Returns a [`TelemetryContext`] that drops `http.client.connection.duration`.
///
/// The metric's wall-clock `sum`/`min`/`max` are non-deterministic; dropping it
/// from the snapshot context keeps metric snapshots stable across runs.
pub fn integration_context() -> TelemetryContext {
    TelemetryContext::builder()
        .with_view(|instrument: &Instrument| {
            if instrument.name() == "http.client.connection.duration" {
                Some(
                    StreamBuilder::default()
                        .with_aggregation(Aggregation::Drop)
                        .build()
                        .unwrap(),
                )
            } else {
                None
            }
        })
        .build()
}

/// Wraps `assert_metrics_snapshot!` with insta filters that normalize the
/// OS-assigned `server.port` and `network.peer.port` to placeholder values
/// before comparing or storing snapshots. `network.peer.port` appears on
/// proxied connections and is the OS-assigned port of the proxy.
macro_rules! snapshot {
    ($ctx:expr, @$snapshot:literal) => {{
        insta::with_settings!({
            filters => [
                (r#"server\.port: "\d+""#, r#"server.port: "<port>""#),
                (r#"network\.peer\.port: "\d+""#, r#"network.peer.port: "<peer_port>""#),
            ]
        }, {
            apollo_opentelemetry_test::assert_metrics_snapshot!($ctx, @$snapshot);
        });
    }};
}

/// Wraps `assert_spans_snapshot!` normalizing OS-assigned port in `server.port`
/// and `url.full`.
macro_rules! span_snapshot {
    ($ctx:expr, @$snapshot:literal) => {{
        insta::with_settings!({
            filters => [
                (r#"server\.port: "\d+""#, r#"server.port: "<port>""#),
                (r#"url\.full: "http://127\.0\.0\.1:\d+/"#, r#"url.full: "http://127.0.0.1:<port>/"#),
            ]
        }, {
            apollo_opentelemetry_test::assert_spans_snapshot!($ctx, @$snapshot);
        });
    }};
}

pub async fn spawn_server(router: Router) -> SocketAddr {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        axum::serve(listener, router).await.unwrap();
    });
    addr
}

/// Owns a unix-socket axum server bound to a private temporary directory. Dropping
/// the guard aborts the server task and removes the directory.
#[cfg(unix)]
pub struct UnixTestServer {
    socket_path: String,
    _tmp: tempfile::TempDir,
    task: tokio::task::JoinHandle<()>,
}

#[cfg(unix)]
impl UnixTestServer {
    /// Returns the socket path.
    pub fn path(&self) -> &str {
        &self.socket_path
    }
}

#[cfg(unix)]
impl Drop for UnixTestServer {
    fn drop(&mut self) {
        self.task.abort();
    }
}

/// Spawns an axum server on a Unix socket in a private temporary directory and
/// returns the [`UnixTestServer`] guard that owns it.
#[cfg(unix)]
pub async fn spawn_unix_server(router: Router) -> UnixTestServer {
    let tmp = tempfile::tempdir().unwrap();
    let socket_path = tmp
        .path()
        .join("test.sock")
        .into_os_string()
        .into_string()
        .expect("tempdir path is UTF-8");
    let listener = tokio::net::UnixListener::bind(&socket_path).unwrap();
    let task = tokio::spawn(async move {
        axum::serve(listener, router).await.unwrap();
    });
    UnixTestServer {
        socket_path,
        _tmp: tmp,
        task,
    }
}

/// Spawns a raw TCP listener that immediately closes each accepted connection.
///
/// Simulates a server that drops the connection before reading the request —
/// used in both H1 chaos tests and H2 error-path tests.
pub async fn spawn_drop_server() -> SocketAddr {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        while let Ok((socket, _)) = listener.accept().await {
            drop(socket);
        }
    });
    addr
}

/// Spawns a raw TCP listener that accepts connections then holds them open
/// without reading or writing, so the client stalls. Used to test handshake
/// timeouts where a server completes the TCP three-way handshake but never
/// responds.
pub async fn spawn_stall_server() -> SocketAddr {
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    tokio::spawn(async move {
        while let Ok((socket, _)) = listener.accept().await {
            tokio::spawn(async move {
                tokio::time::sleep(std::time::Duration::from_secs(60)).await;
                drop(socket);
            });
        }
    });
    addr
}

/// Spawns a self-signed TLS server that speaks HTTP/2 and advertises ALPN
/// `"h2"`. Because the only advertised protocol is `h2`, a successful request
/// proves that the client negotiated HTTP/2 — falling back to HTTP/1.1 would
/// fail the ALPN match. Clients must set `tls.danger_accept_invalid_certs: true`.
pub async fn spawn_tls_h2_server(router: Router) -> SocketAddr {
    use std::sync::Arc as SyncArc;

    use hyper::server::conn::http2;
    use hyper_util::rt::{TokioExecutor, TokioIo};
    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(SyncArc::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"h2".to_vec()];
    let acceptor = TlsAcceptor::from(SyncArc::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);
                http2::Builder::new(TokioExecutor::new())
                    .serve_connection(TokioIo::new(tls_stream), svc)
                    .await
                    .ok();
            });
        }
    });

    addr
}

/// Creates a client using a mock clock so `request.duration` records 0 elapsed
/// (the mock clock is never advanced). This makes duration snapshot values
/// deterministic across all test runs.
pub fn new_client(config: &HttpClientConfig) -> HttpClient {
    let (clock, _mock) = Clock::mock();
    HttpClient::with_clock(config, clock).expect("valid config")
}

pub fn parse_config(yaml: &str) -> HttpClientConfig {
    apollo_configuration::parse_yaml(yaml, &Default::default()).expect("valid config")
}

/// Like [`parse_config`], but supplies values for `${env.…}` references in the
/// YAML through an in-process variable map. Lets tests write configs that
/// substitute runtime values (proxy addresses, port numbers, etc.) without
/// embedding them in the YAML via `format!`, and exercises the same expansion
/// path users hit when they template their configuration.
pub fn parse_config_with_vars(yaml: &str, vars: &[(&str, &str)]) -> HttpClientConfig {
    use std::collections::HashMap;

    use apollo_configuration::ParseYamlOptions;
    use apollo_configuration::expansion::MapVariables;

    let variables: HashMap<String, String> = vars
        .iter()
        .map(|(k, v)| ((*k).to_string(), (*v).to_string()))
        .collect();
    apollo_configuration::parse_yaml(
        yaml,
        &ParseYamlOptions::default().variables(MapVariables(variables)),
    )
    .expect("valid config")
}

pub fn default_config() -> HttpClientConfig {
    parse_config("{}")
}

pub fn config_with_body_size() -> HttpClientConfig {
    parse_config(indoc! {"
        telemetry:
          metrics:
            request_body_size: true
            response_body_size: true
            url_scheme: true
    "})
}

pub fn empty_body() -> HttpBody {
    Full::new(Bytes::new())
        .map_err(|e: std::convert::Infallible| match e {})
        .boxed()
}

pub fn bytes_body(b: Bytes) -> HttpBody {
    Full::new(b)
        .map_err(|e: std::convert::Infallible| match e {})
        .boxed()
}