#![cfg(test)]
use super::*;
#[test]
fn listener_negotiation_logs_do_not_echo_remote_errors() {
let message = "request rejected: token=listener-secret-sentinel";
for method in ["host.getConfiguration", "host.ready"] {
assert_eq!(
remote_error_log_message(method, message),
"listener negotiation request rejected"
);
}
assert_eq!(remote_error_log_message("ping", message), message);
}
#[test]
fn deserialize_notification() {
let json = r#"{"jsonrpc":"2.0","method":"session.event","params":{"id":"e1"}}"#;
let msg: JsonRpcMessage = serde_json::from_str(json).unwrap();
assert!(matches!(msg, JsonRpcMessage::Notification(n) if n.method == "session.event"));
}
#[test]
fn deserialize_request() {
let json =
r#"{"jsonrpc":"2.0","id":5,"method":"permission.request","params":{"kind":"shell"}}"#;
let msg: JsonRpcMessage = serde_json::from_str(json).unwrap();
assert!(
matches!(msg, JsonRpcMessage::Request(r) if r.id == 5 && r.method == "permission.request")
);
}
#[test]
fn deserialize_response_with_result() {
let json = r#"{"jsonrpc":"2.0","id":3,"result":{"ok":true}}"#;
let msg: JsonRpcMessage = serde_json::from_str(json).unwrap();
assert!(matches!(msg, JsonRpcMessage::Response(r) if r.id == 3 && !r.is_error()));
}
#[test]
fn deserialize_error_response() {
let json = r#"{"jsonrpc":"2.0","id":7,"error":{"code":-32600,"message":"Invalid Request","data":{"nested":[1,{"reason":"invalid"}]}}}"#;
let msg: JsonRpcMessage = serde_json::from_str(json).unwrap();
match msg {
JsonRpcMessage::Response(r) => {
assert!(r.is_error());
let err = r.error.unwrap();
assert_eq!(err.code, -32600);
assert_eq!(err.message, "Invalid Request");
assert_eq!(
err.data,
Some(serde_json::json!({"nested": [1, {"reason": "invalid"}]}))
);
}
other => panic!("expected Response, got {other:?}"),
}
}
#[test]
fn null_id_error_decodes_as_an_uncorrelated_error() {
let json = r#"{"jsonrpc":"2.0","id":null,"error":{"code":-32700,"message":"recursion limit exceeded at line 1 column 513"}}"#;
match serde_json::from_str::<JsonRpcMessage>(json).unwrap() {
JsonRpcMessage::UncorrelatedError(error) => {
assert_eq!(error.code, -32700);
assert_eq!(
error.message,
"recursion limit exceeded at line 1 column 513"
);
}
other => panic!("expected UncorrelatedError, got {other:?}"),
}
let json = r#"{"jsonrpc":"2.0","id":3,"error":{"code":-32700,"message":"m"}}"#;
assert!(matches!(
serde_json::from_str::<JsonRpcMessage>(json).unwrap(),
JsonRpcMessage::Response(response) if response.id == 3
));
let json = r#"{"jsonrpc":"2.0","id":null,"result":{}}"#;
assert!(serde_json::from_str::<JsonRpcMessage>(json).is_err());
}
#[test]
fn an_error_code_wider_than_i32_is_kept() {
let json = r#"{"jsonrpc":"2.0","id":3,"error":{"code":4294967296,"message":"wide"}}"#;
let JsonRpcMessage::Response(response) = serde_json::from_str::<JsonRpcMessage>(json).unwrap()
else {
panic!("expected a response");
};
let error = response.error.unwrap();
assert_eq!(error.code, 4_294_967_296);
let error = Error::from_rpc(error.code, error.message, None);
assert_eq!(error.rpc_code(), Some(i32::MAX));
assert!(
error
.message()
.is_some_and(|message| message.contains("4294967296")),
"{error}"
);
}
#[test]
fn a_peer_message_is_repeated_without_its_quoted_values() {
let message = r#"invalid type: string "ghp_secret\"token", expected u64 at line 1 column 40"#;
assert_eq!(
sanitized_peer_message(message),
"invalid type: string \"…\", expected u64 at line 1 column 40"
);
assert_eq!(
sanitized_peer_message("unknown variant `ghp_secret`, expected one of `a`"),
"unknown variant `…`, expected one of `…`"
);
let long = "x".repeat(500);
assert_eq!(
sanitized_peer_message(&long).chars().count(),
PEER_MESSAGE_MAX_CHARS + 1
);
}
#[test]
fn deserialize_rejects_non_object() {
let result = serde_json::from_str::<JsonRpcMessage>(r#""not an object""#);
assert!(result.is_err());
}
#[test]
fn frame_decoder_preserves_message_classification_payloads_and_errors() {
for body in [
r#"{"jsonrpc":"2.0","id":1,"result":{}}"#,
r#"{"jsonrpc":"2.0","id":1,"result":null}"#,
r#"{"jsonrpc":"2.0","id":1,"result":[false,{"value":0.125}]}"#,
r#"{"jsonrpc":"2.0","id":1,"error":{"code":4294967296,"message":"wide","data":{"nested":[null,true]}}}"#,
r#"{"jsonrpc":"2.0","id":null,"error":{"code":-32700,"message":"uncorrelated"}}"#,
r#"{"jsonrpc":"2.0","method":"notify","params":{"text":"value"}}"#,
r#"{"jsonrpc":"2.0","method":"notify","params":null,"result":false,"error":false}"#,
r#"{"jsonrpc":"2.0","id":2,"method":"request","params":[1,null,true]}"#,
r#"{"jsonrpc":"2.0","id":2,"method":"request","result":false,"error":false}"#,
r#"{"jsonrpc":"2.0","id":2,"met\u0068od":"request","params":{}}"#,
r#"{"jsonrpc":"2.0","id":1,"result":{},"extra":{"nested":[1,2]}}"#,
r#"{"jsonrpc":"2.0","id":1,"id":2,"result":{}}"#,
r#"{"jsonrpc":"2.0","method":false,"method":"notify","params":{}}"#,
r#"{"jsonrpc":"2.0","method":"notify","params":{},"params":[1,2]}"#,
r#"{"jsonrpc":"1.0","id":1,"result":{}}"#,
r#"{"jsonrpc":false,"id":1,"result":{}}"#,
r#"{"id":1,"result":{}}"#,
r#"{"jsonrpc":"2.0","id":null,"result":{}}"#,
r#"{"jsonrpc":"2.0","id":null,"method":"request"}"#,
r#"{"jsonrpc":"2.0","id":1,"method":null}"#,
r#"{"jsonrpc":"2.0","id":-1,"result":{}}"#,
r#"{"jsonrpc":"2.0","id":1.5,"result":{}}"#,
r#"{"jsonrpc":"2.0","id":"string","result":{}}"#,
r#"{"jsonrpc":"2.0","id":1,"error":false}"#,
r#"{"jsonrpc":"2.0","id":null,"error":null}"#,
r#"{"jsonrpc":"2.0","result":{}}"#,
r#"["2.0",1,"request",{},null,null]"#,
r#"null"#,
r#""text""#,
r#"{"jsonrpc":"2.0","id":1,"result":"\ud83d"}"#,
r#"{"jsonrpc":"2.0","id":1,"result":{}} trailing"#,
r#"{"jsonrpc":"2.0","id":1,"result":{},}"#,
] {
let expected = serde_json::from_str::<JsonRpcMessage>(body)
.map(|message| serde_json::to_value(message).unwrap())
.map_err(|error| error.to_string());
let actual = JsonRpcMessage::from_slice(body.as_bytes())
.map(|message| serde_json::to_value(message).unwrap())
.map_err(|error| error.to_string());
assert_eq!(actual, expected, "{body}");
}
}
#[test]
fn deserialize_preserves_optional_payloads() {
for payload in [
None,
Some(Value::Null),
Some(serde_json::json!(false)),
Some(serde_json::json!(42)),
Some(serde_json::json!("text")),
Some(serde_json::json!([{"nested": [1, null, true]}])),
Some(serde_json::json!({"rows": [{"content": "result"}]})),
] {
for mut envelope in [
serde_json::json!({"jsonrpc": "2.0", "method": "notify"}),
serde_json::json!({"jsonrpc": "2.0", "id": 1, "method": "request"}),
serde_json::json!({"jsonrpc": "2.0", "id": 1}),
] {
let (payload_key, ignored_key) = if envelope.get("method").is_some() {
("params", "result")
} else {
("result", "params")
};
envelope[ignored_key] = serde_json::json!({"ignored": "opposite payload"});
if let Some(payload) = &payload {
envelope[payload_key] = payload.clone();
}
let actual = match serde_json::from_value::<JsonRpcMessage>(envelope).unwrap() {
JsonRpcMessage::Request(request) => request.params,
JsonRpcMessage::Response(response) => response.result,
JsonRpcMessage::Notification(notification) => notification.params,
JsonRpcMessage::UncorrelatedError(error) => {
panic!("no envelope here has a null id: {error:?}")
}
};
assert_eq!(actual, payload.clone().filter(|value| !value.is_null()));
}
}
}
#[test]
fn deserialize_rejects_invalid_metadata() {
for json in [
r#"{"jsonrpc":null,"method":"notify","params":{"nested":[1]}}"#,
r#"{"jsonrpc":"2.0","method":42,"params":{"nested":[1]}}"#,
r#"{"jsonrpc":"2.0","id":null,"result":{}}"#,
r#"{"jsonrpc":null,"id":null,"error":{"code":-32700,"message":"m"}}"#,
r#"{"jsonrpc":"2.0","id":"1","result":{}}"#,
r#"{"jsonrpc":"2.0","id":-1,"result":{}}"#,
r#"{"jsonrpc":"2.0","id":1,"method":null,"params":{}}"#,
r#"{"jsonrpc":"2.0","id":1,"result":{},"error":{"code":"bad","message":"error"}}"#,
] {
assert!(
serde_json::from_str::<JsonRpcMessage>(json).is_err(),
"{json}"
);
}
}
fn published_frame<T: serde::Serialize>(message: &T) -> Vec<u8> {
let body = serde_json::to_vec(message).unwrap();
let mut frame = format!("Content-Length: {}\r\n\r\n", body.len()).into_bytes();
frame.extend_from_slice(&body);
frame
}
#[test]
fn encode_frame_writes_the_published_frame() {
let message = serde_json::json!({
"jsonrpc": "2.0",
"id": 7,
"result": {"content": "a \"quoted\" line\n"}
});
assert_eq!(
encode_frame(&message).unwrap().as_bytes(),
published_frame(&message)
);
for body_len in [9, 10, 99, 100, 999, 1000, 9999, 10000] {
let message = "x".repeat(body_len - 2);
let frame = encode_frame(&message).unwrap();
assert!(
frame
.as_bytes()
.starts_with(format!("Content-Length: {body_len}\r\n\r\n").as_bytes())
);
assert_eq!(frame.as_bytes(), published_frame(&message));
}
}
#[test]
fn encode_frame_gives_back_large_spare_capacity() {
let message = "x".repeat(64 * 1024);
let frame = encode_frame(&message).unwrap();
assert_eq!(frame.as_bytes(), published_frame(&message));
assert!(
frame.buf.capacity() <= frame.buf.len() + 4096,
"a {}-byte frame keeps {} bytes",
frame.buf.len(),
frame.buf.capacity()
);
}
fn nested(levels: usize) -> Value {
let mut value = Value::Null;
for _ in 0..levels {
value = Value::Array(vec![value]);
}
value
}
#[test]
fn peer_max_nesting_is_what_serde_json_reads() {
fn decodes(response: &JsonRpcResponse) -> bool {
serde_json::from_slice::<Value>(&serde_json::to_vec(response).unwrap()).is_ok()
}
let with_result = |levels| JsonRpcResponse {
jsonrpc: "2.0".into(),
id: 1,
result: Some(nested(levels)),
error: None,
};
let with_data = |levels| JsonRpcResponse {
jsonrpc: "2.0".into(),
id: 1,
result: None,
error: Some(JsonRpcError {
code: -1,
message: "m".into(),
data: Some(nested(levels)),
}),
};
for (case, at_limit, too_deep) in [
(
"result",
with_result(PEER_MAX_NESTING - 1),
with_result(PEER_MAX_NESTING),
),
(
"error data",
with_data(PEER_MAX_NESTING - 2),
with_data(PEER_MAX_NESTING - 1),
),
] {
assert!(
decodes(&at_limit),
"{case}: serde_json reads {PEER_MAX_NESTING} levels"
);
assert!(
!decodes(&too_deep),
"{case}: serde_json refuses one level more"
);
assert!(
!at_limit.nests_too_deep_for_peer(),
"{case}: the guard passes the readable response"
);
assert!(
too_deep.nests_too_deep_for_peer(),
"{case}: the guard stops the unreadable one"
);
}
}
#[test]
fn a_request_passes_the_guard_exactly_when_serde_json_reads_it() {
fn decodes(request: &JsonRpcRequest) -> bool {
serde_json::from_slice::<Value>(&serde_json::to_vec(request).unwrap()).is_ok()
}
let at_limit = JsonRpcRequest::new(1, "m", Some(nested(PEER_MAX_NESTING - 1)));
let too_deep = JsonRpcRequest::new(1, "m", Some(nested(PEER_MAX_NESTING)));
assert!(
decodes(&at_limit),
"serde_json reads a request {PEER_MAX_NESTING} levels deep"
);
assert!(
!decodes(&too_deep),
"serde_json refuses a request one level deeper"
);
assert!(
!at_limit.nests_too_deep_for_peer(),
"the request guard passes the readable request"
);
assert!(
too_deep.nests_too_deep_for_peer(),
"the request guard stops the unreadable request"
);
}
#[tokio::test]
async fn request_after_the_read_loop_ended_leaves_no_pending_entry() {
for eager in [false, true] {
let (client_write, _server_read) = tokio::io::duplex(4096);
let (server_write, client_read) = tokio::io::duplex(4096);
let (notification_tx, _) = broadcast::channel(16);
let (request_tx, _request_rx) = mpsc::unbounded_channel();
let client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);
let closed = client.connection_closed_token();
drop(server_write);
tokio::time::timeout(std::time::Duration::from_secs(5), closed.cancelled())
.await
.expect("the read loop did not end");
let error = if eager {
match client.begin_request_with_inline_callback("ping", None, None) {
Err(error) => error,
Ok(_) => panic!("the eager request after the close was not refused"),
}
} else {
tokio::time::timeout(
std::time::Duration::from_secs(5),
client.send_request("ping", None),
)
.await
.expect("the request after the close was not refused")
.unwrap_err()
};
assert_eq!(error.kind(), &ErrorKind::Io);
assert!(
client.pending_requests.read().is_empty(),
"the refused request left its entry (eager: {eager})"
);
}
}
#[tokio::test]
async fn notification_receivers_share_one_parsed_notification() {
let (client_write, _server_read) = tokio::io::duplex(4096);
let (mut server_write, client_read) = tokio::io::duplex(4096);
let (notification_tx, mut first) = broadcast::channel(16);
let mut second = notification_tx.subscribe();
let (request_tx, _request_rx) = mpsc::unbounded_channel();
let _client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);
let body = br#"{"jsonrpc":"2.0","method":"session.event","params":{"sessionId":"s1","event":{"data":{"deltaContent":"hi"}}}}"#;
server_write
.write_all(format!("Content-Length: {}\r\n\r\n", body.len()).as_bytes())
.await
.unwrap();
server_write.write_all(body).await.unwrap();
let deadline = std::time::Duration::from_secs(5);
let first = tokio::time::timeout(deadline, first.recv())
.await
.expect("first receiver got no notification")
.unwrap();
let second = tokio::time::timeout(deadline, second.recv())
.await
.expect("second receiver got no notification")
.unwrap();
assert!(Arc::ptr_eq(&first, &second));
assert_eq!(first.method, "session.event");
assert_eq!(
first.params.as_ref().unwrap()["event"]["data"]["deltaContent"],
"hi"
);
}
#[tokio::test]
async fn a_frame_body_is_reserved_as_it_arrives_not_as_declared() {
let (client_write, _server_read) = tokio::io::duplex(4096);
let (mut server_write, client_read) = tokio::io::duplex(1);
let (notification_tx, _) = broadcast::channel(16);
let (request_tx, _request_rx) = mpsc::unbounded_channel();
let client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);
let closed = client.connection_closed_token();
server_write
.write_all(b"Content-Length: 1125899906842624\r\n\r\n{\"jsonrpc\":\"2.0\"")
.await
.unwrap();
assert!(
!closed.is_cancelled(),
"the connection ended while the peer was still sending the frame"
);
drop(server_write);
tokio::time::timeout(std::time::Duration::from_secs(5), closed.cancelled())
.await
.expect("the read loop did not end at the truncated frame");
}
#[tokio::test]
async fn a_frame_body_larger_than_the_first_reserve_arrives_whole() {
let (client_write, _server_read) = tokio::io::duplex(4096);
let (mut server_write, client_read) = tokio::io::duplex(4096);
let (notification_tx, mut notifications) = broadcast::channel(16);
let (request_tx, _request_rx) = mpsc::unbounded_channel();
let _client = JsonRpcClient::new(client_write, client_read, notification_tx, request_tx);
let text = "x".repeat(FRAME_BODY_FIRST_RESERVE * 5 + 17);
let body = serde_json::to_vec(&serde_json::json!({
"jsonrpc": "2.0",
"method": "session.event",
"params": { "text": text },
}))
.unwrap();
let writer = tokio::spawn(async move {
server_write
.write_all(format!("Content-Length: {}\r\n\r\n", body.len()).as_bytes())
.await
.unwrap();
server_write.write_all(&body).await.unwrap();
server_write
});
let notification =
tokio::time::timeout(std::time::Duration::from_secs(5), notifications.recv())
.await
.expect("the large notification did not arrive")
.unwrap();
assert_eq!(notification.params.as_ref().unwrap()["text"], text);
drop(writer.await.unwrap());
}