//! A tiny HTTP/1.1 mock server for client tests (one request per connection, scripted or routed replies). use super::client::{ClientOptions, ErpClient}; use super::config::Secret; use serde_json::Value; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::net::TcpListener; pub struct Reply { pub status: u16, pub headers: Vec<(&'static str, String)>, pub body: String, /// Pause before answering, to trigger the read timeout. pub delay_ms: u64, } pub fn reply(status: u16, body: Value) -> Reply { Reply { status, headers: vec![], body: body.to_string(), delay_ms: 0, } } pub struct Mock { pub base: String, pub hits: Arc, /// Raw requests (head and body) in arrival order. pub requests: Arc>>, } /// Serves the scripted replies in order; the last one repeats. pub async fn serve(script: Vec) -> Mock { let script = Arc::new(script); serve_fn(move |i, _| { let r = &script[i.min(script.len() - 1)]; Reply { status: r.status, headers: r.headers.clone(), body: r.body.clone(), delay_ms: r.delay_ms, } }) .await } /// Answers every request with `handler(request_index, request_line)`. pub async fn serve_fn(handler: impl Fn(usize, &str) -> Reply + Send + Sync + 'static) -> Mock { let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let port = listener.local_addr().unwrap().port(); let hits = Arc::new(AtomicUsize::new(0)); let requests = Arc::new(Mutex::new(Vec::new())); let (h, r) = (hits.clone(), requests.clone()); let handler = Arc::new(handler); tokio::spawn(async move { loop { let Ok((mut sock, _)) = listener.accept().await else { return; }; let (h, r, handler) = (h.clone(), r.clone(), handler.clone()); tokio::spawn(async move { let mut buf = Vec::new(); let mut chunk = [0u8; 2048]; loop { let n = sock.read(&mut chunk).await.unwrap_or(0); if n == 0 { break; } buf.extend_from_slice(&chunk[..n]); if let Some(end) = buf.windows(4).position(|w| w == b"\r\n\r\n") { let head = String::from_utf8_lossy(&buf[..end]).to_string(); let want = head .lines() .find_map(|l| { l.to_ascii_lowercase() .strip_prefix("content-length:") .map(|v| v.trim().parse::().unwrap_or(0)) }) .unwrap_or(0); if buf.len() >= end + 4 + want { break; } } } let raw = String::from_utf8_lossy(&buf).to_string(); let request_line = raw.lines().next().unwrap_or("").to_string(); r.lock().unwrap().push(raw); let i = h.fetch_add(1, Ordering::SeqCst); let rep = handler(i, &request_line); if rep.delay_ms > 0 { tokio::time::sleep(Duration::from_millis(rep.delay_ms)).await; } let mut out = format!( "HTTP/1.1 {} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n", rep.status, rep.body.len() ); for (k, v) in &rep.headers { out.push_str(&format!("{k}: {v}\r\n")); } out.push_str("\r\n"); out.push_str(&rep.body); let _ = sock.write_all(out.as_bytes()).await; let _ = sock.shutdown().await; }); } }); Mock { base: format!("http://127.0.0.1:{port}"), hits, requests, } } /// Short timeouts and near-zero backoff so retry tests run in milliseconds. pub fn fast_opts() -> ClientOptions { ClientOptions { connect_timeout: Duration::from_secs(2), read_timeout: Duration::from_millis(250), total_timeout: Duration::from_secs(5), max_retries: 3, backoff_base: Duration::from_millis(1), max_backoff: Duration::from_millis(5), } } pub fn client(mock: &Mock) -> ErpClient { ErpClient::new( &mock.base, "thekey", &Secret::new("thesecret"), "", fast_opts(), ) .unwrap() }