ephemeris_loopback/
lib.rs

1//! A throwaway HTTP/1.1 server on an ephemeral loopback port, shared by the
2//! tests of the forge crates and their consumers. This is a dev-only crate.
3//!
4//! It answers canned responses and keeps every request it was sent, so a test
5//! can assert on the bytes the client put on the wire. It is hand written
6//! rather than taken from a crate such as `wiremock`: the whole server is one
7//! accept loop, and a mock HTTP crate would add a second HTTP stack, and the
8//! transitive versions that come with it, to a crate that deliberately vendors
9//! a small dependency set and runs `clippy::cargo`.
10//!
11//! The accept loop runs on a dedicated background runtime rather than the
12//! caller's own, so [`Server::start`] works the same from a `#[test]` and
13//! from a `#[tokio::test]`.
14#![expect(
15    clippy::expect_used,
16    reason = "dev-only test server code; allow-expect-in-tests does not cover it because it is \
17              not itself a #[test] function"
18)]
19
20use std::fmt::Write as _;
21use std::future::pending;
22use std::net::SocketAddr;
23use std::sync::mpsc::channel;
24use std::sync::{Arc, Mutex, OnceLock, PoisonError};
25use std::thread::spawn;
26
27use reqwest::Url;
28use rustls::crypto::CryptoProvider;
29use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
30use tokio::net::{TcpListener, TcpStream};
31use tokio::runtime::{Builder, Handle};
32use tokio::task::JoinHandle;
33
34/// Serializes callers of [`install_tls_provider`] within this process.
35static INSTALLING: Mutex<()> = Mutex::new(());
36
37/// One request as it arrived on the wire.
38#[derive(Clone, Debug)]
39pub struct Request {
40    /// The request method, such as `GET` or `POST`.
41    pub method: String,
42    /// The request target, including its query string.
43    pub path: String,
44    /// Header names are lower-cased; order matches the wire.
45    pub headers: Vec<(String, String)>,
46    /// The request body, decoded as UTF-8 lossily.
47    pub body: String,
48}
49
50impl Request {
51    /// One header, looked up without regard to case.
52    #[must_use]
53    pub fn header(&self, name: &str) -> Option<&str> {
54        self.headers
55            .iter()
56            .find(|(key, _)| key.eq_ignore_ascii_case(name))
57            .map(|(_, value)| value.as_str())
58    }
59}
60
61/// One canned response.
62///
63/// `headers` is how a test hands the client a `Link` or an `X-Total-Count`,
64/// which is what paging reads.
65#[derive(Clone, Debug)]
66pub struct Response {
67    /// The status code to answer with.
68    pub status: u16,
69    /// The response body, sent back exactly as given.
70    pub body: String,
71    /// Extra headers sent with the response.
72    pub headers: Vec<(String, String)>,
73}
74
75impl Response {
76    /// Build a response with no extra headers.
77    #[must_use]
78    pub fn new<B: Into<String>>(status: u16, body: B) -> Self {
79        Self {
80            status,
81            body: body.into(),
82            headers: Vec::new(),
83        }
84    }
85}
86
87/// A server listening on `127.0.0.1` with a port the kernel picked.
88///
89/// Dropping it stops the accept loop.
90#[derive(Debug)]
91pub struct Server {
92    address: SocketAddr,
93    requests: Arc<Mutex<Vec<Request>>>,
94    accept_loop: JoinHandle<()>,
95}
96
97impl Drop for Server {
98    fn drop(&mut self) {
99        self.accept_loop.abort();
100    }
101}
102
103impl Server {
104    /// Start a server that answers `responses` in order.
105    ///
106    /// Once the list runs out the last entry is repeated, so a retrying client
107    /// keeps meeting the same answer.
108    #[must_use]
109    pub fn start(responses: Vec<Response>) -> Self {
110        Self::start_with(move |_address| responses)
111    }
112
113    /// Start a server whose responses are built once the bound address is
114    /// known, for a response body that embeds the server's own URL.
115    ///
116    /// # Panics
117    ///
118    /// Panics when the background server cannot report its address, which
119    /// only happens if its accept loop could not even bind a port.
120    #[must_use]
121    pub fn start_with<F: FnOnce(SocketAddr) -> Vec<Response> + Send + 'static>(
122        build_responses: F,
123    ) -> Self {
124        install_tls_provider();
125        let handle = background_runtime();
126        let (address_tx, address_rx) = channel();
127        let (responses_tx, responses_rx) = channel::<Vec<Response>>();
128        let requests = Arc::new(Mutex::new(Vec::new()));
129        let recorded = Arc::clone(&requests);
130
131        let accept_loop = handle.spawn(async move {
132            let Ok(listener) = TcpListener::bind("127.0.0.1:0").await else {
133                return;
134            };
135            let Ok(address) = listener.local_addr() else {
136                return;
137            };
138            if address_tx.send(address).is_err() {
139                return;
140            }
141            let Ok(responses) = responses_rx.recv() else {
142                return;
143            };
144
145            let mut served = 0usize;
146            while let Ok((mut stream, _)) = listener.accept().await {
147                let request = read_request(&mut stream).await;
148                let response = responses
149                    .get(served)
150                    .or_else(|| responses.last())
151                    .cloned()
152                    .unwrap_or_else(|| Response::new(500, "the test provided no response"));
153                served = served.saturating_add(1);
154                recorded.lock().expect("lock").push(request);
155                write_response(&mut stream, &response).await;
156            }
157        });
158
159        let address = address_rx
160            .recv()
161            .expect("the background server reports its address");
162        let responses = build_responses(address);
163        responses_tx
164            .send(responses)
165            .expect("the responses channel is open");
166
167        Self {
168            address,
169            requests,
170            accept_loop,
171        }
172    }
173
174    /// The server root to point a client at.
175    ///
176    /// # Panics
177    ///
178    /// Never in practice: the address formats as a valid URL by construction.
179    #[must_use]
180    pub fn url(&self) -> Url {
181        Url::parse(&format!("http://{}/", self.address)).expect("a valid URL")
182    }
183
184    /// Every request the server has answered so far, in order.
185    ///
186    /// # Panics
187    ///
188    /// Panics if an earlier panic on the accept loop poisoned the lock.
189    #[must_use]
190    pub fn requests(&self) -> Vec<Request> {
191        self.requests.lock().expect("lock").clone()
192    }
193}
194
195/// Install the OpenSSL TLS provider for tests, keeping an existing provider.
196///
197/// # Panics
198///
199/// Panics when outside code installs another provider concurrently.
200pub fn install_tls_provider() {
201    let _guard = INSTALLING.lock().unwrap_or_else(PoisonError::into_inner);
202    if CryptoProvider::get_default().is_none() {
203        rustls_openssl::default_provider()
204            .install_default()
205            .expect("no other provider is installed concurrently");
206    }
207}
208
209/// A background tokio runtime shared by every server in this test binary.
210///
211/// Spawning onto it, rather than onto the caller's own runtime, is what lets
212/// a plain `#[test]` function use [`Server`] as freely as a `#[tokio::test]`.
213fn background_runtime() -> Handle {
214    static HANDLE: OnceLock<Handle> = OnceLock::new();
215    HANDLE
216        .get_or_init(|| {
217            let (tx, rx) = channel();
218            spawn(move || {
219                let runtime = Builder::new_current_thread()
220                    .enable_all()
221                    .build()
222                    .expect("a background tokio runtime");
223                tx.send(runtime.handle().clone())
224                    .expect("the handle channel is open");
225                runtime.block_on(pending::<()>());
226            });
227            rx.recv()
228                .expect("the background runtime reports its handle")
229        })
230        .clone()
231}
232
233async fn write_response(stream: &mut TcpStream, response: &Response) {
234    let body = response.body.as_bytes();
235    let mut extra = String::new();
236    for (name, value) in &response.headers {
237        write!(extra, "{name}: {value}\r\n").expect("writing to a String cannot fail");
238    }
239    let head = format!(
240        "HTTP/1.1 {} Test\r\ncontent-type: application/json\r\ncontent-length: \
241         {}\r\n{extra}connection: close\r\n\r\n",
242        response.status,
243        body.len()
244    );
245    stream
246        .write_all(head.as_bytes())
247        .await
248        .expect("the head is written");
249    stream.write_all(body).await.expect("the body is written");
250    drop(stream.shutdown().await);
251}
252
253async fn read_request(stream: &mut TcpStream) -> Request {
254    let mut buffer = Vec::new();
255    let head_end = loop {
256        if let Some(index) = find(&buffer, b"\r\n\r\n") {
257            break index.saturating_add(4);
258        }
259        read_more(stream, &mut buffer).await;
260    };
261
262    let head = String::from_utf8_lossy(buffer.get(..head_end).expect("computed within bounds"))
263        .into_owned();
264    let mut lines = head.lines();
265    let mut start = lines.next().unwrap_or_default().split_whitespace();
266    let method = start.next().unwrap_or_default().to_owned();
267    let path = start.next().unwrap_or_default().to_owned();
268    let headers: Vec<(String, String)> = lines
269        .filter_map(|line| line.split_once(':'))
270        .map(|(key, value)| (key.trim().to_lowercase(), value.trim().to_owned()))
271        .collect();
272
273    let length: usize = headers
274        .iter()
275        .find(|(key, _)| key == "content-length")
276        .and_then(|(_, value)| value.parse().ok())
277        .unwrap_or_default();
278    let body_end = head_end.saturating_add(length);
279    while buffer.len() < body_end {
280        read_more(stream, &mut buffer).await;
281    }
282    let body = String::from_utf8_lossy(
283        buffer
284            .get(head_end..body_end)
285            .expect("computed within bounds"),
286    )
287    .into_owned();
288
289    Request {
290        method,
291        path,
292        headers,
293        body,
294    }
295}
296
297async fn read_more(stream: &mut TcpStream, buffer: &mut Vec<u8>) {
298    let mut chunk = [0u8; 1024];
299    let read = stream
300        .read(&mut chunk)
301        .await
302        .expect("the socket is readable");
303    assert!(read > 0, "the client closed before finishing its request");
304    buffer.extend_from_slice(chunk.get(..read).expect("computed within bounds"));
305}
306
307fn find(haystack: &[u8], needle: &[u8]) -> Option<usize> {
308    haystack
309        .windows(needle.len())
310        .position(|window| window == needle)
311}
ephemeris_loopback/lib.rs

ephemeris_loopback/
lib.rs
