From eaa10b1d1afddceef71beffd422dcc12b55b3640 Mon Sep 17 00:00:00 2001 From: Matthew Jackson <1085847+MattJackson@users.noreply.github.com> Date: Wed, 14 Oct 2026 03:54:14 -0700 Subject: [PATCH 1/5] plane black box: one runner boots the real binary against every linked door's case files A plane door crate states its black-box cases beside its root cells under the same `links` name (`cargo::metadata=blackbox=`); the root's build.rs resolves them identically for a path dependency and a pinned one (DOOR_BLACKBOX) and states BUSBAR_LINKED_AXES as data. tests/plane_blackbox.rs reads every case (schema blackbox/1: requires, files, vars, unset, credentials, a scripted JSON-RPC far end, steps validate | boot | session | call | admin | mint | upstream_set | upstream_counts), boots CARGO_BIN_EXE_busbar and holds every answer to the case. The Rust knows JSON, HTTP/1.1 and JSON-RPC 2.0 framing only; it names no plane, dialect or instance. The rigs the line-served and integrity suites carried move to tests/common (credentials, lines, upstream) and both suites use them. REDs: a malformed case is refused naming the plane crate and the case file; a mismatch fails naming the plane crate, the case and the step; a black-box directory with no case fails; the far end answers what the case states and counts it. The a2a door states its directory with one seed case. Ledger: busbar auth 49 -> 47, plane 772 -> 771, transport 200 -> 195. --- crates/busbar-plane-a2a/build.rs | 9 + .../blackbox/binary-answers-version.json | 9 + crates/busbar/build.rs | 27 +- crates/busbar/tests/common/credentials.rs | 193 +++ crates/busbar/tests/common/lines.rs | 122 ++ crates/busbar/tests/common/mod.rs | 3 + crates/busbar/tests/common/upstream.rs | 188 +++ crates/busbar/tests/mcp_stdio_serve.rs | 239 +-- crates/busbar/tests/plane_blackbox.rs | 1429 +++++++++++++++++ crates/busbar/tests/qa_mcp_integrity.rs | 71 +- qa/kind-isolation.toml | 12 +- 11 files changed, 2009 insertions(+), 293 deletions(-) create mode 100644 crates/busbar-plane-a2a/tests/blackbox/binary-answers-version.json create mode 100644 crates/busbar/tests/common/credentials.rs create mode 100644 crates/busbar/tests/common/lines.rs create mode 100644 crates/busbar/tests/common/upstream.rs create mode 100644 crates/busbar/tests/plane_blackbox.rs diff --git a/crates/busbar-plane-a2a/build.rs b/crates/busbar-plane-a2a/build.rs index aa661b03c9..336ab36049 100644 --- a/crates/busbar-plane-a2a/build.rs +++ b/crates/busbar-plane-a2a/build.rs @@ -54,6 +54,15 @@ fn main() -> Result<(), Box> { .join("tests/root/mod.rs") .display() ); + // THE BLACK-BOX CASES' ADDRESS (the same `links` row): the root's black-box runner boots the + // real binary against every case file here, reading it as `DEP_BUSBAR_DOOR_CELLS__BLACKBOX`. + println!( + "cargo::metadata=blackbox={}", + PathBuf::from(std::env::var("CARGO_MANIFEST_DIR")?) + .join("tests/blackbox") + .display() + ); + println!("cargo:rerun-if-changed=tests/blackbox"); let out_dir = PathBuf::from(std::env::var("OUT_DIR")?); let descriptor_path = out_dir.join("a2a-descriptor.bin"); let mut config = prost_build::Config::new(); diff --git a/crates/busbar-plane-a2a/tests/blackbox/binary-answers-version.json b/crates/busbar-plane-a2a/tests/blackbox/binary-answers-version.json new file mode 100644 index 0000000000..e2e38c57be --- /dev/null +++ b/crates/busbar-plane-a2a/tests/blackbox/binary-answers-version.json @@ -0,0 +1,9 @@ +{ + "schema": "blackbox/1", + "case": "the binary carrying this door answers --version", + "_comment": "The seed case that proves the root resolves this crate's black-box directory; the plane's served cases are written beside it.", + "requires": {"sections": ["agents"]}, + "steps": [ + {"validate": {"args": ["--version"], "exit": 0, "stdout_contains": ["busbar "]}} + ] +} diff --git a/crates/busbar/build.rs b/crates/busbar/build.rs index 77fb07f5e5..1f31b0f701 100644 --- a/crates/busbar/build.rs +++ b/crates/busbar/build.rs @@ -136,6 +136,12 @@ fn main() { for cfg in linked_axis_cfgs(&manifest, &enabled).1 { println!("cargo::rustc-check-cfg=cfg({cfg})"); } + // `BUSBAR_LINKED_AXES`: the same set as data, space-separated, so a black-box case's `requires` + // reads it at run time rather than a test spelling a `cfg` per axis. + println!( + "cargo:rustc-env=BUSBAR_LINKED_AXES={}", + linked_axis_cfgs(&manifest, &enabled).0.join(" ") + ); for cfg in linked_axis_cfgs(&manifest, &enabled).0 { println!("cargo::rustc-cfg={cfg}"); } @@ -419,6 +425,7 @@ fn door_cells_source(manifest: &str, enabled: &dyn Fn(&str) -> bool) -> (String, let mut cells: Vec = Vec::new(); let mut missing: Vec = Vec::new(); let mut files: Vec = Vec::new(); + let mut blackbox: Vec = Vec::new(); for (feature, krate) in metadata_map(manifest, "package.metadata.busbar.linked") { let is_door = axes .iter() @@ -426,6 +433,17 @@ fn door_cells_source(manifest: &str, enabled: &dyn Fn(&str) -> bool) -> (String, if !(enabled(&feature) && is_door) { continue; } + // THE BLACK-BOX CASES the crate states beside its cells (`blackbox=`, the same + // `links` name): read for every door, cells or none, so the runner sees each. + let box_var = format!( + "DEP_{}_BLACKBOX", + door_cells_links(&krate).to_uppercase().replace('-', "_") + ); + println!("cargo:rerun-if-env-changed={box_var}"); + if let Ok(dir) = env::var(&box_var) { + println!("cargo:rerun-if-changed={dir}"); + blackbox.push(format!(" ({krate:?}, {dir:?}),\n")); + } let var = format!( "DEP_{}_CELLS", door_cells_links(&krate).to_uppercase().replace('-', "_") @@ -472,8 +490,13 @@ fn door_cells_source(manifest: &str, enabled: &dyn Fn(&str) -> bool) -> (String, /// Every door plane's root cells: the path the ledger cites them by (the crate's census\n\ /// mount, `crates//tests/root/mod.rs`) and the file its crate states.\n\ #[allow(dead_code)]\n\ - static DOOR_CELLS_FILES: &[(&str, &str)] = &[\n{}];\n", - files.concat() + static DOOR_CELLS_FILES: &[(&str, &str)] = &[\n{}];\n\ + /// Every door plane's black-box case directory (`blackbox=` under its `links` name):\n\ + /// its crate and the directory its crate states.\n\ + #[allow(dead_code)]\n\ + static DOOR_BLACKBOX: &[(&str, &str)] = &[\n{}];\n", + files.concat(), + blackbox.concat() ); (out, files) } diff --git a/crates/busbar/tests/common/credentials.rs b/crates/busbar/tests/common/credentials.rs new file mode 100644 index 0000000000..7755d2c058 --- /dev/null +++ b/crates/busbar/tests/common/credentials.rs @@ -0,0 +1,193 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (C) 2026 Busbar Inc and contributors + +//! THE CREDENTIAL RIGS the black-box suites share: a local token issuer whose JWKS is served over a +//! certificate-verified loopback TLS endpoint, the signed token-verifying `kind: auth` plugin busbar +//! builds packaged into a plugins dir, the first-party key a `test-harness` child takes, and the +//! busbar-key re-signing an operator's audience binding performs. Moved here from the suites that +//! first wrote them, so every suite (and the plane black-box runner) drives the same rigs. + +use std::path::Path; + +/// THE LOCAL ISSUER, one per test process (the loader's `test_issuer`: an ES256 key, its JWKS and +/// genuinely signed tokens), its JWKS served over a certificate-verified loopback TLS endpoint +/// ([`serve_jwks`]) a module trusts through `ca_cert_pem`. The child `busbar` process's OWN +/// connector fetches the JWKS over the dropped-in module's declared need. +pub fn issuer() -> &'static busbar_plugin_loader::test_issuer::Issuer { + static ONE: std::sync::OnceLock = + std::sync::OnceLock::new(); + ONE.get_or_init(|| { + let issuer = busbar_plugin_loader::test_issuer::Issuer::start( + "https://issuer.e2e.invalid", + "e2e-issuer", + ); + let (url, cert_pem) = serve_jwks(issuer.jwks().to_string()); + issuer.served_at(&url, &cert_pem) + }) +} + +/// Serve `jwks` over TLS on a fresh loopback port for the life of the process, under a self-signed +/// certificate for `127.0.0.1`: the JWKS URL and the certificate's PEM. One background thread +/// answers every request with the document. +pub fn serve_jwks(jwks: String) -> (String, String) { + use std::io::{Read as _, Write as _}; + let cert = rcgen::generate_simple_self_signed(vec!["127.0.0.1".to_string()]) + .expect("mint a self-signed certificate"); + let cert_pem = cert.cert.pem(); + let private = rustls_pki_types::PrivateKeyDer::Pkcs8( + rustls_pki_types::PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der()), + ); + let config = std::sync::Arc::new( + rustls::ServerConfig::builder_with_provider(std::sync::Arc::new( + rustls::crypto::ring::default_provider(), + )) + .with_safe_default_protocol_versions() + .expect("protocol versions") + .with_no_client_auth() + .with_single_cert(vec![cert.cert.der().clone()], private) + .expect("server certificate"), + ); + let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind a loopback port"); + let url = format!( + "https://{}/jwks", + listener.local_addr().expect("local addr") + ); + std::thread::spawn(move || { + for socket in listener.incoming() { + let (Ok(socket), Ok(session)) = (socket, rustls::ServerConnection::new(config.clone())) + else { + continue; + }; + let mut stream = rustls::StreamOwned::new(session, socket); + let _ = stream.read(&mut [0u8; 4096]); + let _ = write!( + stream, + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\ + Connection: close\r\n\r\n{jwks}", + jwks.len() + ); + let _ = stream.flush(); + stream.conn.send_close_notify(); + let _ = stream.flush(); + } + }); + (url, cert_pem) +} + +/// A JWT for principal `sub` with `roles`, bound to `aud` and signed by the local issuer. +pub fn token_for(sub: &str, roles: &[&str], aud: &str) -> String { + issuer().mint(sub, roles, aud) +} + +/// The issuer's module settings for `audience`, as YAML mapping lines indented by `indent` spaces. +pub fn issuer_settings_yaml(audience: &str, indent: usize) -> String { + issuer() + .settings(audience) + .into_iter() + .map(|(k, v)| format!("{}{k}: {v}\n", " ".repeat(indent))) + .collect() +} + +/// The variable a `test-harness` child reads its first-party key from: the binary's +/// `root::boot::HARNESS_FIRST_PARTY_KEY_ENV`, whose unit tests pin this spelling. +pub const HARNESS_FIRST_PARTY_KEY_ENV: &str = "BUSBAR_TEST_HARNESS_FIRST_PARTY_KEY"; + +/// The first-party key a suite signs a dropped-in module with and hands the child through the +/// harness seam ([`HARNESS_FIRST_PARTY_KEY_ENV`]). +pub fn harness_release_key() -> busbar_plugin_loader::sign::SigningKey { + super::plugins::key(0x5a) +} + +/// The harness key's public half, hex, as the child reads it from [`HARNESS_FIRST_PARTY_KEY_ENV`]. +pub fn harness_release_key_hex() -> String { + hex::encode(harness_release_key().verifying_key().as_bytes()) +} + +/// Package the REAL token-verifying `kind: auth` plugin cdylib busbar builds (a pinned git +/// dev-dependency, so the build leaves it under `deps/`) into a `kind: auth` tarball named `module` +/// (alias `alias`) in `plugins_dir`, its manifest stating the door's Statement as the pack tool +/// renders it, SIGNED first-party under [`harness_release_key`] when `first_party` (its needs are in +/// the `operator-infrastructure` egress class, which the host grants a first-party plugin only), +/// else under a third-party key. A cdylib that is not built fails the caller in every run. +pub fn install_auth_plugin(plugins_dir: &Path, module: &str, alias: &str, first_party: bool) { + let lib = super::plugins::token_verifier_cdylib_path() + .and_then(|p| std::fs::read(p).ok()) + .unwrap_or_else(|| { + super::plugins::missing( + "pinned token-verifying `kind: auth` plugin's", + super::plugins::BUILD_PINNED, + ) + }); + let publisher = if first_party { + busbar_plugin_loader::sign::FIRST_PARTY_PUBLISHER + } else { + "third-party" + }; + let mut m = super::plugins::manifest("auth", module, publisher); + m.alias = alias.into(); + // One staging file PER CALL: suites run as threads of one process, and rewriting a library + // another thread still has mapped (dlopened for its Statement) truncates it under that mapping. + static STAGED: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0); + let path = std::env::temp_dir().join(format!( + "busbar-auth-module-{}-{}{}", + std::process::id(), + STAGED.fetch_add(1, std::sync::atomic::Ordering::Relaxed), + std::env::consts::DLL_SUFFIX + )); + std::fs::write(&path, &lib).expect("stage the library"); + m.statement = busbar_plugin_loader::dispatch::rendering_of_library(&path) + .expect("the token-verifying cdylib states its door") + .map(hex::encode); + let _ = std::fs::remove_file(&path); + let bytes = if first_party { + super::plugins::signed(&harness_release_key(), m, &lib) + } else { + super::plugins::signed(&super::plugins::key(0x33), m, &lib) + }; + std::fs::create_dir_all(plugins_dir).expect("the plugins dir"); + std::fs::write(plugins_dir.join(format!("{module}.tar.gz")), bytes).expect("write the tarball"); +} + +/// A fresh signing key from the binary itself (`--generate-signing-key`): its 32 bytes and its hex. +pub fn generate_signing_key() -> ([u8; 32], String) { + let out = std::process::Command::new(super::boot::exe()) + .arg("--generate-signing-key") + .output() + .expect("run busbar --generate-signing-key"); + assert!(out.status.success(), "--generate-signing-key failed"); + let hex_key = String::from_utf8(out.stdout) + .expect("the key is text") + .trim() + .to_string(); + let bytes: [u8; 32] = hex::decode(&hex_key) + .expect("the signing key is hex") + .try_into() + .expect("the signing key is 32 bytes"); + (bytes, hex_key) +} + +/// Bind a minted busbar key (`bbk_.`) to `audience`: its claims plus `a`, re-signed +/// with the node's signing key — what an operator's audience binding performs. +pub fn bind_key(secret: &[u8; 32], plain: &str, audience: &str) -> String { + use base64::Engine as _; + use ed25519_dalek::Signer as _; + let b64 = base64::engine::general_purpose::URL_SAFE_NO_PAD; + let body = plain + .strip_prefix("bbk_") + .unwrap_or_else(|| panic!("not a busbar key: {plain}")); + let payload = body.split('.').next().expect("a payload"); + let claims: serde_json::Value = + serde_json::from_slice(&b64.decode(payload).expect("base64 claims")).expect("JSON claims"); + let mut bound = serde_json::json!({ + "sub": claims["sub"], + "exp": claims["exp"], + "kid": claims["kid"], + "a": audience, + }); + if let Some(g) = claims.get("g") { + bound["g"] = g.clone(); + } + let raw = serde_json::to_vec(&bound).expect("claims serialize"); + let sig = ed25519_dalek::SigningKey::from_bytes(secret).sign(&raw); + format!("bbk_{}.{}", b64.encode(&raw), b64.encode(sig.to_bytes())) +} diff --git a/crates/busbar/tests/common/lines.rs b/crates/busbar/tests/common/lines.rs new file mode 100644 index 0000000000..5db49b3fa2 --- /dev/null +++ b/crates/busbar/tests/common/lines.rs @@ -0,0 +1,122 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (C) 2026 Busbar Inc and contributors + +//! THE LINE-SERVED CHILD the black-box suites drive: the real binary spawned as somebody's child +//! process, its stdin written line by line and its stdout read line by line, every stdout line +//! required to be ONE JSON message (the line carrier writes nothing else), and EOF on stdin its +//! shutdown. Moved here from the suite that first wrote it. + +use std::io::{BufRead, BufReader, Write}; +use std::process::{Child, Command, Stdio}; +use std::sync::mpsc; +use std::time::{Duration, Instant}; + +/// The child, its pipes, and a line-reader thread per output stream. +pub struct LineChild { + pub child: Child, + pub stdin: Option, + pub stdout: mpsc::Receiver, + pub stderr: mpsc::Receiver, +} + +/// Spawn `cmd` with all three pipes, a reader thread on each output. +pub fn spawn(mut cmd: Command) -> LineChild { + cmd.stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()); + let mut child = cmd.spawn().expect("spawn the line-served child"); + let stdin = child.stdin.take(); + let (out_tx, out_rx) = mpsc::channel(); + let stdout = child.stdout.take().expect("stdout piped"); + std::thread::spawn(move || { + for line in BufReader::new(stdout).lines().map_while(Result::ok) { + let _ = out_tx.send(line); + } + }); + let (err_tx, err_rx) = mpsc::channel(); + let stderr = child.stderr.take().expect("stderr piped"); + std::thread::spawn(move || { + for line in BufReader::new(stderr).lines().map_while(Result::ok) { + let _ = err_tx.send(line); + } + }); + LineChild { + child, + stdin, + stdout: out_rx, + stderr: err_rx, + } +} + +impl LineChild { + /// Write `value` as one line. + pub fn send(&mut self, value: &serde_json::Value) { + let stdin = self.stdin.as_mut().expect("stdin still open"); + let mut bytes = serde_json::to_vec(value).expect("the line serializes"); + bytes.push(b'\n'); + stdin.write_all(&bytes).expect("write a line"); + stdin.flush().expect("flush a line"); + } + + /// The next stdout line, which MUST be one JSON message. A hang detector, not a latency + /// assertion: a debug child on a saturated host can take most of a minute to answer. + pub fn recv(&self) -> serde_json::Value { + let line = self + .stdout + .recv_timeout(Duration::from_secs(120)) + .expect("the child must answer within the bound"); + serde_json::from_str(&line).unwrap_or_else(|e| { + panic!("every stdout line must be one JSON message ({e}): {line:?}") + }) + } + + /// [`LineChild::recv`] within `bound`, its failure a sentence rather than a panic. + pub fn try_recv(&self, bound: Duration) -> Result { + let line = self + .stdout + .recv_timeout(bound) + .map_err(|_| format!("the child did not answer within {bound:?}"))?; + serde_json::from_str(&line) + .map_err(|e| format!("every stdout line must be one JSON message ({e}): {line:?}")) + } + + /// EOF on stdin, then the exit code — bounded, so a child that ignores EOF fails the caller. + pub fn eof_and_wait(mut self, bound: Duration) -> i32 { + drop(self.stdin.take()); + wait_bounded(&mut self.child, bound) + } + + /// Everything stderr has said so far, drained with a short grace per line. + pub fn stderr_so_far(&self) -> String { + let mut all = String::new(); + while let Ok(line) = self.stderr.recv_timeout(Duration::from_millis(300)) { + all.push_str(&line); + all.push('\n'); + } + all + } +} + +/// The child's exit code within `bound`; a child still running then is killed and the caller fails. +pub fn wait_bounded(child: &mut Child, bound: Duration) -> i32 { + try_wait_bounded(child, bound).unwrap_or_else(|e| panic!("{e}")) +} + +/// [`wait_bounded`], its failure a sentence rather than a panic. +pub fn try_wait_bounded(child: &mut Child, bound: Duration) -> Result { + let deadline = Instant::now() + bound; + loop { + if let Some(status) = child + .try_wait() + .map_err(|e| format!("cannot wait on the child: {e}"))? + { + return Ok(status.code().unwrap_or(-1)); + } + if Instant::now() >= deadline { + let _ = child.kill(); + let _ = child.wait(); + return Err(format!("the child did not exit within {bound:?}")); + } + std::thread::sleep(Duration::from_millis(50)); + } +} diff --git a/crates/busbar/tests/common/mod.rs b/crates/busbar/tests/common/mod.rs index 7775024c67..a5d9eeb043 100644 --- a/crates/busbar/tests/common/mod.rs +++ b/crates/busbar/tests/common/mod.rs @@ -24,8 +24,11 @@ #![allow(dead_code)] pub mod boot; +pub mod credentials; +pub mod lines; pub mod otlp; pub mod plugins; +pub mod upstream; use std::path::{Path, PathBuf}; diff --git a/crates/busbar/tests/common/upstream.rs b/crates/busbar/tests/common/upstream.rs new file mode 100644 index 0000000000..b1ccf76220 --- /dev/null +++ b/crates/busbar/tests/common/upstream.rs @@ -0,0 +1,188 @@ +// SPDX-License-Identifier: Apache-2.0 +// Copyright (C) 2026 Busbar Inc and contributors + +//! A FAR END ON LOOPBACK: an HTTP/1.1 server that reads one request per connection and answers it, +//! and on it a SCRIPTED JSON-RPC 2.0 RESPONDER whose methods are DATA — each method name the case +//! states answers a stated `result`, or echoes a value from the request into a stated result shape. +//! The responder knows JSON-RPC framing (an `id`, a `method`, a `result`/`error`) and nothing of any +//! plane's methods; every method it serves, and every answer, is the caller's. Each method name is +//! counted as received, and a method's answer can be replaced between steps. The HTTP primitives +//! moved here from the suite that first wrote them. + +use std::collections::BTreeMap; +use std::io::{Read as _, Write as _}; +use std::net::{TcpListener, TcpStream}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use serde_json::{json, Value}; + +/// Read one HTTP/1.1 request off `conn`: the head and a `content-length` body. `None` on a +/// connection that closed or sent something that is not a request. +pub fn read_request(conn: &mut TcpStream) -> Option<(String, Vec)> { + conn.set_read_timeout(Some(Duration::from_secs(20))).ok()?; + let mut buf = Vec::new(); + let mut chunk = [0u8; 4096]; + let head_end = loop { + if let Some(i) = buf.windows(4).position(|w| w == b"\r\n\r\n") { + break i + 4; + } + let n = conn.read(&mut chunk).ok()?; + if n == 0 { + return None; + } + buf.extend_from_slice(&chunk[..n]); + }; + let head = String::from_utf8_lossy(&buf[..head_end]).to_string(); + let len = head + .lines() + .find_map(|l| { + let (k, v) = l.split_once(':')?; + k.trim() + .eq_ignore_ascii_case("content-length") + .then(|| v.trim().parse::().ok()) + .flatten() + }) + .unwrap_or(0); + let mut body = buf[head_end..].to_vec(); + while body.len() < len { + let n = conn.read(&mut chunk).ok()?; + if n == 0 { + break; + } + body.extend_from_slice(&chunk[..n]); + } + Some((head, body)) +} + +/// Answer `conn` with `status` and the JSON `body`, closing the connection. +pub fn respond(conn: &mut TcpStream, status: u16, body: &Value) { + let raw = body.to_string(); + let _ = write!( + conn, + "HTTP/1.1 {status} OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\n\ + connection: close\r\n\r\n{raw}", + raw.len() + ); + let _ = conn.flush(); +} + +/// How one method answers. +#[derive(Debug, Clone, PartialEq)] +pub enum Answer { + /// This `result`, as stated. + Result(Value), + /// The value at `from` in the request, rendered through `format` (its `{}` replaced; the value + /// itself when absent) and written at `into` (a pointer under `/result`) of `base` (the result's + /// shape; an empty object when absent). + Echo { + from: String, + into: String, + base: Value, + format: Option, + }, +} + +/// What the responder serves right now, and every method name it has received, counted. +#[derive(Debug, Default)] +pub struct Script { + pub methods: BTreeMap, + pub received: BTreeMap, +} + +/// A scripted JSON-RPC responder on a loopback port. +pub struct Responder { + pub port: u16, + pub script: Arc>, +} + +impl Responder { + /// Start serving `methods` on a fresh loopback port, one thread per connection. + pub fn start(methods: BTreeMap) -> Responder { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind the responder"); + let port = listener.local_addr().expect("its address").port(); + let script = Arc::new(Mutex::new(Script { + methods, + received: BTreeMap::new(), + })); + let state = Arc::clone(&script); + std::thread::spawn(move || { + for conn in listener.incoming().flatten() { + let state = Arc::clone(&state); + std::thread::spawn(move || serve_one(conn, &state)); + } + }); + Responder { port, script } + } + + /// Replace `method`'s answer. + pub fn set(&self, method: &str, answer: Answer) { + self.script + .lock() + .expect("the script") + .methods + .insert(method.to_string(), answer); + } + + /// Every method name received so far, counted. + pub fn received(&self) -> BTreeMap { + self.script.lock().expect("the script").received.clone() + } +} + +/// `answer` for the request `req`. +fn render(answer: &Answer, req: &Value) -> Value { + match answer { + Answer::Result(result) => result.clone(), + Answer::Echo { + from, + into, + base, + format, + } => { + let picked = req.pointer(from).cloned().unwrap_or(Value::Null); + let value = match format { + Some(f) => { + let text = picked + .as_str() + .map_or_else(|| picked.to_string(), str::to_string); + Value::String(f.replacen("{}", &text, 1)) + } + None => picked, + }; + let mut whole = + json!({ "result": if base.is_null() { json!({}) } else { base.clone() } }); + if let Some(slot) = whole.pointer_mut(into) { + *slot = value; + } + whole["result"].clone() + } + } +} + +fn serve_one(mut conn: TcpStream, state: &Mutex