src/federation/testutil.rs
//! A minimal, network-free mock HTTP/1.1 server for harbour tests. Binds a
//! loopback `TcpListener`, records every request (method, path, headers) and
//! answers from a caller-supplied responder. No external crates, no real network,
//! no containers — so every harbour's URL/auth/parse logic is proven offline.
//!
//! Two responder shapes:
//! - [`MockServer::start`] — the simple `(method, path) -> (status, body)` form
//! used by the file-path harbours' fixtures (no custom response headers).
//! - [`MockServer::start_with`] — the full form: the responder sees the whole
//! [`RecordedRequest`] (so it can branch on `Authorization`) and returns a
//! [`MockResponse`] with arbitrary response **headers** (so it can emit
//! `WWW-Authenticate`, `Docker-Content-Digest`, `X-Checksum-Sha256`, …).
//! `start` is a thin wrapper over `start_with`, so existing header-less fixtures
//! keep working unchanged.
use std::collections::HashMap;
use std::io::{Read, Write};
use std::net::TcpListener;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;
#[derive(Clone, Debug)]
pub struct RecordedRequest {
pub method: String,
pub path: String,
pub headers: HashMap<String, String>,
}
impl RecordedRequest {
pub fn header(&self, name: &str) -> Option<&str> {
self.headers.get(&name.to_lowercase()).map(String::as_str)
}
}
/// A mock HTTP response: status + response headers + body. Headers let a fixture
/// emit the challenge/checksum headers real registries return.
#[derive(Clone, Debug)]
pub struct MockResponse {
pub status: u16,
pub headers: Vec<(String, String)>,
pub body: Vec<u8>,
}
impl MockResponse {
pub fn new(status: u16, body: impl Into<Vec<u8>>) -> Self {
Self {
status,
headers: Vec::new(),
body: body.into(),
}
}
/// Add a response header (builder). Repeatable.
pub fn with_header(mut self, name: impl Into<String>, value: impl Into<String>) -> Self {
self.headers.push((name.into(), value.into()));
self
}
}
pub struct MockServer {
pub base_url: String,
requests: Arc<Mutex<Vec<RecordedRequest>>>,
shutdown: Arc<AtomicBool>,
handle: Option<gatling::background::Job<()>>,
}
impl MockServer {
/// Start a server whose responder, given `(method, path)`, returns
/// `(status, body)` — the simple, header-less form. A HEAD request returns
/// headers + Content-Length but no body.
pub fn start<F>(responder: F) -> Self
where
F: Fn(&str, &str) -> (u16, Vec<u8>) + Send + Sync + 'static,
{
Self::start_with(move |req| {
let (status, body) = responder(&req.method, &req.path);
MockResponse::new(status, body)
})
}
/// Start a server whose responder sees the whole [`RecordedRequest`] and
/// returns a [`MockResponse`] (status + response **headers** + body). This is
/// what lets a fixture branch on the request's `Authorization` and emit
/// `WWW-Authenticate` / `Docker-Content-Digest` / `X-Checksum-Sha256`.
pub fn start_with<F>(responder: F) -> Self
where
F: Fn(&RecordedRequest) -> MockResponse + Send + Sync + 'static,
{
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let addr = listener.local_addr().unwrap();
let requests = Arc::new(Mutex::new(Vec::new()));
let shutdown = Arc::new(AtomicBool::new(false));
let responder = Arc::new(responder);
let req_log = requests.clone();
let stop = shutdown.clone();
// ROOT-LAW-#0: the loopback accept-loop is a long-lived background IO
// thread — route it through gatling's ONE sanctioned background spawn
// (`Job`), signalled via `shutdown` and joined on drop below, instead of
// a bare std::thread::spawn.
let handle = gatling::background::Job::spawn(move || {
while !stop.load(Ordering::Relaxed) {
match listener.accept() {
Ok((mut sock, _)) => {
sock.set_read_timeout(Some(Duration::from_secs(2))).ok();
let mut buf = Vec::new();
let mut tmp = [0u8; 1024];
loop {
match sock.read(&mut tmp) {
Ok(0) => break,
Ok(n) => {
buf.extend_from_slice(&tmp[..n]);
if buf.windows(4).any(|w| w == b"\r\n\r\n") {
break;
}
}
Err(_) => break,
}
}
let text = String::from_utf8_lossy(&buf);
let mut lines = text.split("\r\n");
let request_line = lines.next().unwrap_or("");
let mut parts = request_line.split_whitespace();
let method = parts.next().unwrap_or("").to_string();
let path = parts.next().unwrap_or("").to_string();
let mut headers = HashMap::new();
for line in lines {
if line.is_empty() {
break;
}
if let Some((k, v)) = line.split_once(':') {
headers.insert(k.trim().to_lowercase(), v.trim().to_string());
}
}
let recorded = RecordedRequest {
method: method.clone(),
path: path.clone(),
headers,
};
let mock = responder(&recorded);
req_log.lock().unwrap().push(recorded);
let reason = match mock.status {
200 => "OK",
401 => "Unauthorized",
404 => "Not Found",
_ => "Status",
};
let send_body = method != "HEAD";
let mut head = format!(
"HTTP/1.1 {} {reason}\r\nContent-Length: {}\r\nConnection: close\r\n",
mock.status,
mock.body.len()
);
for (k, v) in &mock.headers {
head.push_str(&format!("{k}: {v}\r\n"));
}
head.push_str("\r\n");
let mut resp = head.into_bytes();
if send_body {
resp.extend_from_slice(&mock.body);
}
let _ = sock.write_all(&resp);
let _ = sock.flush();
}
Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
std::thread::sleep(Duration::from_millis(2));
}
Err(_) => break,
}
}
});
MockServer {
base_url: format!("http://{addr}"),
requests,
shutdown,
handle: Some(handle),
}
}
pub fn requests(&self) -> Vec<RecordedRequest> {
self.requests.lock().unwrap().clone()
}
pub fn request_count(&self) -> usize {
self.requests.lock().unwrap().len()
}
}
impl Drop for MockServer {
fn drop(&mut self) {
self.shutdown.store(true, Ordering::Relaxed);
if let Some(h) = self.handle.take() {
let _ = h.join();
}
}
}