#![cfg(feature = "tokio")]
use std::time::Duration;
use aioduct::HttpEngineSend;
use aioduct::runtime::TokioRuntime;
use aioduct::runtime::tokio_rt::TcpConnector;
fn h1_client() -> HttpEngineSend<TokioRuntime, TcpConnector> {
HttpEngineSend::builder()
.pool_idle_timeout(Duration::from_secs(60))
.timeout(Duration::from_secs(5))
.build()
.unwrap()
}
fn h2_client() -> HttpEngineSend<TokioRuntime, TcpConnector> {
HttpEngineSend::builder()
.pool_idle_timeout(Duration::from_secs(60))
.timeout(Duration::from_secs(5))
.build()
.unwrap()
}
fn valid_forward_request<B>(mut request: http::Request<B>) -> http::Request<B> {
if request.version() == http::Version::HTTP_11
&& !request.headers().contains_key(http::header::HOST)
{
request.headers_mut().insert(
http::header::HOST,
http::HeaderValue::from_static("downstream.test"),
);
}
request
}
#[tokio::test]
async fn h1_chunked_transfer_encoding() {
let addr = aioduct_test_server::raw::raw_server(|_req| async {
b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n\
5\r\nhello\r\n\
6\r\n world\r\n\
0\r\n\r\n"
.to_vec()
})
.await;
let client = h1_client();
let resp = client
.get(&format!("http://{addr}/"))
.unwrap()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body = resp.text().await.unwrap();
assert_eq!(
body, "hello world",
"chunked body should be reassembled correctly"
);
}
#[tokio::test]
async fn h1_content_length_body() {
let payload = "exact 42 bytes of payload for this test!!";
assert_eq!(payload.len(), 41);
let addr = aioduct_test_server::raw::raw_server(move |_req| async move {
let body = "exact 42 bytes of payload for this test!!";
format!(
"HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{}",
body.len(),
body
)
.into_bytes()
})
.await;
let client = h1_client();
let resp = client
.get(&format!("http://{addr}/"))
.unwrap()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body = resp.text().await.unwrap();
assert_eq!(body.len(), 41);
assert_eq!(body, "exact 42 bytes of payload for this test!!");
}
#[tokio::test]
async fn h1_head_request_no_body() {
let (addr, _counter) = aioduct_test_server::h1::h1_server_with(|req| async move {
if req.method() == http::Method::HEAD {
let resp = http::Response::builder()
.header("Content-Length", "1000")
.body(http_body_util::Full::new(bytes::Bytes::new()))
.unwrap();
Ok::<_, std::convert::Infallible>(resp)
} else {
let resp = http::Response::builder()
.body(http_body_util::Full::new(bytes::Bytes::from(
"should not see this",
)))
.unwrap();
Ok(resp)
}
})
.await;
let client = h1_client();
let url = format!("http://{addr}/");
let resp = client
.request(http::Method::HEAD, &url)
.unwrap()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
assert_eq!(
resp.headers()
.get("content-length")
.map(|v| v.to_str().unwrap()),
Some("1000"),
"HEAD response should include Content-Length header"
);
let body = resp.bytes().await.unwrap();
assert!(
body.is_empty(),
"HEAD response body must be empty, got {} bytes",
body.len()
);
}
#[tokio::test]
async fn h1_keep_alive_header_respected() {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let conn_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let conn_count2 = conn_count.clone();
tokio::spawn(async move {
loop {
let (mut stream, _) = listener.accept().await.unwrap();
conn_count2.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
tokio::spawn(async move {
let mut buf = [0u8; 4096];
loop {
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") {
let resp =
b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: close\r\n\r\nok";
let _ = stream.write_all(resp).await;
let _ = stream.flush().await;
let _ = stream.shutdown().await;
return;
}
}
});
}
});
let client = h1_client();
let url = format!("http://{addr}/");
let resp = client.get(&url).unwrap().send().await.unwrap();
assert_eq!(resp.status(), 200);
let _ = resp.text().await.unwrap();
tokio::time::sleep(Duration::from_millis(50)).await;
let resp = client.get(&url).unwrap().send().await.unwrap();
assert_eq!(resp.status(), 200);
let _ = resp.text().await.unwrap();
let conns = conn_count.load(std::sync::atomic::Ordering::SeqCst);
assert_eq!(
conns, 2,
"Connection: close should prevent reuse; expected 2 connections, got {conns}"
);
}
#[tokio::test]
async fn h1_keep_alive_reuses_connection() {
let (addr, counter) = aioduct_test_server::h1::h1_server().await;
let client = h1_client();
let url = format!("http://{addr}/");
for _ in 0..5 {
let resp = client.get(&url).unwrap().send().await.unwrap();
assert_eq!(resp.status(), 200);
let _ = resp.text().await.unwrap();
}
assert_eq!(
counter.connections(),
1,
"keep-alive should reuse single connection for 5 sequential requests"
);
}
#[tokio::test]
async fn h1_content_length_mismatch_short() {
let addr = aioduct_test_server::raw::raw_server(|_req| async {
b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\nhello".to_vec()
})
.await;
let client = h1_client();
let resp = client
.get(&format!("http://{addr}/"))
.unwrap()
.send()
.await
.unwrap();
let result = resp.bytes().await;
assert!(
result.is_err(),
"reading body with content-length mismatch should error"
);
}
#[tokio::test]
async fn h1_read_until_close() {
let addr = aioduct_test_server::raw::raw_server(|_req| async {
b"HTTP/1.1 200 OK\r\nConnection: close\r\n\r\nread until EOF".to_vec()
})
.await;
let client = h1_client();
let resp = client
.get(&format!("http://{addr}/"))
.unwrap()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body = resp.text().await.unwrap();
assert_eq!(body, "read until EOF");
}
#[tokio::test]
async fn h2_prior_knowledge_cleartext() {
let (addr, counter) = aioduct_test_server::h2::h2_server().await;
let client = h2_client();
let url = format!("http://{addr}/");
let resp = client
.get(&url)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
assert_eq!(
resp.version(),
http::Version::HTTP_2,
"response should use HTTP/2"
);
let body = resp.text().await.unwrap();
assert_eq!(body, "hello aioduct");
assert_eq!(counter.connections(), 1);
assert_eq!(counter.requests(), 1);
}
#[tokio::test]
async fn h2_goaway_graceful_in_flight() {
let (addr, counter) = aioduct_test_server::h2::h2_goaway_after(1).await;
let client = h2_client();
let url = format!("http://{addr}/");
let resp = client
.get(&url)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body = resp.text().await.unwrap();
assert_eq!(body, "ok");
tokio::time::sleep(Duration::from_millis(100)).await;
let resp = client
.get(&url)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body = resp.text().await.unwrap();
assert_eq!(body, "ok");
assert!(
counter.connections() >= 2,
"expected >= 2 connections after GOAWAY, got {}",
counter.connections()
);
}
#[tokio::test]
async fn h2_stream_count_sequential() {
let (addr, counter) = aioduct_test_server::h2::h2_server().await;
let client = h2_client();
let url = format!("http://{addr}/");
for i in 0..10 {
let resp = client
.get(&url)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200, "request {i} should succeed");
let _ = resp.text().await.unwrap();
}
assert_eq!(
counter.connections(),
1,
"10 sequential H2 requests should multiplex on 1 connection"
);
assert_eq!(counter.requests(), 10);
}
#[tokio::test]
async fn h2_stream_count_concurrent() {
let (addr, counter) = aioduct_test_server::h2::h2_server().await;
let client = h2_client();
let url = format!("http://{addr}/");
let resp = client
.get(&url)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let _ = resp.text().await.unwrap();
let mut handles = Vec::new();
for _ in 0..10 {
let c = client.clone();
let u = url.clone();
handles.push(tokio::spawn(async move {
let resp = c
.get(&u)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let _ = resp.text().await.unwrap();
}));
}
for h in handles {
h.await.unwrap();
}
assert_eq!(counter.requests(), 11);
}
#[tokio::test]
async fn h2_goaway_immediate_still_responds() {
let (addr, counter) = aioduct_test_server::h2::h2_goaway_immediate().await;
let client = h2_client();
let url = format!("http://{addr}/");
let resp = client
.get(&url)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body = resp.text().await.unwrap();
assert_eq!(body, "ok");
assert_eq!(counter.requests(), 1);
}
#[tokio::test]
async fn h2_prior_knowledge_against_h1_server_fails() {
let (addr, _counter) = aioduct_test_server::h1::h1_server().await;
let client = h2_client();
let url = format!("http://{addr}/");
let result = client.get(&url).unwrap().h2c_prior_knowledge().send().await;
assert!(
result.is_err(),
"h2 prior knowledge against h1 server should fail"
);
}
#[tokio::test]
async fn h2_response_version_is_h2() {
let (addr, _counter) = aioduct_test_server::h2::h2_server().await;
let client = h2_client();
let url = format!("http://{addr}/");
let resp = client
.get(&url)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(
resp.version(),
http::Version::HTTP_2,
"H2 response should report HTTP/2 version"
);
let _ = resp.text().await.unwrap();
}
#[tokio::test]
async fn h1_response_version_is_h11() {
let (addr, _counter) = aioduct_test_server::h1::h1_server().await;
let client = h1_client();
let url = format!("http://{addr}/");
let resp = client.get(&url).unwrap().send().await.unwrap();
assert_eq!(
resp.version(),
http::Version::HTTP_11,
"H1 response should report HTTP/1.1 version"
);
let _ = resp.text().await.unwrap();
}
#[cfg(feature = "rustls")]
mod tls_tests {
use super::*;
fn install_provider() {
aioduct_test_server::tls::install_crypto_provider();
}
#[tokio::test]
async fn tls_h2_alpn_negotiation() {
install_provider();
let (addr, cert_der, _counter) = aioduct_test_server::tls::tls_h2_server().await;
let client_config = aioduct_test_server::tls::make_client_config(&cert_der);
let connector = aioduct::tls::RustlsConnector::new(client_config);
let client: HttpEngineSend<TokioRuntime, TcpConnector> = HttpEngineSend::builder()
.tls(connector)
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let resp = client
.get(&format!("https://localhost:{}/", addr.port()))
.unwrap()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
assert_eq!(
resp.version(),
http::Version::HTTP_2,
"TLS with h2 ALPN should negotiate HTTP/2"
);
let body = resp.text().await.unwrap();
assert_eq!(body, "hello tls");
}
#[tokio::test]
async fn tls_h1_fallback() {
install_provider();
let (addr, cert_der, _counter) =
aioduct_test_server::tls::tls_h1_server(&[b"http/1.1"]).await;
let client_config = aioduct_test_server::tls::make_client_config(&cert_der);
let connector = aioduct::tls::RustlsConnector::new(client_config);
let client: HttpEngineSend<TokioRuntime, TcpConnector> = HttpEngineSend::builder()
.tls(connector)
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let resp = client
.get(&format!("https://localhost:{}/", addr.port()))
.unwrap()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
assert_eq!(
resp.version(),
http::Version::HTTP_11,
"server offering only http/1.1 ALPN should fall back to HTTP/1.1"
);
let body = resp.text().await.unwrap();
assert_eq!(body, "hello tls");
}
#[tokio::test]
async fn tls_no_alpn_fallback() {
install_provider();
let (addr, cert_der, _counter) = aioduct_test_server::tls::tls_h1_server(&[]).await;
let client_config = aioduct_test_server::tls::make_client_config(&cert_der);
let connector = aioduct::tls::RustlsConnector::new(client_config);
let client: HttpEngineSend<TokioRuntime, TcpConnector> = HttpEngineSend::builder()
.tls(connector)
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let result = client
.get(&format!("https://localhost:{}/", addr.port()))
.unwrap()
.send()
.await;
match result {
Ok(resp) => {
assert_eq!(resp.status(), 200);
assert_eq!(
resp.version(),
http::Version::HTTP_11,
"no ALPN should fall back to HTTP/1.1"
);
let _ = resp.text().await.unwrap();
}
Err(e) => {
let msg = format!("{e}");
assert!(
!msg.contains("timeout"),
"no-ALPN should not cause a timeout hang, got: {e}"
);
}
}
}
#[tokio::test]
async fn tls_invalid_cert_rejected() {
install_provider();
let (addr, _cert_der, _counter) = aioduct_test_server::tls::tls_h2_server().await;
let connector = aioduct::tls::RustlsConnector::with_webpki_roots();
let client: HttpEngineSend<TokioRuntime, TcpConnector> = HttpEngineSend::builder()
.tls(connector)
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let result = client
.get(&format!("https://localhost:{}/", addr.port()))
.unwrap()
.send()
.await;
assert!(
result.is_err(),
"self-signed cert without trust root should be rejected, but got: {:?}",
result.as_ref().map(|r| r.status())
);
}
#[tokio::test]
async fn tls_response_has_tls_info() {
install_provider();
let (addr, cert_der, _counter) = aioduct_test_server::tls::tls_h2_server().await;
let client_config = aioduct_test_server::tls::make_client_config(&cert_der);
let connector = aioduct::tls::RustlsConnector::new(client_config);
let client: HttpEngineSend<TokioRuntime, TcpConnector> = HttpEngineSend::builder()
.tls(connector)
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let resp = client
.get(&format!("https://localhost:{}/", addr.port()))
.unwrap()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
assert!(
resp.tls_info().is_some(),
"TLS response should include TLS info"
);
let _ = resp.text().await.unwrap();
}
#[tokio::test]
async fn tls_connection_reuse() {
install_provider();
let (addr, cert_der, counter) = aioduct_test_server::tls::tls_h2_server().await;
let client_config = aioduct_test_server::tls::make_client_config(&cert_der);
let connector = aioduct::tls::RustlsConnector::new(client_config);
let client: HttpEngineSend<TokioRuntime, TcpConnector> = HttpEngineSend::builder()
.tls(connector)
.pool_idle_timeout(Duration::from_secs(60))
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let url = format!("https://localhost:{}/", addr.port());
for _ in 0..3 {
let resp = client.get(&url).unwrap().send().await.unwrap();
assert_eq!(resp.status(), 200);
let _ = resp.text().await.unwrap();
}
assert_eq!(
counter.connections(),
1,
"TLS H2 requests should reuse a single connection"
);
}
}
#[tokio::test]
async fn h2_goaway_with_concurrent_streams() {
let (addr, counter) = aioduct_test_server::h2::h2_goaway_after(1).await;
let client = h2_client();
let url = format!("http://{addr}/");
let warm = client
.get(&url)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(warm.status(), 200);
let _ = warm.text().await.unwrap();
tokio::time::sleep(Duration::from_millis(100)).await;
let mut handles = Vec::new();
for _ in 0..3 {
let c = client.clone();
let u = url.clone();
handles.push(tokio::spawn(async move {
let resp = c
.get(&u)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200, "concurrent request should succeed");
let body = resp.text().await.unwrap();
assert_eq!(body, "ok");
}));
}
for h in handles {
h.await.unwrap();
}
assert_eq!(counter.requests(), 4);
assert!(
counter.connections() >= 2,
"expected >=2 connections, got {}",
counter.connections()
);
}
#[tokio::test]
async fn h2_goaway_with_retry() {
use aioduct::retry::RetryConfig;
let (addr, counter) = aioduct_test_server::h2::h2_goaway_after(1).await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(5))
.retry(
RetryConfig::default()
.max_retries(3)
.initial_backoff(Duration::from_millis(10))
.max_backoff(Duration::from_millis(200)),
)
.build()
.unwrap();
let url = format!("http://{addr}/");
let resp = client
.get(&url)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body = resp.text().await.unwrap();
assert_eq!(body, "ok");
let resp = client
.get(&url)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(
resp.status(),
200,
"second request after GOAWAY should succeed (retry if needed)"
);
let body = resp.text().await.unwrap();
assert_eq!(body, "ok");
assert_eq!(counter.requests(), 2);
}
#[tokio::test]
async fn adaptive_h2c_ttl_expiry_reprobes() {
use bytes::Bytes;
use http_body_util::Full;
let (addr, _counter) = aioduct_test_server::h2::h2_server_with(|req| async move {
let version = format!("{:?}", req.version());
Ok::<_, std::convert::Infallible>(http::Response::new(Full::new(Bytes::from(version))))
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.h2c_probe_ttl(Duration::from_secs(60))
.build()
.unwrap();
let req1 = http::Request::builder()
.method("GET")
.uri("/test")
.body(Full::new(Bytes::new()))
.unwrap();
let resp = client
.forward(valid_forward_request(req1))
.upstream(
format!("http://127.0.0.1:{}", addr.port())
.parse::<http::Uri>()
.unwrap(),
)
.adaptive_h2c()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body = resp.text().await.unwrap();
assert!(
body.contains("HTTP/2"),
"adaptive h2c should negotiate HTTP/2 against h2 server, got: {body}"
);
let req2 = http::Request::builder()
.method("GET")
.uri("/test2")
.body(Full::new(Bytes::new()))
.unwrap();
let resp = client
.forward(valid_forward_request(req2))
.upstream(
format!("http://127.0.0.1:{}", addr.port())
.parse::<http::Uri>()
.unwrap(),
)
.adaptive_h2c()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body = resp.text().await.unwrap();
assert!(
body.contains("HTTP/2"),
"cached h2c should still use HTTP/2, got: {body}"
);
}
#[tokio::test]
async fn http2_config_keep_alive_applied() {
let (addr, counter) = aioduct_test_server::h2::h2_server().await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.http2_keep_alive_while_idle(true)
.http2_keep_alive_timeout(Duration::from_secs(10))
.pool_idle_timeout(Duration::from_secs(60))
.timeout(Duration::from_secs(10))
.build()
.unwrap();
let url = format!("http://{addr}/");
let resp = client
.get(&url)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body = resp.text().await.unwrap();
assert_eq!(body, "hello aioduct");
tokio::time::sleep(Duration::from_secs(3)).await;
let resp = client
.get(&url)
.unwrap()
.h2c_prior_knowledge()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body = resp.text().await.unwrap();
assert_eq!(body, "hello aioduct");
assert_eq!(
counter.connections(),
1,
"connection should be reused after 3s idle"
);
assert_eq!(counter.requests(), 2);
}
#[cfg(all(feature = "rustls", feature = "http3"))]
mod h3_edge_case_tests {
use super::*;
fn install_provider() {
aioduct_test_server::tls::install_crypto_provider();
}
#[tokio::test]
async fn h3_connection_refused_is_connect_error() {
install_provider();
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.tls(aioduct::tls::RustlsConnector::danger_accept_invalid_certs())
.http3(true)
.unwrap()
.timeout(Duration::from_millis(500))
.build()
.unwrap();
let result = client.get("https://127.0.0.1:1/").unwrap().send().await;
assert!(result.is_err(), "H3 connection to closed port should fail");
let err = result.unwrap_err();
assert!(
err.is_timeout(),
"H3 to closed port should time out, got: {err}"
);
}
}