jamesnetherton commented on code in PR #8820: URL: https://github.com/apache/camel-quarkus/pull/8820#discussion_r3519089968
########## integration-tests/a2a/src/main/java/org/apache/camel/quarkus/component/a2a/it/A2aResource.java: ########## @@ -0,0 +1,320 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.quarkus.component.a2a.it; + +import java.util.List; + +import com.fasterxml.jackson.databind.ObjectMapper; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import jakarta.ws.rs.Consumes; +import jakarta.ws.rs.DELETE; +import jakarta.ws.rs.GET; +import jakarta.ws.rs.POST; +import jakarta.ws.rs.Path; +import jakarta.ws.rs.Produces; +import jakarta.ws.rs.QueryParam; +import jakarta.ws.rs.core.MediaType; +import org.apache.camel.CamelContext; +import org.apache.camel.Endpoint; +import org.apache.camel.Exchange; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.component.a2a.A2AConstants; +import org.apache.camel.component.a2a.A2AEndpoint; +import org.apache.camel.component.a2a.model.AgentCard; +import org.apache.camel.component.a2a.model.Task; +import org.apache.camel.component.a2a.model.TaskPushNotificationConfig; +import org.apache.camel.component.a2a.model.TextPart; +import org.apache.camel.component.a2a.util.A2AJsonMapper; +import org.eclipse.microprofile.config.inject.ConfigProperty; + +@Path("/a2a") +@ApplicationScoped +public class A2aResource { + + @Inject + CamelContext camelContext; + + @Inject + ProducerTemplate producerTemplate; + + @ConfigProperty(name = "quarkus.http.test-port", defaultValue = "8081") + int httpPort; + + @Path("/send") + @POST + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.APPLICATION_JSON) + public String sendMessage(String message) { + Exchange result = producerTemplate.request("direct:send-message", exchange -> { + exchange.getMessage().setBody(message); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + Task task = result.getMessage().getBody(Task.class); + return String.format("{\"taskId\":\"%s\",\"contextId\":\"%s\",\"state\":\"%s\"}", + task.id(), task.contextId(), task.status().state().name()); + } + + @Path("/send-payload") + @POST + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.TEXT_PLAIN) + public String sendMessagePayload(String message) { + Exchange result = producerTemplate.request("direct:send-message-payload", exchange -> { + exchange.getMessage().setBody(message); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + return result.getMessage().getBody(String.class); + } + + @Path("/send-raw") + @POST + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.TEXT_PLAIN) + public String sendMessageRaw(String message) { + Exchange result = producerTemplate.request("direct:send-message-raw", exchange -> { + exchange.getMessage().setBody(message); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + return result.getMessage().getBody(String.class); + } + + @Path("/get-task") + @GET + @Produces(MediaType.APPLICATION_JSON) + public String getTask(@QueryParam("taskId") String taskId) { + Exchange result = producerTemplate.request("direct:get-task", exchange -> { + exchange.getMessage().setHeader(A2AConstants.TASK_ID, taskId); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + Task task = result.getMessage().getBody(Task.class); + return String.format("{\"taskId\":\"%s\",\"state\":\"%s\"}", + task.id(), task.status().state().name()); + } + + @Path("/cancel-task") + @POST + @Produces(MediaType.APPLICATION_JSON) + public String cancelTask(@QueryParam("taskId") String taskId) { + Exchange result = producerTemplate.request("direct:cancel-task", exchange -> { + exchange.getMessage().setHeader(A2AConstants.TASK_ID, taskId); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + Task task = result.getMessage().getBody(Task.class); + return String.format("{\"taskId\":\"%s\",\"state\":\"%s\"}", + task.id(), task.status().state().name()); + } + + @Path("/create-rest-endpoint") + @GET + @Produces(MediaType.TEXT_PLAIN) + public String createRestEndpoint() { + try { + camelContext.getEndpoint("a2a:test?protocolBinding=REST&validateAuth=false"); + return "unexpected-success"; + } catch (Exception e) { + Throwable cause = e; + while (cause.getCause() != null) { + cause = cause.getCause(); + } + return cause.getMessage(); + } + } + + // --- New endpoints for expanded test coverage --- + + @Path("/list-tasks") + @GET + @Produces(MediaType.APPLICATION_JSON) + @SuppressWarnings("unchecked") + public String listTasks(@QueryParam("contextId") String contextId) { + Exchange result = producerTemplate.request("direct:list-tasks", exchange -> { + if (contextId != null) { + exchange.getMessage().setHeader(A2AConstants.LIST_CONTEXT_ID, contextId); + } + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + Object body = result.getMessage().getBody(); + if (body instanceof List) { + return String.format("{\"count\":%d}", ((List<Task>) body).size()); + } + return "{\"count\":0}"; + } + + @Path("/send-context") + @POST + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.APPLICATION_JSON) + public String sendMessageWithContext(String message, @QueryParam("contextId") String contextId) { + Exchange result = producerTemplate.request("direct:send-message", exchange -> { + exchange.getMessage().setBody(message); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + if (contextId != null) { + exchange.getMessage().setHeader(A2AConstants.CONTEXT_ID, contextId); + } + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + Task task = result.getMessage().getBody(Task.class); + return String.format("{\"taskId\":\"%s\",\"contextId\":\"%s\",\"state\":\"%s\"}", + task.id(), task.contextId(), task.status().state().name()); + } + + @Path("/send-pojo") + @POST + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.APPLICATION_JSON) + public String sendToPojo(String message) { + Exchange result = producerTemplate.request("direct:send-to-pojo", exchange -> { + exchange.getMessage().setBody(message); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + Task task = result.getMessage().getBody(Task.class); + String latestText = ""; + if (task.latest() != null && task.latest().parts() != null) { + for (Object part : task.latest().parts()) { + if (part instanceof TextPart tp) { + latestText = tp.text(); + break; + } + } + } + return String.format("{\"taskId\":\"%s\",\"state\":\"%s\",\"response\":\"%s\"}", + task.id(), task.status().state().name(), latestText); + } + + @Path("/send-to-push") + @POST + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.APPLICATION_JSON) + public String sendToPush(String message) { + Exchange result = producerTemplate.request("direct:send-to-push", exchange -> { + exchange.getMessage().setBody(message); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + Task task = result.getMessage().getBody(Task.class); + return String.format("{\"taskId\":\"%s\",\"contextId\":\"%s\",\"state\":\"%s\"}", + task.id(), task.contextId(), task.status().state().name()); + } + + @Path("/agent-card-roundtrip") + @GET + @Produces(MediaType.APPLICATION_JSON) + public String agentCardRoundTrip() { + try { + ObjectMapper mapper = A2AJsonMapper.instance(); + Endpoint endpoint = camelContext.getEndpoints().stream() + .filter(ep -> ep instanceof A2AEndpoint) + .filter(ep -> ep.getEndpointUri().contains("full-agent-card")) + .findFirst() + .orElseThrow(() -> new IllegalStateException("full-agent-card endpoint not found")); + AgentCard card = ((A2AEndpoint) endpoint).getResolvedCard(); + String json = mapper.writeValueAsString(card); + AgentCard roundTripped = mapper.readValue(json, AgentCard.class); + return mapper.writeValueAsString(roundTripped); + } catch (Exception e) { + return String.format("{\"error\":\"%s\"}", e.getMessage()); + } + } + + @Path("/push-config/create") + @POST + @Produces(MediaType.APPLICATION_JSON) + public String createPushConfig(@QueryParam("taskId") String taskId, @QueryParam("url") String url) { + Exchange result = producerTemplate.request("direct:push-config-create", exchange -> { + exchange.getMessage().setHeader(A2AConstants.TASK_ID, taskId); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + TaskPushNotificationConfig config = new TaskPushNotificationConfig(); + config.setTaskId(taskId); + config.setUrl(url); + exchange.getMessage().setBody(config); + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + TaskPushNotificationConfig config = result.getMessage().getBody(TaskPushNotificationConfig.class); + return String.format("{\"id\":\"%s\",\"taskId\":\"%s\",\"url\":\"%s\"}", + config.getId(), config.getTaskId(), config.getUrl()); + } + + @Path("/push-config/get") + @GET + @Produces(MediaType.APPLICATION_JSON) + public String getPushConfig(@QueryParam("taskId") String taskId, @QueryParam("configId") String configId) { + Exchange result = producerTemplate.request("direct:push-config-get", exchange -> { + exchange.getMessage().setHeader(A2AConstants.TASK_ID, taskId); + exchange.getMessage().setHeader(A2AConstants.PUSH_CONFIG_ID, configId); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + TaskPushNotificationConfig config = result.getMessage().getBody(TaskPushNotificationConfig.class); + return String.format("{\"id\":\"%s\",\"taskId\":\"%s\",\"url\":\"%s\"}", + config.getId(), config.getTaskId(), config.getUrl()); + } + + @Path("/push-config/list") + @GET + @Produces(MediaType.APPLICATION_JSON) + @SuppressWarnings("unchecked") + public String listPushConfigs(@QueryParam("taskId") String taskId) { Review Comment: Method renamed as suggested. ########## integration-tests/a2a/src/test/java/org/apache/camel/quarkus/component/a2a/it/A2aTest.java: ########## @@ -0,0 +1,575 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.quarkus.component.a2a.it; + +import io.quarkus.test.common.QuarkusTestResource; +import io.quarkus.test.junit.QuarkusTest; +import io.restassured.RestAssured; +import io.restassured.http.ContentType; +import org.apache.camel.component.a2a.A2AConstants; +import org.eclipse.microprofile.config.ConfigProvider; +import org.junit.jupiter.api.Test; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.notNullValue; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +@QuarkusTest +@QuarkusTestResource(A2aKeycloakTestResource.class) +class A2aTest { + + @Test + void agentCardServed() { + RestAssured.get("/.well-known/agent-card.json") + .then() + .statusCode(200) + .body( + "name", is("Test Agent"), + "version", is("1.0.0"), + "description", is("A Quarkus test agent")); + } + + @Test + void sendMessageJsonRpc() { + RestAssured.given() + .contentType("application/json") + .body(jsonRpcSendMessage("msg-1", "req-1", "{\"text\":\"Hello A2A\"}")) + .post("/") + .then() + .statusCode(200) + .body( + "jsonrpc", is("2.0"), + "result.task.id", notNullValue(), + "result.task.contextId", notNullValue(), + "id", is("req-1")); + } + + @Test + void producerSendMessagePojo() { + RestAssured.given() + .contentType(ContentType.TEXT) + .body("Hello from producer") + .post("/a2a/send") + .then() + .statusCode(200) + .body( + "taskId", notNullValue(), + "contextId", notNullValue(), + "state", is("COMPLETED")); + } + + @Test + void producerSendMessagePayloadDataFormat() { + String response = RestAssured.given() + .contentType(ContentType.TEXT) + .body("Hello payload") + .post("/a2a/send-payload") + .then() + .statusCode(200) + .extract().asString(); + + assertTrue( + response.contains("Echo:"), + "Expected PAYLOAD response to contain echoed text, got: " + response); + } + + @Test + void producerSendMessageRawDataFormat() { + String response = RestAssured.given() + .contentType(ContentType.TEXT) + .body("Hello raw") + .post("/a2a/send-raw") + .then() + .statusCode(200) + .extract().asString(); + + assertTrue( + response.contains("\"task\"") || response.contains("\"result\""), + "Expected RAW response to contain JSON structure, got: " + response); + } + + @Test + void producerGetTask() { + String taskId = RestAssured.given() + .contentType(ContentType.TEXT) + .body("Create task for get") + .post("/a2a/send") + .then() + .statusCode(200) + .extract() + .jsonPath() + .getString("taskId"); + + RestAssured.given() + .queryParam("taskId", taskId) + .get("/a2a/get-task") + .then() + .statusCode(200) + .body( + "taskId", is(taskId), + "state", is("COMPLETED")); + } + + @Test + void producerCancelTask() { + String taskId = RestAssured.given() + .contentType(ContentType.TEXT) + .body("Create task for cancel") + .post("/a2a/send") + .then() + .statusCode(200) + .extract() + .jsonPath() + .getString("taskId"); + + RestAssured.given() + .queryParam("taskId", taskId) + .post("/a2a/cancel-task") + .then() + .statusCode(200) + .body("error", notNullValue()); + } + + @Test + void sendStreamingMessageJsonRpc() { + String response = RestAssured.given() + .contentType("application/json") + .body(jsonRpcStreamMessage("msg-stream", "req-stream", "Stream this")) + .post("/streaming/") + .then() + .statusCode(200) + .header("Content-Type", "text/event-stream") + .extract().asString(); + + String[] events = response.split("\n\n"); + assertTrue(events.length >= 3, + "Expected at least 3 SSE events (submitted + 2 progress + completed), got " + + events.length + ": " + response); + + for (String event : events) { + assertTrue(event.startsWith("data: "), + "SSE event should start with 'data: ', got: " + event); + } + + assertTrue(response.contains("Step 1 complete"), + "Expected progress event 'Step 1 complete' in response: " + response); + assertTrue(response.contains("Step 2 complete"), + "Expected progress event 'Step 2 complete' in response: " + response); + } + + @Test + void restProtocolBindingRejected() { + RestAssured.get("/a2a/create-rest-endpoint") + .then() + .statusCode(200) + .body(containsString("REST (HTTP+JSON) protocol binding is not supported on Quarkus")); + } + + @Test + void sendMessageWithDataPart() { + RestAssured.given() + .contentType("application/json") + .body(jsonRpcSendMessage("msg-data", "req-data", + "{\"kind\":\"data\",\"data\":{\"key\":\"value\",\"count\":42}}")) + .post("/") + .then() + .statusCode(200) + .body( + "jsonrpc", is("2.0"), + "result.task.id", notNullValue(), + "result.task.status.state", is("TASK_STATE_COMPLETED"), + "id", is("req-data")); + } + + @Test + void sendMessageWithFilePart() { + RestAssured.given() + .contentType("application/json") + .body(jsonRpcSendMessage("msg-file", "req-file", + "{\"kind\":\"file\",\"raw\":\"aGVsbG8=\",\"mediaType\":\"text/plain\",\"filename\":\"test.txt\"}")) + .post("/") + .then() + .statusCode(200) + .body( + "jsonrpc", is("2.0"), + "result.task.id", notNullValue(), + "result.task.status.state", is("TASK_STATE_COMPLETED"), + "id", is("req-file")); + } + + @Test + void fullAgentCardServed() { + RestAssured.get("/full/.well-known/agent-card.json") + .then() + .statusCode(200) + .body( + "name", is("Full Feature Agent"), + "version", is("2.0.0"), + "description", is("A Quarkus test agent with all features"), + "provider.name", is("Test Provider"), + "provider.url", is("https://example.com/provider";), + "capabilities.streaming", is(true), + "capabilities.pushNotifications", is(true), + "skills[0].id", is("echo"), + "skills[0].name", is("Echo Skill"), + "skills[0].tags[0]", is("test"), + "skills[0].inputModes[0]", is("text/plain"), + "skills[1].id", is("transform"), + "defaultInputModes[0]", is("text/plain"), + "defaultOutputModes[0]", is("text/plain")); + } + + @Test + void agentCardJsonRoundTrip() { + RestAssured.get("/a2a/agent-card-roundtrip") + .then() + .statusCode(200) + .body( + "name", is("Full Feature Agent"), + "version", is("2.0.0"), + "provider.name", is("Test Provider"), + "provider.url", is("https://example.com/provider";), + "capabilities.streaming", is(true), + "capabilities.pushNotifications", is(true), + "skills[0].id", is("echo")); + } + + @Test + void producerListTasks() { + RestAssured.given() + .contentType(ContentType.TEXT) + .body("Task for list 1") + .post("/a2a/send") + .then() + .statusCode(200); + RestAssured.given() + .contentType(ContentType.TEXT) + .body("Task for list 2") + .post("/a2a/send") + .then() + .statusCode(200); + + RestAssured.get("/a2a/list-tasks") + .then() + .statusCode(200) + .body("count", greaterThanOrEqualTo(2)); + } + + @Test + void multiTurnConversation() { + String contextId = RestAssured.given() + .contentType(ContentType.TEXT) + .body("Turn 1") + .post("/a2a/send-context") + .then() + .statusCode(200) + .extract() + .jsonPath() + .getString("contextId"); + assertNotNull(contextId, "First turn should return a contextId"); + + String contextId2 = RestAssured.given() + .contentType(ContentType.TEXT) + .body("Turn 2") + .queryParam("contextId", contextId) + .post("/a2a/send-context") + .then() + .statusCode(200) + .extract() + .jsonPath() + .getString("contextId"); + + assertEquals(contextId, contextId2, "contextId should be preserved across turns"); + } + + @Test + void pushNotificationConfigCrud() { + String taskId = RestAssured.given() + .contentType("application/json") + .body(jsonRpcSendMessage("msg-push", "req-push", "{\"text\":\"Push test\"}")) + .post("/push/") + .then() + .statusCode(200) + .body("result.task.id", notNullValue()) + .extract() + .jsonPath() + .getString("result.task.id"); + assertNotNull(taskId, "Task ID should be returned"); + + String configId = RestAssured.given() + .contentType("application/json") + .body(jsonRpcRequest("CreateTaskPushNotificationConfig", "req-pc-create", + "\"taskId\":\"" + taskId + "\",\"url\":\"http://localhost:9999/webhook\"";)) + .post("/push/") + .then() + .statusCode(200) + .body("result.url", is("http://localhost:9999/webhook";)) + .extract() + .jsonPath() + .getString("result.id"); + assertNotNull(configId, "Push config ID should be returned"); + + String listRequest = jsonRpcRequest("ListTaskPushNotificationConfigs", "req-pc-list", + "\"taskId\":\"" + taskId + "\""); + + RestAssured.given() + .contentType("application/json") + .body(listRequest) + .post("/push/") + .then() + .statusCode(200) + .body("result.configs.size()", is(1)); + + RestAssured.given() + .contentType("application/json") + .body(jsonRpcRequest("GetTaskPushNotificationConfig", "req-pc-get", + "\"taskId\":\"" + taskId + "\",\"id\":\"" + configId + "\"")) + .post("/push/") + .then() + .statusCode(200) + .body("result.url", is("http://localhost:9999/webhook";)); + + RestAssured.given() + .contentType("application/json") + .body(jsonRpcRequest("DeleteTaskPushNotificationConfig", "req-pc-delete", + "\"taskId\":\"" + taskId + "\",\"id\":\"" + configId + "\"")) + .post("/push/") + .then() + .statusCode(200); + + RestAssured.given() + .contentType("application/json") + .body(listRequest) + .post("/push/") + .then() + .statusCode(200) + .body("result.configs.size()", is(0)); + } + + @Test + void streamingWithArtifactEmission() { + String response = RestAssured.given() + .contentType("application/json") + .body(jsonRpcStreamMessage("msg-art", "req-art", "Stream with artifact")) + .post("/streaming/") + .then() + .statusCode(200) + .header("Content-Type", "text/event-stream") + .extract().asString(); + + assertTrue(response.contains("artifactUpdate"), + "Expected artifact update event in SSE response: " + response); + assertTrue(response.contains("test-artifact"), + "Expected artifact name 'test-artifact' in SSE response: " + response); + } + + @Test + void consumerPojoDataFormat() { + RestAssured.given() + .contentType(ContentType.TEXT) + .body("Hello POJO") + .post("/a2a/send-pojo") + .then() + .statusCode(200) + .body( + "state", is("COMPLETED"), + "response", containsString("role=ROLE_USER")); + } + + @Test + void jsonRpcParseError() { + RestAssured.given() Review Comment: Fair point. Removed the test and also similar cases (jsonRpcMethodNotFound and jsonRpcInvalidParams). ########## integration-tests/a2a/src/test/java/org/apache/camel/quarkus/component/a2a/it/A2aTest.java: ########## @@ -0,0 +1,575 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.quarkus.component.a2a.it; + +import io.quarkus.test.common.QuarkusTestResource; +import io.quarkus.test.junit.QuarkusTest; +import io.restassured.RestAssured; +import io.restassured.http.ContentType; +import org.apache.camel.component.a2a.A2AConstants; +import org.eclipse.microprofile.config.ConfigProvider; +import org.junit.jupiter.api.Test; + +import static org.hamcrest.Matchers.containsString; +import static org.hamcrest.Matchers.greaterThanOrEqualTo; +import static org.hamcrest.Matchers.is; +import static org.hamcrest.Matchers.notNullValue; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +@QuarkusTest +@QuarkusTestResource(A2aKeycloakTestResource.class) +class A2aTest { + + @Test + void agentCardServed() { + RestAssured.get("/.well-known/agent-card.json") + .then() + .statusCode(200) + .body( + "name", is("Test Agent"), + "version", is("1.0.0"), + "description", is("A Quarkus test agent")); + } + + @Test + void sendMessageJsonRpc() { + RestAssured.given() + .contentType("application/json") + .body(jsonRpcSendMessage("msg-1", "req-1", "{\"text\":\"Hello A2A\"}")) + .post("/") + .then() + .statusCode(200) + .body( + "jsonrpc", is("2.0"), + "result.task.id", notNullValue(), + "result.task.contextId", notNullValue(), + "id", is("req-1")); + } + + @Test + void producerSendMessagePojo() { + RestAssured.given() + .contentType(ContentType.TEXT) + .body("Hello from producer") + .post("/a2a/send") + .then() + .statusCode(200) + .body( + "taskId", notNullValue(), + "contextId", notNullValue(), + "state", is("COMPLETED")); + } + + @Test + void producerSendMessagePayloadDataFormat() { + String response = RestAssured.given() + .contentType(ContentType.TEXT) + .body("Hello payload") + .post("/a2a/send-payload") + .then() + .statusCode(200) + .extract().asString(); + + assertTrue( + response.contains("Echo:"), + "Expected PAYLOAD response to contain echoed text, got: " + response); + } + + @Test + void producerSendMessageRawDataFormat() { + String response = RestAssured.given() + .contentType(ContentType.TEXT) + .body("Hello raw") + .post("/a2a/send-raw") + .then() + .statusCode(200) + .extract().asString(); + + assertTrue( + response.contains("\"task\"") || response.contains("\"result\""), + "Expected RAW response to contain JSON structure, got: " + response); + } + + @Test + void producerGetTask() { + String taskId = RestAssured.given() + .contentType(ContentType.TEXT) + .body("Create task for get") + .post("/a2a/send") + .then() + .statusCode(200) + .extract() + .jsonPath() + .getString("taskId"); + + RestAssured.given() + .queryParam("taskId", taskId) + .get("/a2a/get-task") + .then() + .statusCode(200) + .body( + "taskId", is(taskId), + "state", is("COMPLETED")); + } + + @Test + void producerCancelTask() { + String taskId = RestAssured.given() + .contentType(ContentType.TEXT) + .body("Create task for cancel") + .post("/a2a/send") + .then() + .statusCode(200) + .extract() + .jsonPath() + .getString("taskId"); + + RestAssured.given() + .queryParam("taskId", taskId) + .post("/a2a/cancel-task") + .then() + .statusCode(200) + .body("error", notNullValue()); Review Comment: Renamed producerCancelTask to cancelCompletedTaskReturnsError and added a comment explaining the echo agent completes immediately, so canceling returns an error. ########## integration-tests/a2a/src/main/java/org/apache/camel/quarkus/component/a2a/it/A2aResource.java: ########## @@ -0,0 +1,320 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.quarkus.component.a2a.it; + +import java.util.List; + +import com.fasterxml.jackson.databind.ObjectMapper; +import jakarta.enterprise.context.ApplicationScoped; +import jakarta.inject.Inject; +import jakarta.ws.rs.Consumes; +import jakarta.ws.rs.DELETE; +import jakarta.ws.rs.GET; +import jakarta.ws.rs.POST; +import jakarta.ws.rs.Path; +import jakarta.ws.rs.Produces; +import jakarta.ws.rs.QueryParam; +import jakarta.ws.rs.core.MediaType; +import org.apache.camel.CamelContext; +import org.apache.camel.Endpoint; +import org.apache.camel.Exchange; +import org.apache.camel.ProducerTemplate; +import org.apache.camel.component.a2a.A2AConstants; +import org.apache.camel.component.a2a.A2AEndpoint; +import org.apache.camel.component.a2a.model.AgentCard; +import org.apache.camel.component.a2a.model.Task; +import org.apache.camel.component.a2a.model.TaskPushNotificationConfig; +import org.apache.camel.component.a2a.model.TextPart; +import org.apache.camel.component.a2a.util.A2AJsonMapper; +import org.eclipse.microprofile.config.inject.ConfigProperty; + +@Path("/a2a") +@ApplicationScoped +public class A2aResource { + + @Inject + CamelContext camelContext; + + @Inject + ProducerTemplate producerTemplate; + + @ConfigProperty(name = "quarkus.http.test-port", defaultValue = "8081") + int httpPort; + + @Path("/send") + @POST + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.APPLICATION_JSON) + public String sendMessage(String message) { + Exchange result = producerTemplate.request("direct:send-message", exchange -> { + exchange.getMessage().setBody(message); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + Task task = result.getMessage().getBody(Task.class); + return String.format("{\"taskId\":\"%s\",\"contextId\":\"%s\",\"state\":\"%s\"}", + task.id(), task.contextId(), task.status().state().name()); + } + + @Path("/send-payload") + @POST + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.TEXT_PLAIN) + public String sendMessagePayload(String message) { + Exchange result = producerTemplate.request("direct:send-message-payload", exchange -> { + exchange.getMessage().setBody(message); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + return result.getMessage().getBody(String.class); + } + + @Path("/send-raw") + @POST + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.TEXT_PLAIN) + public String sendMessageRaw(String message) { + Exchange result = producerTemplate.request("direct:send-message-raw", exchange -> { + exchange.getMessage().setBody(message); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + return result.getMessage().getBody(String.class); + } + + @Path("/get-task") + @GET + @Produces(MediaType.APPLICATION_JSON) + public String getTask(@QueryParam("taskId") String taskId) { + Exchange result = producerTemplate.request("direct:get-task", exchange -> { + exchange.getMessage().setHeader(A2AConstants.TASK_ID, taskId); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + Task task = result.getMessage().getBody(Task.class); + return String.format("{\"taskId\":\"%s\",\"state\":\"%s\"}", + task.id(), task.status().state().name()); + } + + @Path("/cancel-task") + @POST + @Produces(MediaType.APPLICATION_JSON) + public String cancelTask(@QueryParam("taskId") String taskId) { + Exchange result = producerTemplate.request("direct:cancel-task", exchange -> { + exchange.getMessage().setHeader(A2AConstants.TASK_ID, taskId); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + Task task = result.getMessage().getBody(Task.class); + return String.format("{\"taskId\":\"%s\",\"state\":\"%s\"}", + task.id(), task.status().state().name()); + } + + @Path("/create-rest-endpoint") + @GET + @Produces(MediaType.TEXT_PLAIN) + public String createRestEndpoint() { + try { + camelContext.getEndpoint("a2a:test?protocolBinding=REST&validateAuth=false"); + return "unexpected-success"; + } catch (Exception e) { + Throwable cause = e; + while (cause.getCause() != null) { + cause = cause.getCause(); + } + return cause.getMessage(); + } + } + + // --- New endpoints for expanded test coverage --- + + @Path("/list-tasks") + @GET + @Produces(MediaType.APPLICATION_JSON) + @SuppressWarnings("unchecked") + public String listTasks(@QueryParam("contextId") String contextId) { + Exchange result = producerTemplate.request("direct:list-tasks", exchange -> { + if (contextId != null) { + exchange.getMessage().setHeader(A2AConstants.LIST_CONTEXT_ID, contextId); + } + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + Object body = result.getMessage().getBody(); + if (body instanceof List) { + return String.format("{\"count\":%d}", ((List<Task>) body).size()); + } + return "{\"count\":0}"; + } + + @Path("/send-context") + @POST + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.APPLICATION_JSON) + public String sendMessageWithContext(String message, @QueryParam("contextId") String contextId) { + Exchange result = producerTemplate.request("direct:send-message", exchange -> { + exchange.getMessage().setBody(message); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + if (contextId != null) { + exchange.getMessage().setHeader(A2AConstants.CONTEXT_ID, contextId); + } + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + Task task = result.getMessage().getBody(Task.class); + return String.format("{\"taskId\":\"%s\",\"contextId\":\"%s\",\"state\":\"%s\"}", + task.id(), task.contextId(), task.status().state().name()); + } + + @Path("/send-pojo") + @POST + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.APPLICATION_JSON) + public String sendToPojo(String message) { + Exchange result = producerTemplate.request("direct:send-to-pojo", exchange -> { + exchange.getMessage().setBody(message); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + Task task = result.getMessage().getBody(Task.class); + String latestText = ""; + if (task.latest() != null && task.latest().parts() != null) { + for (Object part : task.latest().parts()) { + if (part instanceof TextPart tp) { + latestText = tp.text(); + break; + } + } + } + return String.format("{\"taskId\":\"%s\",\"state\":\"%s\",\"response\":\"%s\"}", + task.id(), task.status().state().name(), latestText); + } + + @Path("/send-to-push") + @POST + @Consumes(MediaType.TEXT_PLAIN) + @Produces(MediaType.APPLICATION_JSON) + public String sendToPush(String message) { + Exchange result = producerTemplate.request("direct:send-to-push", exchange -> { + exchange.getMessage().setBody(message); + exchange.getMessage().setHeader("CamelA2APort", httpPort); + }); + if (result.getException() != null) { + return String.format("{\"error\":\"%s\"}", result.getException().getMessage()); + } + Task task = result.getMessage().getBody(Task.class); + return String.format("{\"taskId\":\"%s\",\"contextId\":\"%s\",\"state\":\"%s\"}", + task.id(), task.contextId(), task.status().state().name()); + } + + @Path("/agent-card-roundtrip") + @GET + @Produces(MediaType.APPLICATION_JSON) + public String agentCardRoundTrip() { Review Comment: Good catch. It is a redundant test. The JSON serialization / deseralization is already exercised elsewhere via code paths in the Camel component. ########## extensions/a2a/deployment/src/main/java/org/apache/camel/quarkus/component/a2a/deployment/A2aProcessor.java: ########## @@ -0,0 +1,65 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.camel.quarkus.component.a2a.deployment; + +import java.util.Set; +import java.util.stream.Collectors; + +import io.quarkus.deployment.annotations.BuildStep; +import io.quarkus.deployment.annotations.ExecutionTime; +import io.quarkus.deployment.annotations.Record; +import io.quarkus.deployment.builditem.CombinedIndexBuildItem; +import io.quarkus.deployment.builditem.FeatureBuildItem; +import io.quarkus.deployment.builditem.nativeimage.ReflectiveClassBuildItem; +import io.quarkus.deployment.pkg.steps.NativeOrNativeSourcesBuild; +import org.apache.camel.component.a2a.A2AComponent; +import org.apache.camel.quarkus.component.a2a.A2aRecorder; +import org.apache.camel.quarkus.core.deployment.spi.CamelRuntimeBeanBuildItem; +import org.jboss.jandex.ClassInfo; +import org.jboss.jandex.DotName; + +class A2aProcessor { + + private static final String FEATURE = "camel-a2a"; + + @BuildStep + FeatureBuildItem feature() { + return new FeatureBuildItem(FEATURE); + } + + @BuildStep + @Record(ExecutionTime.RUNTIME_INIT) + CamelRuntimeBeanBuildItem configureA2aComponent(A2aRecorder recorder) { + return new CamelRuntimeBeanBuildItem("a2a", A2AComponent.class.getName(), + recorder.createA2aComponent()); + } + + @BuildStep(onlyIf = NativeOrNativeSourcesBuild.class) + ReflectiveClassBuildItem registerModelClassesForReflection(CombinedIndexBuildItem combinedIndex) { + Set<String> modelClasses = combinedIndex.getIndex() + .getClassesInPackage("org.apache.camel.component.a2a.model") + .stream() + .map(ClassInfo::name) + .map(DotName::toString) + .collect(Collectors.toSet()); + + return ReflectiveClassBuildItem + .builder(modelClasses.toArray(new String[0])) + .methods(true) Review Comment: The A2A model classes use a mix of patterns — AgentCard uses @JsonDeserialize(builder=...) with setter methods on the Builder, Task is a Java Record (Jackson uses the canonical constructor + accessor methods), and TaskPushNotificationConfig is a standard bean with getters/setters. In all cases Jackson operates through methods, not direct field access, so .methods(true) should be sufficient. The native integration tests pass with this configuration. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
