use super::wire::{Frame, Pipe, read};
use serde_json::{Value, json};
use tokio::io::{AsyncWrite, AsyncWriteExt as _, BufReader};
async fn write<W: AsyncWrite + Unpin>(writer: &mut W, value: Value) -> anyhow::Result<()> {
let mut bytes = serde_json::to_vec(&value)?;
bytes.push(b'\n');
writer.write_all(&bytes).await?;
writer.flush().await?;
Ok(())
}
#[tokio::test]
async fn r1_row_12_729_answers_do_not_stop_stdout_drain() -> anyhow::Result<()> {
burst(729).await
}
#[tokio::test]
async fn r1_row_12_40_requests_over_1024_byte_pipe_drain() -> anyhow::Result<()> {
burst(40).await
}
async fn burst(count: usize) -> anyhow::Result<()> {
let (client, agent) = tokio::io::duplex(1024);
let (input, output) = tokio::io::split(client);
let mut pipe = Pipe::start(input, output, 65536, 1024, 1);
let script = tokio::spawn(async move {
let (input, mut output) = tokio::io::split(agent);
for id in 0..count {
write(
&mut output,
json!({"jsonrpc":"2.0","id":id,"method":"session/request_permission","params":{}}),
)
.await?;
}
let mut input = BufReader::new(input);
for expected in 0..count {
let answer = read(&mut input, 65536)
.await?
.ok_or_else(|| anyhow::anyhow!("permission response absent"))?;
let Frame::Response { id, result } = answer else {
anyhow::bail!("permission response is not a response");
};
assert_eq!(id, json!(expected));
let result = result.map_err(|_| anyhow::anyhow!("permission answer failed"))?;
assert_eq!(result["outcome"]["outcome"], "cancelled");
}
Ok::<_, anyhow::Error>(())
});
for _ in 0..count {
let frame = pipe
.input
.recv()
.await
.ok_or_else(|| anyhow::anyhow!("ACP reader closed"))??
.ok_or_else(|| anyhow::anyhow!("ACP stdout closed"))?;
let Frame::Request { id, .. } = frame else {
anyhow::bail!("peer sent a non-request");
};
pipe.output
.response(id, json!({"outcome":{"outcome":"cancelled"}}))?;
}
script.await??;
pipe.reader.abort();
drop(pipe.output);
pipe.writer.await??;
Ok(())
}
#[tokio::test]
async fn r4_answer_receipt_follows_the_actual_pipe_write() -> anyhow::Result<()> {
use tokio::io::AsyncReadExt as _;
let (client, mut agent) = tokio::io::duplex(1);
let (input, output) = tokio::io::split(client);
let mut pipe = Pipe::start(input, output, 65536, 4, 1);
pipe.output.response_tracked(
json!(1),
json!({"outcome":{"outcome":"cancelled"}}),
"receipt".to_owned(),
)?;
let mut first = [0];
agent.read_exact(&mut first).await?;
assert!(matches!(
pipe.written.try_recv(),
Err(tokio::sync::mpsc::error::TryRecvError::Empty)
));
while first[0] != b'\n' {
agent.read_exact(&mut first).await?;
}
let receipt = pipe
.written
.recv()
.await
.ok_or_else(|| anyhow::anyhow!("receipt absent"))?;
assert!(receipt.delivered);
assert_eq!(receipt.key, "receipt");
pipe.reader.abort();
drop(pipe.output);
pipe.writer.await??;
Ok(())
}
#[tokio::test]
async fn r1_write_queue_full_and_closed_are_distinct() -> anyhow::Result<()> {
use tokio::io::AsyncReadExt as _;
let (client, mut peer) = tokio::io::duplex(1);
let (input, output) = tokio::io::split(client);
let pipe = Pipe::start(input, output, 65536, 1, 1);
pipe.output.response(json!(1), json!({}))?;
peer.read_exact(&mut [0]).await?;
pipe.output.response(json!(2), json!({}))?;
let full = pipe
.output
.response(json!(3), json!({}))
.err()
.ok_or_else(|| anyhow::anyhow!("full queue accepted"))?;
assert_eq!(full.to_string(), "acp_write_queue_full");
pipe.writer.abort();
assert!(pipe.writer.await.is_err_and(|error| error.is_cancelled()));
let closed = pipe
.output
.response(json!(4), json!({}))
.err()
.ok_or_else(|| anyhow::anyhow!("closed writer accepted"))?;
assert_eq!(closed.to_string(), "acp_writer_closed");
pipe.reader.abort();
assert!(pipe.reader.await.is_err_and(|error| error.is_cancelled()));
Ok(())
}