Fix elicitation and resource subscription tests that deadlock on a single CPU (modelcontextprotocol#854)

Signed-off-by: Dariusz Jędrzejczyk <dariusz.jedrzejczyk@broadcom.com>

Author: chemicL

mcp-test/src/main/java/io/modelcontextprotocol/AbstractMcpClientServerIntegrationTests.java

@@ -53,7 +53,6 @@
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.ValueSource;
 import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
 
 import static io.modelcontextprotocol.util.ToolsUtils.EMPTY_JSON_SCHEMA;
 import static net.javacrumbs.jsonunit.assertj.JsonAssertions.assertThatJson;
@@ -404,6 +403,8 @@ void testCreateElicitationSuccess(String clientType) {
 			.addContent(new McpSchema.TextContent("CALL RESPONSE"))
 			.build();
 
+		AtomicReference<McpSchema.ElicitResult> elicitResultRef = new AtomicReference<>();
+
 		McpServerFeatures.AsyncToolSpecification tool = McpServerFeatures.AsyncToolSpecification.builder()
 			.tool(Tool.builder().name("tool1").description("tool1 description").inputSchema(EMPTY_JSON_SCHEMA).build())
 			.callHandler((exchange, request) -> {
@@ -414,13 +415,9 @@ void testCreateElicitationSuccess(String clientType) {
 							Map.of("type", "object", "properties", Map.of("message", Map.of("type", "string"))))
 					.build();
 
-				StepVerifier.create(exchange.createElicitation(elicitationRequest)).consumeNextWith(result -> {
-					assertThat(result).isNotNull();
-					assertThat(result.action()).isEqualTo(McpSchema.ElicitResult.Action.ACCEPT);
-					assertThat(result.content().get("message")).isEqualTo("Test message");
-				}).verifyComplete();
-
-				return Mono.just(callResponse);
+				return exchange.createElicitation(elicitationRequest)
+					.doOnNext(elicitResultRef::set)
+					.thenReturn(callResponse);
 			})
 			.build();
 
@@ -438,6 +435,11 @@ void testCreateElicitationSuccess(String clientType) {
 
 			assertThat(response).isNotNull();
 			assertThat(response).isEqualTo(callResponse);
+			assertWith(elicitResultRef.get(), result -> {
+				assertThat(result).isNotNull();
+				assertThat(result.action()).isEqualTo(McpSchema.ElicitResult.Action.ACCEPT);
+				assertThat(result.content().get("message")).isEqualTo("Test message");
+			});
 		}
 		finally {
 			mcpServer.closeGracefully().block();

mcp-test/src/main/java/io/modelcontextprotocol/client/AbstractMcpAsyncClientTests.java

@@ -610,22 +610,17 @@ void testListAllResourceTemplatesReturnsImmutableList() {
 		});
 	}
 
-	// @Test
+	@Test
 	void testResourceSubscription() {
 		withClient(createMcpTransport(), mcpAsyncClient -> {
-			StepVerifier.create(mcpAsyncClient.listResources()).consumeNextWith(resources -> {
-				if (!resources.resources().isEmpty()) {
-					Resource firstResource = resources.resources().get(0);
-
-					// Test subscribe
-					StepVerifier.create(mcpAsyncClient.subscribeResource(new SubscribeRequest(firstResource.uri())))
-						.verifyComplete();
-
-					// Test unsubscribe
-					StepVerifier.create(mcpAsyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri())))
-						.verifyComplete();
+			StepVerifier.create(mcpAsyncClient.listResources().flatMap(resources -> {
+				if (resources.resources().isEmpty()) {
+					return Mono.empty();
 				}
-			}).verifyComplete();
+				Resource firstResource = resources.resources().get(0);
+				return mcpAsyncClient.subscribeResource(new SubscribeRequest(firstResource.uri()))
+					.then(mcpAsyncClient.unsubscribeResource(new UnsubscribeRequest(firstResource.uri())));
+			})).verifyComplete();
 		});
 	}