#![cfg(feature = "tokio")]
use std::time::Duration;
use aioduct::HttpEngineSend;
use aioduct::runtime::TokioRuntime;
use aioduct::runtime::tokio_rt::TcpConnector;
use tokio::io::AsyncWriteExt;
use tokio::net::TcpListener;
#[tokio::test]
async fn invalid_http_status_line() {
let addr = aioduct_test_server::raw::raw_server(|_req| async {
b"NOT HTTP AT ALL\r\ngarbage garbage garbage\r\n\r\n".to_vec()
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(2))
.build()
.unwrap();
let result = client.get(&format!("http://{addr}/")).unwrap().send().await;
assert!(
result.is_err(),
"invalid HTTP status line must produce an error, got: {:?}",
result.ok().map(|r| r.status())
);
}
#[tokio::test]
async fn truncated_chunked_encoding() {
let addr = aioduct_test_server::raw::raw_streaming_server(|_req, mut stream| async move {
let headers = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n";
let _ = stream.write_all(headers).await;
let _ = stream.write_all(b"5\r\nhello\r\n").await;
let _ = stream.flush().await;
let _ = stream.write_all(b"64\r\npartial").await;
let _ = stream.flush().await;
let _ = stream.shutdown().await;
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(2))
.build()
.unwrap();
let resp = client.get(&format!("http://{addr}/")).unwrap().send().await;
match resp {
Err(_) => {} Ok(resp) => {
let body_result = resp.bytes().await;
assert!(
body_result.is_err(),
"truncated chunked body must produce an error on read"
);
}
}
}
#[tokio::test]
async fn headers_too_large() {
let addr = aioduct_test_server::raw::raw_server(|_req| async {
let mut response = Vec::with_capacity(70_000);
response.extend_from_slice(b"HTTP/1.1 200 OK\r\n");
response.extend_from_slice(b"X-Huge: ");
response.extend(vec![b'x'; 65_536]);
response.extend_from_slice(b"\r\n\r\nbody");
response
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(2))
.build()
.unwrap();
let result = client.get(&format!("http://{addr}/")).unwrap().send().await;
match result {
Err(_) => {} Ok(resp) => {
assert_eq!(resp.status(), 200);
}
}
}
#[tokio::test]
async fn response_timeout_mid_headers() {
let addr = aioduct_test_server::raw::raw_streaming_server(|_req, mut stream| async move {
let _ = stream
.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\n")
.await;
let _ = stream.flush().await;
tokio::time::sleep(Duration::from_secs(60)).await;
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_millis(500))
.build()
.unwrap();
let start = tokio::time::Instant::now();
let result = client.get(&format!("http://{addr}/")).unwrap().send().await;
assert!(result.is_err(), "stalled headers must trigger timeout");
let err = result.unwrap_err();
assert!(err.is_timeout(), "expected timeout error, got: {err:?}");
assert!(
start.elapsed() < Duration::from_secs(5),
"timeout should fire within a reasonable time, not hang"
);
}
#[tokio::test]
async fn response_timeout_mid_body() {
let addr = aioduct_test_server::raw::raw_streaming_server(|_req, mut stream| async move {
let _ = stream
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 1000\r\n\r\npartial")
.await;
let _ = stream.flush().await;
tokio::time::sleep(Duration::from_secs(60)).await;
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.read_timeout(Duration::from_millis(300))
.timeout(Duration::from_secs(3))
.build()
.unwrap();
let resp = client.get(&format!("http://{addr}/")).unwrap().send().await;
match resp {
Err(e) => {
assert!(e.is_timeout(), "expected timeout error, got: {e:?}");
}
Ok(resp) => {
let body_result = resp.bytes().await;
assert!(
body_result.is_err(),
"stalled body must produce an error (timeout or connection error)"
);
}
}
}
#[tokio::test]
async fn incomplete_chunked_never_terminated() {
let addr = aioduct_test_server::raw::raw_streaming_server(|_req, mut stream| async move {
let _ = stream
.write_all(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n")
.await;
let _ = stream.flush().await;
tokio::time::sleep(Duration::from_secs(60)).await;
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.read_timeout(Duration::from_millis(300))
.timeout(Duration::from_secs(3))
.build()
.unwrap();
let resp = client.get(&format!("http://{addr}/")).unwrap().send().await;
match resp {
Err(e) => {
assert!(e.is_timeout(), "expected timeout, got: {e:?}");
}
Ok(resp) => {
let body_result = resp.bytes().await;
assert!(
body_result.is_err(),
"never-terminated chunked stream must error on body read"
);
}
}
}
#[tokio::test]
async fn content_length_mismatch_short_body() {
let addr = aioduct_test_server::raw::raw_server(|_req| async {
b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\nshort".to_vec()
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(2))
.build()
.unwrap();
let resp = client.get(&format!("http://{addr}/")).unwrap().send().await;
match resp {
Err(_) => {} Ok(resp) => {
let body_result = resp.bytes().await;
assert!(
body_result.is_err(),
"content-length mismatch (short body) must produce an error"
);
}
}
}
#[tokio::test]
async fn empty_response_immediate_close() {
let addr = aioduct_test_server::raw::raw_streaming_server(|_req, mut stream| async move {
let _ = stream.shutdown().await;
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(2))
.build()
.unwrap();
let result = client.get(&format!("http://{addr}/")).unwrap().send().await;
assert!(
result.is_err(),
"empty response (immediate close) must produce an error"
);
}
#[tokio::test]
async fn http09_style_response() {
let addr = aioduct_test_server::raw::raw_server(|_req| async {
b"Hello this is the body with no status line\r\n".to_vec()
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(2))
.build()
.unwrap();
let result = client.get(&format!("http://{addr}/")).unwrap().send().await;
assert!(
result.is_err(),
"HTTP/0.9 style response (no status line) must produce an error"
);
}
#[tokio::test]
async fn null_bytes_in_headers() {
let addr = aioduct_test_server::raw::raw_server(|_req| async {
let mut resp = Vec::new();
resp.extend_from_slice(b"HTTP/1.1 200 OK\r\n");
resp.extend_from_slice(b"X-Bad: value\x00with\x00nulls\r\n");
resp.extend_from_slice(b"Content-Length: 2\r\n\r\nok");
resp
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(2))
.build()
.unwrap();
let result = client.get(&format!("http://{addr}/")).unwrap().send().await;
match result {
Err(_) => {} Ok(resp) => {
let _ = resp.status();
}
}
}
#[tokio::test]
async fn duplicate_conflicting_content_length() {
let addr = aioduct_test_server::raw::raw_server(|_req| async {
b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\nContent-Length: 100\r\n\r\nhello".to_vec()
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(2))
.build()
.unwrap();
let result = client.get(&format!("http://{addr}/")).unwrap().send().await;
match result {
Err(_) => {} Ok(resp) => {
assert_eq!(resp.status(), 200);
}
}
}
#[tokio::test]
async fn connection_refused() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
drop(listener);
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(2))
.build()
.unwrap();
let result = client.get(&format!("http://{addr}/")).unwrap().send().await;
assert!(result.is_err(), "connection to closed port must fail");
let err = result.unwrap_err();
assert!(
err.is_connect(),
"expected connect error for connection refused, got: {err:?}"
);
}
#[tokio::test]
async fn connection_reset_during_body() {
let addr = aioduct_test_server::raw::raw_streaming_server(|_req, mut stream| async move {
let _ = stream
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 10000\r\n\r\nstart")
.await;
let _ = stream.flush().await;
tokio::time::sleep(Duration::from_millis(50)).await;
let raw = stream.into_std().unwrap();
let sock = socket2::SockRef::from(&raw);
let _ = sock.set_linger(Some(Duration::from_secs(0)));
drop(raw);
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let resp = client.get(&format!("http://{addr}/")).unwrap().send().await;
match resp {
Err(_) => {} Ok(resp) => {
let body_result = resp.bytes().await;
assert!(
body_result.is_err(),
"connection reset mid-body must produce an error on body read"
);
}
}
}
#[tokio::test]
async fn connect_timeout_fires() {
let addr = aioduct_test_server::raw::blackhole_server().await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_millis(500))
.build()
.unwrap();
let start = tokio::time::Instant::now();
let result = client.get(&format!("http://{addr}/")).unwrap().send().await;
assert!(result.is_err(), "blackhole server must trigger timeout");
let err = result.unwrap_err();
assert!(
err.is_timeout(),
"expected timeout error for blackhole, got: {err:?}"
);
assert!(
start.elapsed() < Duration::from_secs(5),
"timeout should fire promptly"
);
}
#[tokio::test]
async fn connect_timeout_with_nonroutable_address() {
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.connect_timeout(Duration::from_millis(200))
.timeout(Duration::from_secs(30))
.build()
.unwrap();
let start = tokio::time::Instant::now();
let result = client.get("http://192.0.2.1:80/").unwrap().send().await;
assert!(result.is_err(), "non-routable address must fail");
let elapsed = start.elapsed();
assert!(
elapsed < Duration::from_secs(5),
"connect_timeout should fire quickly, took {:?}",
elapsed
);
}
#[tokio::test]
async fn slowloris_response() {
let addr = aioduct_test_server::raw::raw_streaming_server(|_req, mut stream| async move {
let response = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok";
for &byte in response.iter() {
let _ = stream.write_all(&[byte]).await;
let _ = stream.flush().await;
tokio::time::sleep(Duration::from_millis(200)).await;
}
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_millis(500))
.build()
.unwrap();
let start = tokio::time::Instant::now();
let result = client.get(&format!("http://{addr}/")).unwrap().send().await;
assert!(result.is_err(), "slowloris response must trigger timeout");
assert!(
start.elapsed() < Duration::from_secs(3),
"timeout should fire promptly"
);
}
#[tokio::test]
async fn error_is_connect_for_connection_failures() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
drop(listener);
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(2))
.build()
.unwrap();
let err = client
.get(&format!("http://{addr}/"))
.unwrap()
.send()
.await
.unwrap_err();
assert!(
err.is_connect(),
"connection refused must have is_connect() == true, got: {err:?}"
);
assert!(
!err.is_timeout(),
"connection refused should NOT be a timeout, got: {err:?}"
);
}
#[tokio::test]
async fn error_is_timeout_for_timeouts() {
let addr = aioduct_test_server::raw::blackhole_server().await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_millis(200))
.build()
.unwrap();
let err = client
.get(&format!("http://{addr}/"))
.unwrap()
.send()
.await
.unwrap_err();
assert!(
err.is_timeout(),
"timeout must have is_timeout() == true, got: {err:?}"
);
assert!(
!err.is_connect(),
"generic timeout should NOT be is_connect(); use connect_timeout() for that"
);
}
#[tokio::test]
async fn error_classification_for_malformed_response() {
let addr =
aioduct_test_server::raw::raw_server(|_req| async { b"GARBAGE\r\n\r\n".to_vec() }).await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(2))
.build()
.unwrap();
let result = client.get(&format!("http://{addr}/")).unwrap().send().await;
if let Err(err) = result {
assert!(
!err.is_timeout(),
"malformed response should NOT be a timeout: {err:?}"
);
}
}
#[tokio::test]
async fn client_recovers_after_connection_error() {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let dead_addr = listener.local_addr().unwrap();
drop(listener);
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(2))
.build()
.unwrap();
let result = client
.get(&format!("http://{dead_addr}/"))
.unwrap()
.send()
.await;
assert!(result.is_err(), "dead port should fail");
let (live_addr, _counter) = aioduct_test_server::h1::h1_server().await;
let resp = client
.get(&format!("http://{live_addr}/"))
.unwrap()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
assert_eq!(resp.text().await.unwrap(), "hello aioduct");
}
#[tokio::test]
async fn client_recovers_after_timeout() {
let blackhole_addr = aioduct_test_server::raw::blackhole_server().await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_millis(200))
.build()
.unwrap();
let result = client
.get(&format!("http://{blackhole_addr}/"))
.unwrap()
.send()
.await;
assert!(result.is_err());
assert!(result.unwrap_err().is_timeout());
let (live_addr, _counter) = aioduct_test_server::h1::h1_server().await;
let resp = client
.get(&format!("http://{live_addr}/"))
.unwrap()
.timeout(Duration::from_secs(5))
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
assert_eq!(resp.text().await.unwrap(), "hello aioduct");
}
#[tokio::test]
async fn concurrent_requests_mixed_healthy_and_broken() {
let (live_addr, _counter) = aioduct_test_server::h1::h1_server().await;
let blackhole_addr = aioduct_test_server::raw::blackhole_server().await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_millis(500))
.build()
.unwrap();
let mut handles = Vec::new();
for _ in 0..5 {
let client = client.clone();
let url = format!("http://{live_addr}/");
handles.push(tokio::spawn(async move {
let resp = client.get(&url).unwrap().send().await.unwrap();
assert_eq!(resp.status(), 200);
true
}));
}
for _ in 0..3 {
let client = client.clone();
let url = format!("http://{blackhole_addr}/");
handles.push(tokio::spawn(async move {
let result = client.get(&url).unwrap().send().await;
result.is_err()
}));
}
for handle in handles {
let ok = handle.await.unwrap();
assert!(
ok,
"each request should either succeed or error without panic"
);
}
}
#[tokio::test]
async fn partial_body_rst_mid_stream_returns_error() {
let addr = aioduct_test_server::raw::raw_streaming_server(|_req, mut stream| async move {
let headers = b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n";
let _ = stream.write_all(headers).await;
for _ in 0..2 {
let chunk = format!("{:x}\r\n{}\r\n", 1024, "x".repeat(1024));
let _ = stream.write_all(chunk.as_bytes()).await;
let _ = stream.flush().await;
}
drop(stream);
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let resp = client
.get(&format!("http://{addr}/"))
.unwrap()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let result = resp.bytes().await;
assert!(
result.is_err(),
"Incomplete chunked transfer (missing final chunk) should return error, not truncate"
);
}
#[tokio::test]
async fn download_zero_byte_body() {
use bytes::Bytes;
use http_body_util::Full;
use hyper::Response;
use std::convert::Infallible;
let (addr, _) = aioduct_test_server::h1::h1_server_with(|_req| async {
let resp = Response::builder()
.header("Content-Length", "0")
.body(Full::new(Bytes::new()))
.unwrap();
Ok::<_, Infallible>(resp)
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let resp = client
.get(&format!("http://{addr}/"))
.unwrap()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body = resp.bytes().await.unwrap();
assert_eq!(body.len(), 0, "0-byte body should be empty");
}
#[tokio::test]
async fn stuttered_chunked_download_complete() {
let (addr, _) =
aioduct_test_server::h1::h1_slow_body_server(100, Duration::from_millis(5)).await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(10))
.build()
.unwrap();
let resp = client
.get(&format!("http://{addr}/"))
.unwrap()
.send()
.await
.unwrap();
assert_eq!(resp.status(), 200);
let body = resp.bytes().await.unwrap();
assert_eq!(
body.len(),
1000,
"stuttered chunked download should deliver complete body"
);
}
#[tokio::test]
async fn error_timeout_should_distinguish_connect_vs_read() {
let addr =
aioduct_test_server::raw::raw_streaming_server(|_request_bytes, mut stream| async move {
use tokio::io::AsyncWriteExt;
stream
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 1000\r\n\r\n")
.await
.unwrap();
stream.write_all(b"partial").await.unwrap();
stream.flush().await.unwrap();
tokio::time::sleep(Duration::from_secs(60)).await;
})
.await;
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.read_timeout(Duration::from_millis(200))
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let resp = client
.get(&format!("http://{addr}/"))
.unwrap()
.send()
.await
.unwrap();
let err = resp.bytes().await.unwrap_err();
assert!(
err.is_timeout(),
"read timeout error should report is_timeout() = true, got: {err}"
);
assert!(
!err.is_connect(),
"BUG: error.rs:135 includes Error::Timeout in is_connect(). \
A read timeout during body streaming is NOT a connection failure. \
is_connect() should be false for read timeouts, but got true."
);
}
#[tokio::test]
async fn error_hyper_connection_refused_should_be_connect() {
let client = HttpEngineSend::<TokioRuntime, TcpConnector>::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let result = client.get("http://127.0.0.1:1/").unwrap().send().await;
let err = result.unwrap_err();
assert!(
err.is_connect(),
"Connection refused error should report is_connect() = true. \
Error: {err}"
);
}