use std::io::{BufRead, BufReader, Write};
use std::process::{Command, Stdio};
use std::sync::mpsc::{self, Receiver};
use std::time::{Duration, Instant};
use serde_json::{json, Value};
use tempfile::TempDir;
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
#[cfg(unix)]
fn write_slow_mock_log(directory: &std::path::Path) -> std::path::PathBuf {
let path = directory.join("slow-mock-log");
std::fs::write(
&path,
"#!/bin/sh\nprintf '%s\\n' '{\"eventMessage\":\"during scan\"}'\nexec sleep 2\n",
)
.expect("mock log");
let mut permissions = std::fs::metadata(&path).expect("metadata").permissions();
permissions.set_mode(0o755);
std::fs::set_permissions(&path, permissions).expect("chmod");
path
}
fn send(stdin: &mut impl Write, message: Value) {
writeln!(stdin, "{message}").expect("write MCP message");
stdin.flush().expect("flush MCP message");
}
fn response(receiver: &Receiver<Value>, id: i64) -> Value {
loop {
let message = receiver
.recv_timeout(Duration::from_secs(15))
.expect("MCP response before timeout");
if message.get("id").and_then(Value::as_i64) == Some(id) {
return message;
}
}
}
fn response_within(receiver: &Receiver<Value>, id: i64, timeout: Duration) -> Option<Value> {
let deadline = Instant::now() + timeout;
loop {
let remaining = deadline.checked_duration_since(Instant::now())?;
let message = receiver.recv_timeout(remaining).ok()?;
if message.get("id").and_then(Value::as_i64) == Some(id) {
return Some(message);
}
}
}
#[test]
fn exposes_code_mode_and_executes_a_devsql_query() {
let mut child = Command::new(env!("CARGO_BIN_EXE_devsql"))
.arg("--mcp")
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("start DevSQL MCP server");
let mut stdin = child.stdin.take().expect("MCP stdin");
let stdout = child.stdout.take().expect("MCP stdout");
let (sender, receiver) = mpsc::channel();
let reader = std::thread::spawn(move || {
for line in BufReader::new(stdout).lines() {
let line = line.expect("read MCP line");
if let Ok(value) = serde_json::from_str(&line) {
sender.send(value).expect("forward MCP message");
}
}
});
send(
&mut stdin,
json!({
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2025-11-25",
"capabilities": {},
"clientInfo": {"name": "devsql-test", "version": "1.0.0"}
}
}),
);
let initialized = response(&receiver, 1);
assert!(initialized.get("result").is_some(), "{initialized}");
send(
&mut stdin,
json!({"jsonrpc": "2.0", "method": "notifications/initialized"}),
);
send(
&mut stdin,
json!({"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}),
);
let listed = response(&receiver, 2);
let tools = listed["result"]["tools"].as_array().expect("tools array");
let mut names = tools
.iter()
.filter_map(|tool| tool["name"].as_str())
.collect::<Vec<_>>();
names.sort_unstable();
assert_eq!(
names,
[
"codemode_cancel",
"codemode_decide",
"codemode_execute",
"codemode_execution",
"codemode_search",
]
);
send(
&mut stdin,
json!({
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {"name": "codemode_search", "arguments": {"query": "query"}}
}),
);
let searched = response(&receiver, 3);
assert!(searched.get("error").is_none(), "{searched}");
assert!(searched.to_string().contains("devsql.query"), "{searched}");
send(
&mut stdin,
json!({
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {
"name": "codemode_execute",
"arguments": {"code": "devsql.query({ query: 'SELECT 1 AS value' })"}
}
}),
);
let started = response(&receiver, 4);
assert!(started.get("error").is_none(), "{started}");
let execution_id = started["result"]["structuredContent"]["id"]
.as_str()
.expect("execution id")
.to_string();
let mut executed = started;
for id in 5..105 {
if executed["result"]["structuredContent"]["status"] == "completed" {
break;
}
std::thread::sleep(Duration::from_millis(20));
send(
&mut stdin,
json!({
"jsonrpc": "2.0",
"id": id,
"method": "tools/call",
"params": {
"name": "codemode_execution",
"arguments": {"id": execution_id}
}
}),
);
executed = response(&receiver, id);
}
assert_eq!(
executed["result"]["structuredContent"]["status"], "completed",
"{executed}"
);
assert!(
executed.to_string().contains("value") && executed.to_string().contains('1'),
"{executed}"
);
drop(stdin);
let status = child.wait().expect("wait for MCP server");
reader.join().expect("join MCP reader");
assert!(status.success(), "MCP server exited with {status}");
}
#[test]
#[cfg(unix)]
fn code_mode_stays_responsive_during_concurrent_queries() {
let temp = TempDir::new().expect("temp");
let log_bin = write_slow_mock_log(temp.path());
let mut child = Command::new(env!("CARGO_BIN_EXE_devsql"))
.arg("--mcp")
.env("DEVSQL_MACOS_LOG_BIN", log_bin)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.spawn()
.expect("start DevSQL MCP server");
let mut stdin = child.stdin.take().expect("MCP stdin");
let stdout = child.stdout.take().expect("MCP stdout");
let (sender, receiver) = mpsc::channel();
let reader = std::thread::spawn(move || {
for line in BufReader::new(stdout).lines() {
let line = line.expect("read MCP line");
if let Ok(value) = serde_json::from_str(&line) {
sender.send(value).expect("forward MCP message");
}
}
});
send(
&mut stdin,
json!({
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2025-11-25",
"capabilities": {},
"clientInfo": {"name": "devsql-test", "version": "1.0.0"}
}
}),
);
assert!(response(&receiver, 1).get("result").is_some());
send(
&mut stdin,
json!({"jsonrpc": "2.0", "method": "notifications/initialized"}),
);
let slow_query = "SELECT message FROM macos_logs WHERE timeout = 2 AND max_rows = 100";
let code = format!(
"Promise.all([devsql.query({{ query: {slow_query:?}, log_last: '1m', log_level: 'info' }}), devsql.query({{ query: {slow_query:?}, log_last: '1m', log_level: 'info' }})])"
);
send(
&mut stdin,
json!({
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {"name": "codemode_execute", "arguments": {"code": code}}
}),
);
let started = response(&receiver, 2);
let execution_id = started["result"]["structuredContent"]["id"]
.as_str()
.expect("execution id")
.to_string();
send(
&mut stdin,
json!({
"jsonrpc": "2.0",
"id": 3,
"method": "tools/call",
"params": {
"name": "codemode_execution",
"arguments": {"id": execution_id}
}
}),
);
let status = response_within(&receiver, 3, Duration::from_secs(1));
if status.is_none() {
child.kill().expect("stop unresponsive MCP server");
}
assert!(
status.is_some(),
"execution status was blocked by query work"
);
send(
&mut stdin,
json!({
"jsonrpc": "2.0",
"id": 4,
"method": "tools/call",
"params": {"name": "codemode_search", "arguments": {"query": "query"}}
}),
);
let search = response_within(&receiver, 4, Duration::from_secs(1));
if search.is_none() {
child.kill().expect("stop unresponsive MCP server");
}
assert!(
search.is_some(),
"Code Mode search was blocked by query work"
);
let deadline = Instant::now() + Duration::from_secs(30);
let mut executed = None;
for id in 5.. {
send(
&mut stdin,
json!({
"jsonrpc": "2.0",
"id": id,
"method": "tools/call",
"params": {
"name": "codemode_execution",
"arguments": {"id": execution_id}
}
}),
);
let snapshot = response(&receiver, id);
if snapshot["result"]["structuredContent"]["status"] == "completed" {
executed = Some(snapshot);
break;
}
assert!(
Instant::now() < deadline,
"concurrent query execution did not complete"
);
std::thread::sleep(Duration::from_millis(20));
}
let executed = executed.expect("completed execution");
let result = executed["result"]["structuredContent"]["result"]
.as_array()
.expect("Promise.all result");
assert_eq!(result.len(), 2, "{executed}");
assert!(result
.iter()
.all(|value| value.to_string().contains("during scan")));
drop(stdin);
let status = child.wait().expect("wait for MCP server");
reader.join().expect("join MCP reader");
assert!(status.success(), "MCP server exited with {status}");
}