use std::sync::Arc;
use axum::body::Body;
use axum::http::{Request, StatusCode};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use tower::ServiceExt;
use crate::adapters::primary::a2a::jsonrpc_serve::serve_jsonrpc;
use crate::application::a2a::A2aRequestHandler;
use crate::domain::a2a::A2aVersion;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct Vector {
pub name: String,
pub request: Value,
pub expected: ExpectedResponse,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(untagged)]
pub enum ExpectedResponse {
Result {
result: Value,
},
Error {
error: ExpectedError,
},
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ExpectedError {
pub code: i64,
pub message_contains: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct VectorResult {
pub name: String,
pub outcome: VectorOutcome,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum VectorOutcome {
Pass,
Fail {
diff: String,
actual: ObservedResponse,
},
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(untagged)]
pub enum ObservedResponse {
Result {
result: Value,
},
Error {
error: ObservedError,
},
Transport {
status: u16,
body: String,
},
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ObservedError {
pub code: i64,
pub message: String,
}
impl core::fmt::Display for ObservedResponse {
fn fmt(&self, formatter: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::Result { result } => write!(formatter, "Result({result})"),
Self::Error { error } => {
write!(formatter, "Error({}, {})", error.code, error.message)
}
Self::Transport { status, body } => {
write!(formatter, "Transport({status}, {body})")
}
}
}
}
pub async fn run_against_in_process(
handler: Arc<dyn A2aRequestHandler>,
protocol_version: A2aVersion,
vectors: &[Vector],
) -> Vec<VectorResult> {
let router = serve_jsonrpc("/a2a", handler, protocol_version);
let mut out = Vec::with_capacity(vectors.len());
for vector in vectors {
out.push(run_one(&router, vector).await);
}
out
}
pub async fn run_one(router: &axum::Router, vector: &Vector) -> VectorResult {
let body_bytes = match serde_json::to_vec(&vector.request) {
Ok(bytes) => bytes,
Err(error) => {
return VectorResult {
name: vector.name.clone(),
outcome: VectorOutcome::Fail {
diff: format!("the request could not be serialised: {error}"),
actual: ObservedResponse::Transport {
status: 0,
body: format!("encode: {error}"),
},
},
};
}
};
let request = match Request::builder()
.method("POST")
.uri("/a2a")
.header("content-type", "application/json")
.body(Body::from(body_bytes))
{
Ok(request) => request,
Err(error) => {
return VectorResult {
name: vector.name.clone(),
outcome: VectorOutcome::Fail {
diff: format!("the harness built a request the http layer refused: {error}"),
actual: ObservedResponse::Transport {
status: 0,
body: format!("build: {error}"),
},
},
};
}
};
let response = match router.clone().oneshot(request).await {
Ok(response) => response,
Err(never) => match never {},
};
let observed = observe_response(response).await;
compare(vector, &observed)
}
async fn observe_response(response: axum::response::Response) -> ObservedResponse {
let status = response.status();
let body_bytes = match axum::body::to_bytes(response.into_body(), 64 * 1024).await {
Ok(bytes) => bytes,
Err(error) => {
return ObservedResponse::Transport {
status: status.as_u16(),
body: format!("read body: {error}"),
};
}
};
let body_text = String::from_utf8_lossy(&body_bytes).into_owned();
if status != StatusCode::OK {
return ObservedResponse::Transport {
status: status.as_u16(),
body: body_text,
};
}
let envelope: Value = match serde_json::from_slice(&body_bytes) {
Ok(value) => value,
Err(error) => {
return ObservedResponse::Transport {
status: status.as_u16(),
body: format!("invalid JSON-RPC envelope: {error}; body: {body_text}"),
};
}
};
let Some(envelope_object) = envelope.as_object() else {
return ObservedResponse::Transport {
status: status.as_u16(),
body: format!("envelope was not a JSON object: {envelope}"),
};
};
envelope_object.get("result").map_or_else(
|| {
envelope_object.get("error").map_or_else(
|| ObservedResponse::Transport {
status: status.as_u16(),
body: format!("envelope carried neither result nor error: {envelope}"),
},
|error| {
let code = error.get("code").and_then(Value::as_i64).unwrap_or(0);
let message = error
.get("message")
.and_then(Value::as_str)
.unwrap_or("")
.to_owned();
ObservedResponse::Error {
error: ObservedError { code, message },
}
},
)
},
|result| ObservedResponse::Result {
result: result.clone(),
},
)
}
fn compare(vector: &Vector, observed: &ObservedResponse) -> VectorResult {
let pass = match (&vector.expected, observed) {
(ExpectedResponse::Result { result: expected }, ObservedResponse::Result { result }) => {
expected == result
}
(
ExpectedResponse::Error {
error:
ExpectedError {
code,
message_contains,
},
},
ObservedResponse::Error {
error:
ObservedError {
code: obs_code,
message,
},
},
) => code == obs_code && message.contains(message_contains.as_str()),
_ => false,
};
if pass {
VectorResult {
name: vector.name.clone(),
outcome: VectorOutcome::Pass,
}
} else {
VectorResult {
name: vector.name.clone(),
outcome: VectorOutcome::Fail {
diff: format_diff(&vector.expected, observed),
actual: observed.clone(),
},
}
}
}
fn format_diff(expected: &ExpectedResponse, observed: &ObservedResponse) -> String {
match (expected, observed) {
(_, ObservedResponse::Transport { status, body }) => {
format!("the transport answered {status} with body {body}")
}
(ExpectedResponse::Error { error: expected }, ObservedResponse::Error { error }) => {
format!(
"expected an error with code {} (containing {:?}), got code {} message {}",
expected.code, expected.message_contains, error.code, error.message
)
}
(ExpectedResponse::Error { error: expected }, ObservedResponse::Result { result }) => {
format!(
"expected an error with code {} (containing {:?}), got result {}",
expected.code, expected.message_contains, result
)
}
(ExpectedResponse::Result { result: expected }, ObservedResponse::Result { result }) => {
format!("expected result {expected}, got {result}")
}
(ExpectedResponse::Result { .. }, ObservedResponse::Error { error }) => {
format!(
"expected a result, got an error envelope with code {} message {}",
error.code, error.message
)
}
}
}
pub fn parse_jsonl(contents: &str) -> Result<Vec<Vector>, JsonlError> {
let mut out = Vec::new();
for (line_number, line) in contents.lines().enumerate() {
let trimmed = line.trim();
if trimmed.is_empty() || trimmed.starts_with('#') {
continue;
}
let vector: Vector = serde_json::from_str(trimmed).map_err(|error| JsonlError {
line: line_number + 1,
message: format!("{error}"),
})?;
out.push(vector);
}
Ok(out)
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct JsonlError {
pub line: usize,
pub message: String,
}
impl core::fmt::Display for JsonlError {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
write!(f, "line {}: {}", self.line, self.message)
}
}
impl std::error::Error for JsonlError {}
pub const DEFAULT_PROTOCOL_VERSION: A2aVersion = A2aVersion::V1_0;
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn the_public_types_are_constructible() {
let vector = Vector {
name: "sample".to_owned(),
request: serde_json::json!({"jsonrpc": "2.0", "id": 1, "method": "ping"}),
expected: ExpectedResponse::Result {
result: serde_json::json!({}),
},
};
let result = VectorResult {
name: vector.name,
outcome: VectorOutcome::Pass,
};
let _ = ObservedResponse::Result {
result: serde_json::json!({}),
};
let _ = ObservedResponse::Error {
error: ObservedError {
code: -32601,
message: "Method not found".to_owned(),
},
};
let _ = ObservedResponse::Transport {
status: 500,
body: "internal".to_owned(),
};
assert_eq!(result.name, "sample");
assert_eq!(DEFAULT_PROTOCOL_VERSION, A2aVersion::V1_0);
}
#[test]
fn parse_jsonl_round_trips_a_known_good_fixture() -> Result<(), JsonlError> {
let fixture = "{\"name\": \"a\", \"request\": {}, \"expected\": {\"result\": {}}}\n{\"name\": \"b\", \"request\": {}, \"expected\": {\"error\": {\"code\": -1, \"message_contains\": \"x\"}}}\n";
let vectors = parse_jsonl(fixture)?;
assert_eq!(vectors.len(), 2);
assert_eq!(vectors.first().map(|v| v.name.as_str()), Some("a"));
assert_eq!(vectors.get(1).map(|v| v.name.as_str()), Some("b"));
Ok(())
}
#[test]
fn parse_jsonl_names_the_offending_line() -> Result<(), String> {
let fixture =
"{\"name\": \"ok\", \"request\": {}, \"expected\": {\"result\": {}}}\nbad json\n";
match parse_jsonl(fixture) {
Ok(vectors) => Err(format!(
"a malformed fixture should be an error, got {vectors_len} vectors",
vectors_len = vectors.len()
)),
Err(error) => {
if error.line == 2 {
Ok(())
} else {
Err(format!(
"the line number should be 2, got {line}",
line = error.line
))
}
}
}
}
#[test]
fn jsonl_error_renders_one_line() {
let error = JsonlError {
line: 3,
message: "expected `,` at line 1".to_owned(),
};
let text = format!("{error}");
assert!(text.contains('3'), "the line number should appear: {text}");
assert!(
text.contains("expected"),
"the serde message should appear: {text}"
);
}
#[tokio::test]
async fn run_against_in_process_runs_zero_vectors() -> Result<(), &'static str> {
let report = run_against_in_process(
Arc::new(DeterministicAgentForUnit) as Arc<dyn A2aRequestHandler>,
A2aVersion::V1_0,
&[],
)
.await;
if report.is_empty() {
Ok(())
} else {
Err("an empty input should produce an empty report")
}
}
#[tokio::test]
async fn run_one_produces_a_pass_for_a_well_known_response() {
let vector = Vector {
name: "sample".to_owned(),
request: serde_json::json!({
"jsonrpc": "2.0",
"id": 1,
"method": "NonExistent",
"params": {}
}),
expected: ExpectedResponse::Error {
error: ExpectedError {
code: -32601,
message_contains: "Method not found".to_owned(),
},
},
};
let router = crate::adapters::primary::a2a::jsonrpc_serve::serve_jsonrpc(
"/a2a",
Arc::new(DeterministicAgentForUnit) as Arc<dyn A2aRequestHandler>,
A2aVersion::V1_0,
);
let result = run_one(&router, &vector).await;
assert!(matches!(result.outcome, VectorOutcome::Pass));
}
struct DeterministicAgentForUnit;
impl A2aRequestHandler for DeterministicAgentForUnit {
fn dispatch(
&self,
_method: crate::adapters::secondary::a2a::A2aMethod,
_params: Option<Value>,
_context: &crate::application::a2a::A2aRequestContext,
) -> std::pin::Pin<
Box<
dyn std::future::Future<Output = Result<Value, crate::application::a2a::A2aError>>
+ Send
+ '_,
>,
> {
Box::pin(async move { Err(crate::application::a2a::A2aError::MethodNotFound) })
}
}
}