From 0c1e3a418d0c7108d0b3a855f4b1483bebcb3a86 Mon Sep 17 00:00:00 2001 From: Lance Tuller Date: Mon, 29 Jun 2026 09:57:34 -0400 Subject: [PATCH] test(mcp): live read/write/negative MCP e2e in the netbox-integration lane MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wire the local MCP write path into CI so it can't silently regress. Extract the stdio JSON-RPC ServeChild harness into tests/support/serve.rs (now shared by the offline mcp_serve_tests and the new live tests — no duplication), and add three #[ignore] live tests to it_netbox.rs: - mcp_read_returns_live_data — handshake + nbox_status + nbox_get device. - mcp_local_write_reserves_ip_and_verifies — `nbox serve --local-writes` plans and applies an IP reservation under the profile token, then reads the new IP back to confirm the mutation landed. - mcp_write_refused_without_local_writes — a plain stdio server refuses the write. The netbox-integration 4.2 and 4.6 lanes already run `cargo test --test it_netbox -- --ignored`, so they pick these up with no workflow change. The full live suite (25 tests) and the offline mcp_serve_tests both pass locally. --- tests/it_netbox.rs | 109 ++++++++++++++- tests/mcp_serve_tests.rs | 280 +++++---------------------------------- tests/support/mod.rs | 1 + tests/support/serve.rs | 206 ++++++++++++++++++++++++++++ 4 files changed, 347 insertions(+), 249 deletions(-) create mode 100644 tests/support/serve.rs diff --git a/tests/it_netbox.rs b/tests/it_netbox.rs index bb4f59c..64b211b 100644 --- a/tests/it_netbox.rs +++ b/tests/it_netbox.rs @@ -26,9 +26,13 @@ use std::io::Write; use std::process::{Command, Output}; -use serde_json::Value; +use serde_json::{Value, json}; use tempfile::NamedTempFile; +mod support; + +use support::serve::{ServeChild, tool_error, tool_payload}; + /// The seeded API token's default — kept in sync with `docker-compose.yml`'s /// `SUPERUSER_API_TOKEN` and `seed.py`. `NETBOX_TOKEN` overrides it. const DEFAULT_TOKEN: &str = "0123456789abcdef0123456789abcdef0fedcba9"; @@ -43,6 +47,16 @@ fn netbox_token() -> String { std::env::var("NETBOX_TOKEN").unwrap_or_else(|_| DEFAULT_TOKEN.to_string()) } +/// A serve config (TOML string) pointing at the live NetBox. The token rides in +/// via `NBOX_TOKEN` (the harness sets it), so `token_env` is just a placeholder. +fn serve_config_toml() -> String { + format!( + "active_profile = \"ci\"\n\n[profiles.ci]\nurl = \"{url}\"\n\ + token_env = \"NETBOX_TOKEN_UNUSED\"\nverify_tls = false\n", + url = netbox_url(), + ) +} + /// A throwaway config file holding one profile that points at the live NetBox. /// `page_size` is configurable so the pagination test can force multiple pages. /// REST is the canonical backend; tests that need GraphQL pass an optional @@ -764,3 +778,96 @@ fn next_prefix_length_too_large_is_graceful_empty() { "no /24 fits in a /28 — expected empty available: {v}" ); } + +// ===== live MCP server end-to-end (read / local write / negative) =========== +// +// Drive the real `nbox serve` stdio transport against the seeded live NetBox via +// JSON-RPC (the `support::serve` harness), covering ADR-0002 local single-user +// writes end to end: a read, a write that actually mutates NetBox and is +// verified, and the negative (the write is refused without the opt-in). + +#[test] +#[ignore = "requires a live NetBox (netbox-integration workflow)"] +fn mcp_read_returns_live_data() { + let mut server = ServeChild::spawn(&serve_config_toml(), &[], &netbox_token()); + let init = server.handshake(); + assert_eq!( + init["result"]["protocolVersion"], "2025-11-25", + "initialize: {init}" + ); + + // nbox_status: a live read confirming the connection (no error payload). + let _status = tool_payload(&server.call(2, "nbox_status", json!({}))); + + // nbox_get device: the seeded ci-dev1 resolves with its (active) status. + let dev = + tool_payload(&server.call(3, "nbox_get", json!({"kind": "device", "ref": "ci-dev1"}))); + assert_eq!(dev["status"], "active", "device read: {dev}"); + + server.shutdown(); +} + +#[test] +#[ignore = "requires a live NetBox (netbox-integration workflow)"] +fn mcp_local_write_reserves_ip_and_verifies() { + let mut server = ServeChild::spawn(&serve_config_toml(), &["--local-writes"], &netbox_token()); + server.handshake(); + + // Plan an IP reservation in a seeded child prefix. + let plan = tool_payload(&server.call( + 2, + "nbox_plan_write", + json!({"operation": {"kind": "ip_reserve", "prefix": "10.10.1.0/24"}}), + )); + assert_eq!(plan["operation"], "allocate", "plan: {plan}"); + assert!( + !plan["confirm_token"] + .as_str() + .unwrap_or_default() + .is_empty(), + "plan carries a confirm token: {plan}" + ); + + // Apply it — a real POST to NetBox under the profile token (local_writes). + let receipt = tool_payload(&server.call(3, "nbox_apply_write", json!({"plan": plan}))); + assert_eq!(receipt["applied"], true, "receipt: {receipt}"); + assert_eq!(receipt["status"], 201, "receipt: {receipt}"); + let address = receipt["object"]["address"] + .as_str() + .unwrap_or_else(|| panic!("receipt has no created address: {receipt}")); + assert!( + address.starts_with("10.10.1."), + "reserved address should be in the prefix: {address}" + ); + + // Verify the mutation landed: read the new IP back through MCP. + let host = address.split('/').next().unwrap_or(address); + let ip = tool_payload(&server.call(4, "nbox_get", json!({"kind": "ip", "ref": host}))); + assert_eq!( + ip["address"].as_str().unwrap_or("").split('/').next(), + Some(host), + "the reserved IP must now exist in NetBox: {ip}" + ); + + server.shutdown(); +} + +#[test] +#[ignore = "requires a live NetBox (netbox-integration workflow)"] +fn mcp_write_refused_without_local_writes() { + // The opt-in actually gates: a plain stdio server (no --local-writes) must + // refuse a write with a message pointing at local_writes. + let mut server = ServeChild::spawn(&serve_config_toml(), &[], &netbox_token()); + server.handshake(); + let msg = server.call( + 2, + "nbox_plan_write", + json!({"operation": {"kind": "ip_reserve", "prefix": "10.10.1.0/24"}}), + ); + let err = tool_error(&msg).unwrap_or_else(|| panic!("write should be refused, got: {msg}")); + assert!( + err.contains("local_writes") || err.contains("not enabled"), + "refusal should point at local_writes: {err}" + ); + server.shutdown(); +} diff --git a/tests/mcp_serve_tests.rs b/tests/mcp_serve_tests.rs index 7ec78e3..c18b961 100644 --- a/tests/mcp_serve_tests.rs +++ b/tests/mcp_serve_tests.rs @@ -1,8 +1,8 @@ //! End-to-end test of `nbox serve` — the MCP server (read-only by default) over stdio. //! //! This launches the *real* compiled binary (`env!("CARGO_BIN_EXE_nbox")`) with -//! `serve`, then speaks newline-delimited JSON-RPC over its stdin/stdout. It -//! proves three things at once: +//! `serve`, then speaks newline-delimited JSON-RPC over its stdin/stdout (via the +//! shared `support::serve::ServeChild` harness). It proves three things at once: //! //! 1. `nbox serve` is wired correctly: `connect()` builds a client from a //! minimal profile + a dummy token without hitting the network, and the MCP @@ -12,33 +12,21 @@ //! expected tool set. //! 3. The stdio invariant holds: every line the child writes to stdout is valid //! JSON-RPC (nothing — banners, logs, prompts — leaks onto stdout). -//! -//! Determinism / no-hang: the handshake is driven entirely off reading the -//! responses we expect (matched by JSON-RPC `id`), never off timing. A dedicated -//! reader thread streams stdout lines onto a channel; every read is bounded by -//! `recv_timeout`, so a missing/garbled response fails the test fast instead of -//! hanging. stdin is closed to signal EOF, and the child is waited on (then -//! killed as a backstop) so no process is ever left behind. -use std::io::{BufRead, BufReader, Write}; -use std::process::{Child, Command, Stdio}; -use std::sync::mpsc::{Receiver, RecvTimeoutError, channel}; -use std::time::Duration; +mod support; use serde_json::{Value, json}; -use tempfile::NamedTempFile; +use support::serve::{PROTOCOL_VERSION, ServeChild, tool_payload}; use wiremock::matchers::{header, method, path}; use wiremock::{Mock, MockServer, ResponseTemplate}; -/// The protocol version the server advertises (`ProtocolVersion::LATEST` in -/// `src/mcp/mod.rs`). The server negotiates down to the client's version if it -/// is older, so sending the latest is always accepted. -const PROTOCOL_VERSION: &str = "2025-11-25"; - -/// Every read off the child's stdout is bounded by this. The handshake is local -/// and offline, so responses arrive in milliseconds; this is a generous ceiling -/// that exists only so a broken server fails fast rather than hanging the suite. -const READ_TIMEOUT: Duration = Duration::from_secs(20); +/// Build a minimal serve config pointing at `url`. The token comes from the +/// `NBOX_TOKEN` override the harness sets, so `token_env` is just a placeholder. +fn serve_config(url: &str) -> String { + format!( + "active_profile = \"test\"\n\n[profiles.test]\nurl = \"{url}\"\ntoken_env = \"NBOX_TEST_TOKEN_UNUSED\"\n" + ) +} /// The exact tool set `nbox serve` must expose (see the `#[tool(...)]` adapters /// in `src/mcp/mod.rs`). Order-independent: we compare as sorted sets. @@ -58,153 +46,6 @@ const EXPECTED_TOOLS: &[&str] = &[ "nbox_apply_write", ]; -/// A live `nbox serve` child plus the plumbing to talk to it. Holds the child's -/// stdin (to send requests) and a channel of stdout lines (filled by a reader -/// thread). On drop, the child is killed and reaped as a backstop. -struct ServeChild { - child: Child, - // `Option` so `shutdown` can drop stdin (sending EOF) without leaving an - // invalid field behind. - stdin: Option, - lines: Receiver, - // Kept so the temp config file lives at least as long as the running child. - _config: NamedTempFile, -} - -impl ServeChild { - /// Spawn `nbox serve` against a throwaway config + dummy token. The profile - /// URL is unreachable on purpose: the initialize/tools/list handshake never - /// makes a network call, so the bogus URL is fine and keeps the test offline. - fn spawn() -> Self { - Self::spawn_with("http://127.0.0.1:1/", &[], "dummy") - } - - /// Spawn `nbox serve` against a caller-provided NetBox base URL and extra - /// serve args. Used by tests that need the real stdio transport plus a - /// wiremock NetBox. - fn spawn_with(url: &str, serve_args: &[&str], token: &str) -> Self { - let mut config = NamedTempFile::new().expect("create temp config"); - write!( - config, - "active_profile = \"test\"\n\ - \n\ - [profiles.test]\n\ - url = \"{url}\"\n\ - token_env = \"NBOX_TEST_TOKEN_UNUSED\"\n" - ) - .expect("write temp config"); - config.flush().expect("flush temp config"); - - let mut command = Command::new(env!("CARGO_BIN_EXE_nbox")); - command.arg("--config").arg(config.path()).arg("serve"); - command.args(serve_args); - let mut child = command - // The direct-override token env so `connect()` finds a token. Tests - // that hit wiremock can require this exact token on every request. - .env("NBOX_TOKEN", token) - // Pin logging quiet and to stderr regardless of the caller's env, so - // the stdout-cleanliness assertion isn't perturbed by NBOX_LOG/RUST_LOG. - .env_remove("NBOX_LOG") - .env_remove("RUST_LOG") - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .expect("spawn nbox serve"); - - let stdin = child.stdin.take().expect("child stdin"); - let stdout = child.stdout.take().expect("child stdout"); - - // Reader thread: stream stdout lines onto a channel so every read on the - // test side is bounded by recv_timeout (no blocking read can hang us). - let (tx, rx) = channel::(); - std::thread::spawn(move || { - let reader = BufReader::new(stdout); - for line in reader.lines() { - match line { - Ok(l) => { - if tx.send(l).is_err() { - break; // receiver gone — test finished - } - } - Err(_) => break, - } - } - }); - - ServeChild { - child, - stdin: Some(stdin), - lines: rx, - _config: config, - } - } - - /// Send one compact JSON value as a single newline-terminated line. - fn send(&mut self, msg: &Value) { - let line = serde_json::to_string(msg).expect("serialize message"); - // Exactly one compact JSON object per line — the stdio framing. - assert!(!line.contains('\n'), "framing must be one line per message"); - let stdin = self.stdin.as_mut().expect("child stdin is open"); - writeln!(stdin, "{line}").expect("write to child stdin"); - stdin.flush().expect("flush child stdin"); - } - - /// Read the next stdout line within the bounded timeout, asserting it parses - /// as JSON (the stdout-is-pure-JSON-RPC invariant), and return the value. - fn next_json(&self) -> Value { - match self.lines.recv_timeout(READ_TIMEOUT) { - Ok(line) => serde_json::from_str(&line) - .unwrap_or_else(|e| panic!("stdout line was not valid JSON: {e}\nline: {line:?}")), - Err(RecvTimeoutError::Timeout) => { - panic!("timed out waiting for a stdout line from `nbox serve`") - } - Err(RecvTimeoutError::Disconnected) => { - panic!("`nbox serve` closed stdout before sending the expected response") - } - } - } - - /// Read responses until one carries `id == want_id`. Notifications and other - /// ids are skipped; every consumed line is still validated as JSON by - /// `next_json`, so this also enforces stdout cleanliness as a side effect. - fn read_response(&self, want_id: i64) -> Value { - loop { - let v = self.next_json(); - if v.get("id").and_then(Value::as_i64) == Some(want_id) { - return v; - } - } - } - - /// Close stdin (EOF) and ensure the child has exited, killing it as a - /// backstop so the test can never leave a process behind. - fn shutdown(mut self) { - // Dropping stdin closes it → the server's stdio transport sees EOF and - // `serve` returns. - self.stdin = None; - - // Give the child a brief, bounded chance to exit cleanly on EOF; if it - // hasn't, kill it. Either way we reap it so there's no zombie. - for _ in 0..100 { - match self.child.try_wait() { - Ok(Some(_)) => return, - Ok(None) => std::thread::sleep(Duration::from_millis(20)), - Err(_) => break, - } - } - let _ = self.child.kill(); - let _ = self.child.wait(); - } -} - -impl Drop for ServeChild { - fn drop(&mut self) { - let _ = self.child.kill(); - let _ = self.child.wait(); - } -} - async fn mount_device_status_update_requiring_token(mock: &MockServer, token: &str) { let auth = format!("Bearer {token}"); Mock::given(method("GET")) @@ -284,29 +125,11 @@ async fn mount_device_status_update_requiring_token(mock: &MockServer, token: &s .await; } -/// Extract a successful tool's structured payload (the `Json` it returns). -fn tool_payload(msg: &Value) -> Value { - assert!(msg.get("error").is_none(), "tool returned an error: {msg}"); - let result = &msg["result"]; - assert_ne!( - result["isError"], - json!(true), - "tool execution error: {msg}" - ); - if let Some(sc) = result.get("structuredContent") - && !sc.is_null() - { - return sc.clone(); - } - let text = result["content"][0]["text"] - .as_str() - .unwrap_or_else(|| panic!("tool result has no structuredContent/text: {msg}")); - serde_json::from_str(text).unwrap_or_else(|_| panic!("tool result text not JSON: {text}")) -} - #[test] fn serve_handshake_lists_all_tools_with_clean_stdout() { - let mut server = ServeChild::spawn(); + // The profile URL is unreachable on purpose: the initialize/tools/list + // handshake never makes a network call, so the bogus URL keeps it offline. + let mut server = ServeChild::spawn(&serve_config("http://127.0.0.1:1/"), &[], "dummy"); // 1) initialize (id 1) with a valid protocolVersion, empty capabilities, and // a clientInfo. The server replies with the negotiated InitializeResult. @@ -326,28 +149,23 @@ fn serve_handshake_lists_all_tools_with_clean_stdout() { let result = init .get("result") .unwrap_or_else(|| panic!("initialize had no result: {init}")); - // serverInfo identifies the server (name/version present). let server_info = &result["serverInfo"]; assert!( server_info.get("name").and_then(Value::as_str).is_some(), "initialize result missing serverInfo.name: {init}" ); - // capabilities.tools must be advertised (the server enables tools). assert!( result["capabilities"].get("tools").is_some(), "initialize result missing capabilities.tools: {init}" ); - // capabilities.resources must be advertised (the server enables resources). assert!( result["capabilities"].get("resources").is_some(), "initialize result missing capabilities.resources: {init}" ); - // capabilities.prompts must be advertised (the server enables the prompt catalog). assert!( result["capabilities"].get("prompts").is_some(), "initialize result missing capabilities.prompts: {init}" ); - // The negotiated protocol version is echoed back. assert_eq!(result["protocolVersion"], PROTOCOL_VERSION); // 2) notifications/initialized (no id — a notification, no response expected). @@ -385,9 +203,7 @@ fn serve_handshake_lists_all_tools_with_clean_stdout() { assert_eq!(got, want, "tools/list returned an unexpected tool set"); - // 4) resources/templates/list (id 3): the server advertises the single - // `nbox://{kind}/{ref}` template. This proves the resource ServerHandler - // methods are wired through the real protocol, not just the inner helper. + // 4) resources/templates/list (id 3): the single `nbox://{kind}/{ref}` template. server.send(&json!({ "jsonrpc": "2.0", "id": 3, @@ -402,8 +218,7 @@ fn serve_handshake_lists_all_tools_with_clean_stdout() { assert_eq!(list.len(), 1, "expected exactly one resource template"); assert_eq!(list[0]["uriTemplate"], "nbox://{kind}/{ref}"); - // 5) resources/list (id 4): no static resources (the template covers - // everything), so an empty list — but the method must succeed, not 404. + // 5) resources/list (id 4): no static resources, but the method must succeed. server.send(&json!({ "jsonrpc": "2.0", "id": 4, @@ -420,8 +235,7 @@ fn serve_handshake_lists_all_tools_with_clean_stdout() { "expected no static resources: {resources}" ); - // 6) prompts/list (id 5): the curated investigation-prompt catalog. Proves - // the prompt ServerHandler methods are wired through the real protocol. + // 6) prompts/list (id 5): the curated investigation-prompt catalog. server.send(&json!({ "jsonrpc": "2.0", "id": 5, @@ -453,8 +267,7 @@ fn serve_handshake_lists_all_tools_with_clean_stdout() { "prompts/list returned an unexpected catalog" ); - // 7) prompts/get (id 6): expanding a named prompt returns a user-role - // message whose text references the nbox tools to call. + // 7) prompts/get (id 6): expanding a named prompt returns a user-role message. server.send(&json!({ "jsonrpc": "2.0", "id": 6, @@ -483,7 +296,6 @@ fn serve_handshake_lists_all_tools_with_clean_stdout() { "plan missing tool ref: {text}" ); - // Close stdin and make sure the process exits (killed as a backstop). server.shutdown(); } @@ -491,19 +303,13 @@ fn serve_handshake_lists_all_tools_with_clean_stdout() { async fn local_stdio_writes_plan_and_apply_through_json_rpc() { let netbox = MockServer::start().await; mount_device_status_update_requiring_token(&netbox, "nbt_profile.key").await; - let mut server = ServeChild::spawn_with(&netbox.uri(), &["--local-writes"], "nbt_profile.key"); + let mut server = ServeChild::spawn( + &serve_config(&netbox.uri()), + &["--local-writes"], + "nbt_profile.key", + ); - server.send(&json!({ - "jsonrpc": "2.0", - "id": 1, - "method": "initialize", - "params": { - "protocolVersion": PROTOCOL_VERSION, - "capabilities": {}, - "clientInfo": { "name": "nbox-local-write-e2e-test", "version": "0.0.0" } - } - })); - let init = server.read_response(1); + let init = server.handshake(); assert!( init["result"]["instructions"] .as_str() @@ -511,27 +317,14 @@ async fn local_stdio_writes_plan_and_apply_through_json_rpc() { .contains("Local writes are enabled"), "initialize instructions should describe local writes: {init}" ); - server.send(&json!({ - "jsonrpc": "2.0", - "method": "notifications/initialized" - })); - server.send(&json!({ - "jsonrpc": "2.0", - "id": 2, - "method": "tools/call", - "params": { - "name": "nbox_plan_write", - "arguments": { - "operation": { - "kind": "device_status", - "device": "edge01", - "status": "active" - } - } - } - })); - let plan_msg = server.read_response(2); + let plan_msg = server.call( + 2, + "nbox_plan_write", + json!({ + "operation": { "kind": "device_status", "device": "edge01", "status": "active" } + }), + ); let plan = tool_payload(&plan_msg); assert_eq!(plan["target"]["kind"], "device", "plan: {plan}"); assert_eq!(plan["patch"], json!({"status": "active"}), "plan: {plan}"); @@ -543,16 +336,7 @@ async fn local_stdio_writes_plan_and_apply_through_json_rpc() { "plan carries a confirm token" ); - server.send(&json!({ - "jsonrpc": "2.0", - "id": 3, - "method": "tools/call", - "params": { - "name": "nbox_apply_write", - "arguments": { "plan": plan } - } - })); - let receipt_msg = server.read_response(3); + let receipt_msg = server.call(3, "nbox_apply_write", json!({ "plan": plan })); let receipt = tool_payload(&receipt_msg); assert_eq!(receipt["applied"], true, "receipt: {receipt}"); assert_eq!(receipt["status"], 200, "receipt: {receipt}"); diff --git a/tests/support/mod.rs b/tests/support/mod.rs index 8c42cc4..32360ce 100644 --- a/tests/support/mod.rs +++ b/tests/support/mod.rs @@ -10,3 +10,4 @@ pub mod binary; pub mod fixtures; pub mod json_contract; pub mod netbox; +pub mod serve; diff --git a/tests/support/serve.rs b/tests/support/serve.rs new file mode 100644 index 0000000..4993f18 --- /dev/null +++ b/tests/support/serve.rs @@ -0,0 +1,206 @@ +//! Shared harness for driving `nbox serve` (the stdio MCP transport) in +//! integration tests. Spawns the *real* compiled binary +//! (`env!("CARGO_BIN_EXE_nbox")`) and speaks newline-delimited JSON-RPC over its +//! stdin/stdout. +//! +//! Determinism / no-hang: a reader thread streams stdout lines onto a channel so +//! every read is bounded by `recv_timeout`; a missing/garbled response fails +//! fast instead of hanging. On drop the child is killed and reaped, so no process +//! is left behind. + +use std::io::{BufRead, BufReader, Write}; +use std::process::{Child, Command, Stdio}; +use std::sync::mpsc::{Receiver, RecvTimeoutError, channel}; +use std::time::Duration; + +use serde_json::{Value, json}; +use tempfile::NamedTempFile; + +/// The protocol version the server advertises (`ProtocolVersion::LATEST`). The +/// server negotiates down to an older client version, so sending the latest is +/// always accepted. +pub const PROTOCOL_VERSION: &str = "2025-11-25"; + +/// Every read off the child's stdout is bounded by this — a generous ceiling so +/// a broken server fails fast rather than hanging the suite. +const READ_TIMEOUT: Duration = Duration::from_secs(30); + +/// A live `nbox serve` child plus the plumbing to talk to it. +pub struct ServeChild { + child: Child, + // `Option` so `shutdown` can drop stdin (EOF) without leaving an invalid field. + stdin: Option, + lines: Receiver, + // Kept so the temp config outlives the running child. + _config: NamedTempFile, +} + +impl ServeChild { + /// Spawn `nbox serve ` with `config_toml` written to a temp file + /// (`--config`) and `token` exported as `NBOX_TOKEN` (the direct override + /// `connect()` reads first). Logging is forced quiet + to stderr so the + /// stdout-is-pure-JSON-RPC invariant holds regardless of the caller's env. + pub fn spawn(config_toml: &str, serve_args: &[&str], token: &str) -> Self { + let mut config = NamedTempFile::new().expect("create temp config"); + config + .write_all(config_toml.as_bytes()) + .expect("write temp config"); + config.flush().expect("flush temp config"); + + let mut command = Command::new(env!("CARGO_BIN_EXE_nbox")); + command.arg("--config").arg(config.path()).arg("serve"); + command.args(serve_args); + let mut child = command + .env("NBOX_TOKEN", token) + .env_remove("NBOX_LOG") + .env_remove("RUST_LOG") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn nbox serve"); + + let stdin = child.stdin.take().expect("child stdin"); + let stdout = child.stdout.take().expect("child stdout"); + + let (tx, rx) = channel::(); + std::thread::spawn(move || { + let reader = BufReader::new(stdout); + for line in reader.lines() { + match line { + Ok(l) => { + if tx.send(l).is_err() { + break; // receiver gone — test finished + } + } + Err(_) => break, + } + } + }); + + ServeChild { + child, + stdin: Some(stdin), + lines: rx, + _config: config, + } + } + + /// Send one compact JSON value as a single newline-terminated line. + pub fn send(&mut self, msg: &Value) { + let line = serde_json::to_string(msg).expect("serialize message"); + assert!(!line.contains('\n'), "framing must be one line per message"); + let stdin = self.stdin.as_mut().expect("child stdin is open"); + writeln!(stdin, "{line}").expect("write to child stdin"); + stdin.flush().expect("flush child stdin"); + } + + /// Read the next stdout line within the bounded timeout, asserting it parses + /// as JSON (the stdout-is-pure-JSON-RPC invariant), and return the value. + fn next_json(&self) -> Value { + match self.lines.recv_timeout(READ_TIMEOUT) { + Ok(line) => serde_json::from_str(&line) + .unwrap_or_else(|e| panic!("stdout line was not valid JSON: {e}\nline: {line:?}")), + Err(RecvTimeoutError::Timeout) => { + panic!("timed out waiting for a stdout line from `nbox serve`") + } + Err(RecvTimeoutError::Disconnected) => { + panic!("`nbox serve` closed stdout before sending the expected response") + } + } + } + + /// Read responses until one carries `id == want_id` (skipping notifications). + pub fn read_response(&self, want_id: i64) -> Value { + loop { + let v = self.next_json(); + if v.get("id").and_then(Value::as_i64) == Some(want_id) { + return v; + } + } + } + + /// `initialize` (id 1) + `notifications/initialized`. Returns the initialize + /// result message. + pub fn handshake(&mut self) -> Value { + self.send(&json!({ + "jsonrpc": "2.0", "id": 1, "method": "initialize", + "params": { + "protocolVersion": PROTOCOL_VERSION, + "capabilities": {}, + "clientInfo": { "name": "nbox-it", "version": "0.0.0" } + } + })); + let init = self.read_response(1); + self.send(&json!({ "jsonrpc": "2.0", "method": "notifications/initialized" })); + init + } + + /// `tools/call` `name` with `args` at `id`; returns the raw response message. + pub fn call(&mut self, id: i64, name: &str, args: Value) -> Value { + self.send(&json!({ + "jsonrpc": "2.0", "id": id, "method": "tools/call", + "params": { "name": name, "arguments": args } + })); + self.read_response(id) + } + + /// Close stdin (EOF) and ensure the child has exited, killing it as a backstop. + pub fn shutdown(mut self) { + self.stdin = None; + for _ in 0..100 { + match self.child.try_wait() { + Ok(Some(_)) => return, + Ok(None) => std::thread::sleep(Duration::from_millis(20)), + Err(_) => break, + } + } + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +impl Drop for ServeChild { + fn drop(&mut self) { + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +/// Extract a successful tool's structured payload (the `Json` it returns). +pub fn tool_payload(msg: &Value) -> Value { + assert!(msg.get("error").is_none(), "tool returned an error: {msg}"); + let result = &msg["result"]; + assert_ne!( + result["isError"], + json!(true), + "tool execution error: {msg}" + ); + if let Some(sc) = result.get("structuredContent") + && !sc.is_null() + { + return sc.clone(); + } + let text = result["content"][0]["text"] + .as_str() + .unwrap_or_else(|| panic!("tool result has no structuredContent/text: {msg}")); + serde_json::from_str(text).unwrap_or_else(|_| panic!("tool result text not JSON: {text}")) +} + +/// The error text of a *failed* tool call — a JSON-RPC `error`, or an `isError` +/// content payload — or `None` when the call succeeded. For negative tests. +pub fn tool_error(msg: &Value) -> Option { + if let Some(e) = msg.get("error") { + return Some(e.to_string()); + } + let result = msg.get("result")?; + if result.get("isError") == Some(&json!(true)) { + return Some( + result["content"][0]["text"] + .as_str() + .unwrap_or_default() + .to_string(), + ); + } + None +}