use std::time::{Duration, Instant};
use futures_util::StreamExt;
use leyline::http::Method;
use leyline::{ProtocolPolicy, ProxyConfig, RetryPolicy, Session, TimeoutConfig};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
async fn serve<F, Fut>(after: F) -> std::net::SocketAddr
where
F: FnOnce(tokio::net::TcpStream) -> Fut + Send + 'static,
Fut: std::future::Future<Output = ()> + Send,
{
let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind");
let addr = listener.local_addr().expect("local addr");
drop(tokio::spawn(async move {
let (mut sock, _) = listener.accept().await.expect("accept");
let mut buf = [0u8; 2048];
drop(sock.read(&mut buf).await);
after(sock).await;
}));
addr
}
#[tokio::test]
async fn connect_timeout_fires_on_a_black_hole_address() {
let session = Session::builder()
.proxy(ProxyConfig::new().env(false))
.timeout(
TimeoutConfig::new()
.total(Duration::from_secs(10))
.connect(Duration::from_millis(300)),
)
.build()
.expect("session builds");
let start = Instant::now();
let err = session
.request(Method::GET, "https://192.0.2.1:81/")
.retry(RetryPolicy::none())
.send()
.await
.expect_err("a black-hole address must not connect");
let elapsed = start.elapsed();
assert!(err.is_timeout(), "expected a timeout, got {err:?}");
assert!(
elapsed < Duration::from_secs(2),
"connect cap should fire near 300ms, took {elapsed:?}"
);
}
#[tokio::test]
async fn response_header_timeout_fires_when_the_server_never_writes() {
let addr = serve(|sock| async move {
tokio::time::sleep(Duration::from_secs(30)).await;
drop(sock);
})
.await;
let session = Session::builder()
.protocol(ProtocolPolicy::Http1)
.proxy(ProxyConfig::new().env(false))
.build()
.expect("session builds");
let start = Instant::now();
let err = session
.request(Method::GET, format!("http://{addr}/"))
.timeout(
TimeoutConfig::default()
.total(Duration::from_secs(10))
.response_header(Duration::from_millis(300)),
)
.send()
.await
.expect_err("a silent server must not resolve");
let elapsed = start.elapsed();
assert!(err.is_timeout(), "expected a timeout, got {err:?}");
assert!(
elapsed < Duration::from_secs(2),
"per-request TTFB cap should fire near 300ms, took {elapsed:?}"
);
}
async fn half_chunked(mut sock: tokio::net::TcpStream) {
drop(
sock.write_all(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n4\r\nhalf\r\n")
.await,
);
drop(sock.flush().await);
tokio::time::sleep(Duration::from_secs(30)).await;
drop(sock);
}
#[tokio::test]
async fn read_timeout_fires_between_chunks_of_a_streamed_body() {
let addr = serve(half_chunked).await;
let session = Session::builder()
.protocol(ProtocolPolicy::Http1)
.proxy(ProxyConfig::new().env(false))
.build()
.expect("session builds");
let resp = session
.request(Method::GET, format!("http://{addr}/"))
.timeout(
TimeoutConfig::default()
.total(Duration::from_secs(10))
.read(Duration::from_millis(300)),
)
.stream()
.send()
.await
.expect("headers arrive before the stall");
assert_eq!(resp.status(), 200);
let start = Instant::now();
let mut body = resp.into_stream().expect("streaming body");
let first = body
.next()
.await
.expect("first chunk")
.expect("chunk is ok");
assert_eq!(&first[..], b"half");
let stalled = body.next().await.expect("a stalled stream yields an error");
let elapsed = start.elapsed();
let err = leyline::Error::from(stalled.expect_err("second chunk must time out"));
assert!(err.is_timeout(), "expected a timeout, got {err:?}");
assert!(
elapsed < Duration::from_secs(2),
"read cap should fire near 300ms, took {elapsed:?}"
);
}
#[tokio::test]
async fn read_timeout_does_not_cover_a_buffered_body() {
let addr = serve(half_chunked).await;
let session = Session::builder()
.protocol(ProtocolPolicy::Http1)
.proxy(ProxyConfig::new().env(false))
.build()
.expect("session builds");
let start = Instant::now();
let err = session
.request(Method::GET, format!("http://{addr}/"))
.timeout(
TimeoutConfig::default()
.total(Duration::from_secs(10))
.read(Duration::from_millis(200))
.response_header(Duration::from_millis(700)),
)
.send()
.await
.expect_err("a half-written buffered body must not resolve");
let elapsed = start.elapsed();
assert!(err.is_timeout(), "expected a timeout, got {err:?}");
assert!(
elapsed >= Duration::from_millis(600),
"a buffered body is read inside the TTFB window, so `read` must not clip it at 200ms; bailed at {elapsed:?}"
);
assert!(
elapsed < Duration::from_secs(2),
"TTFB cap should fire near 700ms, took {elapsed:?}"
);
}
#[tokio::test]
async fn total_timeout_bounds_a_slow_server() {
let addr = serve(|mut sock| async move {
tokio::time::sleep(Duration::from_secs(30)).await;
drop(
sock.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
.await,
);
})
.await;
let session = Session::builder()
.protocol(ProtocolPolicy::Http1)
.proxy(ProxyConfig::new().env(false))
.build()
.expect("session builds");
let start = Instant::now();
let err = session
.request(Method::GET, format!("http://{addr}/"))
.timeout(TimeoutConfig::default().total(Duration::from_millis(400)))
.send()
.await
.expect_err("a 30s server must not beat a 400ms total");
let elapsed = start.elapsed();
assert!(err.is_timeout(), "expected a timeout, got {err:?}");
assert!(
elapsed >= Duration::from_millis(350) && elapsed < Duration::from_secs(2),
"total cap should fire near 400ms, took {elapsed:?}"
);
}
#[tokio::test]
async fn a_total_timeout_of_duration_max_sets_no_deadline() {
let addr = serve(|mut sock| async move {
drop(
sock.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
.await,
);
})
.await;
let session = Session::builder()
.protocol(ProtocolPolicy::Http1)
.proxy(ProxyConfig::new().env(false))
.timeout(TimeoutConfig::default().total(Duration::MAX))
.build()
.expect("session builds");
let response = session
.request(Method::GET, format!("http://{addr}/"))
.send()
.await
.expect("Duration::MAX means no total deadline");
assert_eq!(&response.bytes().await.expect("body")[..], b"ok");
}