test: tighten rust a2a coverage checks

Signed-off-by: lucarlig <luca.carlig@ibm.com>

Author: lucarlig

tools_rust/a2a_runtime/src/event_store.rs

@@ -335,6 +335,79 @@ impl EventStore {
         let meta_key = format!("{KEY_PREFIX}:{task_id}:meta");
         let _ = self.storage.hset(&meta_key, "stream_active", "0").await;
     }
+
+    #[doc(hidden)]
+    pub fn seeded_for_test(events: Vec<StoredEvent>, stream_active: bool) -> Self {
+        #[derive(Clone)]
+        struct SeededEventStoreStorage {
+            events: Arc<Vec<StoredEvent>>,
+            active: bool,
+        }
+
+        #[async_trait]
+        impl EventStoreStorage for SeededEventStoreStorage {
+            async fn store_event(
+                &self,
+                _meta_key: &str,
+                _events_key: &str,
+                _messages_key: &str,
+                _event_id: &str,
+                _payload_json: &str,
+                _ttl_secs: u64,
+                _max_events: usize,
+            ) -> Result<i64, String> {
+                Err("store not supported in seeded test store".to_string())
+            }
+
+            async fn replay_entries(
+                &self,
+                _events_key: &str,
+                after_sequence: i64,
+            ) -> Result<Vec<(String, f64)>, String> {
+                Ok(self
+                    .events
+                    .iter()
+                    .filter(|event| event.sequence > after_sequence)
+                    .map(|event| (event.event_id.clone(), event.sequence as f64))
+                    .collect())
+            }
+
+            async fn payloads(
+                &self,
+                _messages_key: &str,
+                event_ids: &[String],
+            ) -> Result<Vec<Option<String>>, String> {
+                Ok(event_ids
+                    .iter()
+                    .map(|event_id| {
+                        self.events
+                            .iter()
+                            .find(|event| &event.event_id == event_id)
+                            .map(|event| event.payload.clone())
+                    })
+                    .collect())
+            }
+
+            async fn hget(&self, _key: &str, _field: &str) -> Result<Option<String>, String> {
+                Ok(Some(if self.active { "1" } else { "0" }.to_string()))
+            }
+
+            async fn hset(&self, _key: &str, _field: &str, _value: &str) -> Result<(), String> {
+                Ok(())
+            }
+        }
+
+        let (flush_tx, _flush_rx) = mpsc::channel(1);
+        Self::new_with_storage(
+            Arc::new(SeededEventStoreStorage {
+                events: Arc::new(events),
+                active: stream_active,
+            }),
+            256,
+            60,
+            flush_tx,
+        )
+    }
 }
 
 // ---------------------------------------------------------------------------

tools_rust/a2a_runtime/src/lib.rs

@@ -211,6 +211,14 @@ pub mod test_support {
 
     /// Build the Axum app without starting a listener.
     pub fn build_app(config: RuntimeConfig) -> axum::Router {
+        build_app_with_event_store(config, None)
+    }
+
+    /// Build the Axum app with a test-supplied event store.
+    pub fn build_app_with_event_store(
+        config: RuntimeConfig,
+        event_store: Option<Arc<crate::event_store::EventStore>>,
+    ) -> axum::Router {
         let client = build_http_client(&config).expect("failed to build reqwest client");
         let config_arc = Arc::new(config);
 
@@ -247,7 +255,7 @@ pub mod test_support {
             redis_pool: None,
             agent_cache,
             session_manager: None,
-            event_store: None,
+            event_store,
         };
 
         server::router(state)

tools_rust/a2a_runtime/src/server.rs

@@ -1112,7 +1112,7 @@ async fn proxy_to_backend(
         .client
         .request(reqwest_method, &url)
         .headers(forwarded)
-        .body(body.to_vec())
+        .body(body)
         .send()
         .await
         .map_err(|e| {
@@ -1484,4 +1484,51 @@ mod tests {
         let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
         assert_eq!(&body[..], b"backend-body");
     }
+
+    #[tokio::test]
+    async fn handle_streaming_method_replays_from_store_when_last_event_id_is_present() {
+        let mut state = test_state("http://127.0.0.1:1".to_string());
+        state.event_store = Some(Arc::new(crate::event_store::EventStore::seeded_for_test(
+            vec![
+                crate::event_store::StoredEvent {
+                    event_id: "evt-1".to_string(),
+                    sequence: 1,
+                    event_type: "unknown".to_string(),
+                    payload: r#"{"status":"queued"}"#.to_string(),
+                },
+                crate::event_store::StoredEvent {
+                    event_id: "evt-2".to_string(),
+                    sequence: 2,
+                    event_type: "status".to_string(),
+                    payload: r#"{"status":"working"}"#.to_string(),
+                },
+            ],
+            false,
+        )));
+
+        let response = handle_streaming_method(
+            &state,
+            "test-agent",
+            &json!({"params": {"id": "task-123"}}),
+            &HashMap::from([("last-event-id".to_string(), "task-123:0".to_string())]),
+            &json!({"sub": "user"}),
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(response.status(), StatusCode::OK);
+        assert_eq!(
+            response
+                .headers()
+                .get("content-type")
+                .and_then(|v| v.to_str().ok()),
+            Some("text/event-stream")
+        );
+
+        let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
+        let text = String::from_utf8_lossy(&body);
+        assert!(text.contains("id: evt-1:1"), "expected first replayed id, body: {text}");
+        assert!(text.contains("data: {\"status\":\"queued\"}"), "expected first payload, body: {text}");
+        assert!(text.contains("id: evt-2:2"), "expected second replayed id, body: {text}");
+    }
 }

tools_rust/a2a_runtime/tests/integration.rs

@@ -15,8 +15,9 @@ use axum::http::{Request, StatusCode};
 use contextforge_a2a_runtime::config::RuntimeConfig;
 use http_body_util::BodyExt;
 use serde_json::{Value, json};
+use std::sync::Arc;
 use tower::ServiceExt;
-use wiremock::matchers::{method, path};
+use wiremock::matchers::{body_json, method, path};
 use wiremock::{Mock, MockServer, ResponseTemplate};
 use test_helpers::*;
 
@@ -1652,6 +1653,12 @@ async fn test_a2a_invoke_create_push_config_routes_to_python() {
 
     Mock::given(method("POST"))
         .and(path_regex(".*push/create$"))
+        .and(body_json(json!({
+            "a2a_agent_id": "agent-001",
+            "task_id": "task-1",
+            "webhook_url": "https://example.com/hook",
+            "enabled": true
+        })))
         .respond_with(ResponseTemplate::new(200).set_body_json(json!({
             "config_id": "cfg-1",
             "enabled": true
@@ -1672,7 +1679,12 @@ async fn test_a2a_invoke_create_push_config_routes_to_python() {
             "jsonrpc": "2.0",
             "method": "CreateTaskPushNotificationConfig",
             "id": 8,
-            "params": {"task_id": "task-1", "webhook_url": "https://example.com/hook", "enabled": true}
+            "params": {
+                "a2a_agent_id": "agent-001",
+                "task_id": "task-1",
+                "webhook_url": "https://example.com/hook",
+                "enabled": true
+            }
         }),
     )
     .await;
@@ -1706,6 +1718,9 @@ async fn test_a2a_invoke_get_push_config_routes_to_python() {
 
     Mock::given(method("POST"))
         .and(path_regex(".*push/get$"))
+        .and(body_json(json!({
+            "task_id": "task-1"
+        })))
         .respond_with(ResponseTemplate::new(200).set_body_json(json!({
             "config_id": "cfg-1",
             "enabled": true
@@ -1726,7 +1741,7 @@ async fn test_a2a_invoke_get_push_config_routes_to_python() {
             "jsonrpc": "2.0",
             "method": "GetTaskPushNotificationConfig",
             "id": 9,
-            "params": {"config_id": "cfg-1"}
+            "params": {"task_id": "task-1"}
         }),
     )
     .await;
@@ -1760,6 +1775,9 @@ async fn test_a2a_invoke_list_push_configs_routes_to_python() {
 
     Mock::given(method("POST"))
         .and(path_regex(".*push/list$"))
+        .and(body_json(json!({
+            "task_id": "task-1"
+        })))
         .respond_with(ResponseTemplate::new(200).set_body_json(json!({
             "configs": [{"config_id": "cfg-1"}, {"config_id": "cfg-2"}]
         })))
@@ -1813,6 +1831,9 @@ async fn test_a2a_invoke_delete_push_config_routes_to_python() {
 
     Mock::given(method("POST"))
         .and(path_regex(".*push/delete$"))
+        .and(body_json(json!({
+            "config_id": "cfg-1"
+        })))
         .respond_with(ResponseTemplate::new(200).set_body_json(json!({
             "deleted": true
         })))
@@ -2309,6 +2330,13 @@ async fn test_streaming_method_forwards_sse_stream() {
 
     let response = app.oneshot(request).await.unwrap();
     assert_eq!(response.status(), StatusCode::OK);
+    assert_eq!(
+        response
+            .headers()
+            .get("content-type")
+            .and_then(|v| v.to_str().ok()),
+        Some("text/event-stream")
+    );
 
     let body = String::from_utf8_lossy(
         &response.into_body().collect().await.unwrap().to_bytes(),
@@ -2975,3 +3003,72 @@ async fn test_streaming_method_with_last_event_id_without_redis() {
 
     mock_server.verify().await;
 }
+
+#[tokio::test]
+async fn test_streaming_method_replays_from_store_when_last_event_id_present() {
+    let mock_server = MockServer::start().await;
+    setup_auth_mock(&mock_server, 1).await;
+    setup_authz_mock(&mock_server, 1).await;
+
+    let mut config = default_test_config();
+    config.backend_base_url = mock_server.uri();
+    config.auth_secret = Some("test-secret".to_string());
+    let event_store = Arc::new(contextforge_a2a_runtime::event_store::EventStore::seeded_for_test(
+        vec![
+            contextforge_a2a_runtime::event_store::StoredEvent {
+                event_id: "evt-1".to_string(),
+                sequence: 1,
+                event_type: "unknown".to_string(),
+                payload: r#"{"status":"queued"}"#.to_string(),
+            },
+            contextforge_a2a_runtime::event_store::StoredEvent {
+                event_id: "evt-2".to_string(),
+                sequence: 2,
+                event_type: "unknown".to_string(),
+                payload: r#"{"status":"working"}"#.to_string(),
+            },
+        ],
+        false,
+    ));
+    let app = contextforge_a2a_runtime::test_support::build_app_with_event_store(
+        config,
+        Some(event_store),
+    );
+
+    let request = Request::builder()
+        .method("POST")
+        .uri("/a2a/test-agent/invoke")
+        .header("content-type", "application/json")
+        .header("last-event-id", "task-123:0")
+        .body(Body::from(
+            serde_json::to_vec(&json!({
+                "jsonrpc": "2.0",
+                "method": "SendStreamingMessage",
+                "id": 1,
+                "params": {"id": "task-123", "message": {"role": "ROLE_USER", "parts": [{"text": "hello"}]}}
+            }))
+            .unwrap(),
+        ))
+        .unwrap();
+
+    let response = app.oneshot(request).await.unwrap();
+    assert_eq!(response.status(), StatusCode::OK);
+    assert_eq!(
+        response
+            .headers()
+            .get("content-type")
+            .and_then(|v| v.to_str().ok()),
+        Some("text/event-stream")
+    );
+
+    let body = String::from_utf8_lossy(
+        &response.into_body().collect().await.unwrap().to_bytes(),
+    )
+    .to_string();
+    assert!(body.contains("id: evt-1:1"), "expected first replayed event, body: {body}");
+    assert!(body.contains("id: evt-2:2"), "expected second replayed event, body: {body}");
+    assert!(body.contains("data: {\"status\":\"queued\"}"), "expected queued payload, body: {body}");
+    assert!(body.contains("data: {\"status\":\"working\"}"), "expected working payload, body: {body}");
+
+    mock_server.verify().await;
+}