Skip to content

Commit cb86c74

Browse files
imuromtsevimuromtsev
authored
Merge pull request #12 from explyt/imuromtsev/await-listening-stream-in-roots-tests
test: Await the listening SSE stream before roots notifications (backport upstream modelcontextprotocol#1119)
2 parents b2a6964 + b721af6 commit cb86c74

3 files changed

Lines changed: 58 additions & 4 deletions

File tree

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

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,13 @@ public abstract class AbstractMcpClientServerIntegrationTests {
7878

7979
abstract protected McpServer.SyncSpecification<?> prepareSyncServerBuilder();
8080

81+
// There is, for Streamable HTTP, a race condition between establishing the SSE stream
82+
// and the server sending notifications. This breaks some `roots/list` tests (and
83+
// could in theory break sampling and elicitation tests). This utility method allows
84+
// delaying the test until the stream is opened.
85+
protected void awaitClientStreamEstablished() {
86+
}
87+
8188
@ParameterizedTest(name = "{0} : {displayName} ")
8289
@MethodSource("clientsForTesting")
8390
void simple(String clientType) {
@@ -600,6 +607,7 @@ void testRootsSuccess(String clientType) {
600607

601608
InitializeResult initResult = mcpClient.initialize();
602609
assertThat(initResult).isNotNull();
610+
awaitClientStreamEstablished();
603611

604612
assertThat(rootsRef.get()).isNull();
605613

@@ -686,7 +694,7 @@ void testRootsNotificationWithEmptyRootsList(String clientType) {
686694

687695
InitializeResult initResult = mcpClient.initialize();
688696
assertThat(initResult).isNotNull();
689-
697+
awaitClientStreamEstablished();
690698
mcpClient.rootsListChangedNotification();
691699

692700
await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
@@ -719,7 +727,7 @@ void testRootsWithMultipleHandlers(String clientType) {
719727
.build()) {
720728

721729
assertThat(mcpClient.initialize()).isNotNull();
722-
730+
awaitClientStreamEstablished();
723731
mcpClient.rootsListChangedNotification();
724732

725733
await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {
@@ -752,7 +760,7 @@ void testRootsServerCloseWithActiveSubscription(String clientType) {
752760

753761
InitializeResult initResult = mcpClient.initialize();
754762
assertThat(initResult).isNotNull();
755-
763+
awaitClientStreamEstablished();
756764
mcpClient.rootsListChangedNotification();
757765

758766
await().atMost(Duration.ofSeconds(5)).untilAsserted(() -> {

‎mcp-test/src/test/java/io/modelcontextprotocol/server/HttpServletStreamableIntegrationTests.java‎

Lines changed: 12 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
import io.modelcontextprotocol.server.McpServer.AsyncSpecification;
1616
import io.modelcontextprotocol.server.McpServer.SyncSpecification;
1717
import io.modelcontextprotocol.server.transport.HttpServletStreamableServerTransportProvider;
18+
import io.modelcontextprotocol.server.transport.SseStreamOpenedFilter;
1819
import io.modelcontextprotocol.server.transport.TomcatTestUtil;
1920
import jakarta.servlet.http.HttpServletRequest;
2021
import org.apache.catalina.LifecycleException;
@@ -26,6 +27,7 @@
2627
import org.junit.jupiter.params.provider.Arguments;
2728

2829
import static org.assertj.core.api.Assertions.assertThat;
30+
import static org.awaitility.Awaitility.await;
2931

3032
@Timeout(15)
3133
class HttpServletStreamableIntegrationTests extends AbstractMcpClientServerIntegrationTests {
@@ -36,6 +38,8 @@ class HttpServletStreamableIntegrationTests extends AbstractMcpClientServerInteg
3638

3739
private HttpServletStreamableServerTransportProvider mcpServerTransportProvider;
3840

41+
private SseStreamOpenedFilter sseStreamOpenedFilter;
42+
3943
private Tomcat tomcat;
4044

4145
static Stream<Arguments> clientsForTesting() {
@@ -51,7 +55,8 @@ public void before() {
5155
.keepAliveInterval(Duration.ofSeconds(1))
5256
.build();
5357

54-
tomcat = TomcatTestUtil.createTomcatServer("", PORT, mcpServerTransportProvider);
58+
sseStreamOpenedFilter = new SseStreamOpenedFilter();
59+
tomcat = TomcatTestUtil.createTomcatServer("", PORT, mcpServerTransportProvider, sseStreamOpenedFilter);
5560
try {
5661
tomcat.start();
5762
assertThat(tomcat.getServer().getState()).isEqualTo(LifecycleState.STARTED);
@@ -67,6 +72,12 @@ public void before() {
6772
.build()).requestTimeout(Duration.ofHours(10)));
6873
}
6974

75+
@Override
76+
protected void awaitClientStreamEstablished() {
77+
await("MCP client SSE stream opened").atMost(Duration.ofSeconds(5))
78+
.until(sseStreamOpenedFilter::isSseStreamOpened);
79+
}
80+
7081
@Override
7182
protected AsyncSpecification<?> prepareAsyncServerBuilder() {
7283
return McpServer.async(this.mcpServerTransportProvider);

‎mcp-test/src/test/java/io/modelcontextprotocol/server/transport/SseStreamOpenedFilter.java‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,35 @@
1+
/*
2+
* Copyright 2026 - 2026 the original author or authors.
3+
*/
4+
5+
package io.modelcontextprotocol.server.transport;
6+
7+
import java.io.IOException;
8+
import java.util.concurrent.atomic.AtomicBoolean;
9+
10+
import jakarta.servlet.Filter;
11+
import jakarta.servlet.FilterChain;
12+
import jakarta.servlet.ServletException;
13+
import jakarta.servlet.ServletRequest;
14+
import jakarta.servlet.ServletResponse;
15+
import jakarta.servlet.http.HttpServletRequest;
16+
17+
public class SseStreamOpenedFilter implements Filter {
18+
19+
private final AtomicBoolean sseStreamOpened = new AtomicBoolean();
20+
21+
@Override
22+
public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain)
23+
throws IOException, ServletException {
24+
chain.doFilter(request, response);
25+
if (request instanceof HttpServletRequest httpRequest && "GET".equals(httpRequest.getMethod())
26+
&& httpRequest.isAsyncStarted()) {
27+
this.sseStreamOpened.set(true);
28+
}
29+
}
30+
31+
public boolean isSseStreamOpened() {
32+
return this.sseStreamOpened.get();
33+
}
34+
35+
}

0 commit comments

Comments
 (0)