|
7 | 7 | import java.util.HashMap; |
8 | 8 | import java.util.List; |
9 | 9 | import java.util.Map; |
| 10 | +import java.util.stream.Stream; |
10 | 11 |
|
11 | 12 | import io.modelcontextprotocol.common.McpTransportContext; |
12 | 13 | import io.modelcontextprotocol.json.McpJsonDefaults; |
13 | 14 | import io.modelcontextprotocol.modern.JsonRpc.JSONRPCNotification; |
14 | 15 | import io.modelcontextprotocol.modern.JsonRpc.JSONRPCRequest; |
15 | 16 | import io.modelcontextprotocol.modern.JsonRpc.JSONRPCResponse; |
| 17 | +import io.modelcontextprotocol.modern.JsonRpc.JSONRPCResponse.JSONRPCError; |
16 | 18 | import io.modelcontextprotocol.modern.McpSchema.CacheScope; |
| 19 | +import io.modelcontextprotocol.modern.McpSchema.ErrorCodes; |
17 | 20 | import io.modelcontextprotocol.modern.McpSchema.MetaKeys; |
18 | 21 | import io.modelcontextprotocol.modern.McpSchema.SubscriptionFilter; |
19 | 22 | import io.modelcontextprotocol.modern.McpSchema.SubscriptionsAcknowledgedParams; |
|
27 | 30 | import io.modelcontextprotocol.modern.server.feature.ServerChange; |
28 | 31 | import io.modelcontextprotocol.modern.server.feature.ToolsFeature; |
29 | 32 | import org.junit.jupiter.api.Test; |
| 33 | +import org.junit.jupiter.params.ParameterizedTest; |
| 34 | +import org.junit.jupiter.params.provider.MethodSource; |
30 | 35 | import reactor.core.publisher.Mono; |
31 | 36 | import reactor.test.StepVerifier; |
32 | 37 |
|
|
35 | 40 | import static io.modelcontextprotocol.modern.server.ModernTestFixtures.emptyTools; |
36 | 41 | import static io.modelcontextprotocol.modern.server.ModernTestFixtures.meta; |
37 | 42 | import static io.modelcontextprotocol.modern.server.ModernTestFixtures.respond; |
| 43 | +import static org.assertj.core.api.Assertions.as; |
38 | 44 | import static org.assertj.core.api.Assertions.assertThat; |
| 45 | +import static org.assertj.core.api.InstanceOfAssertFactories.STRING; |
| 46 | +import static org.assertj.core.api.InstanceOfAssertFactories.type; |
39 | 47 |
|
40 | 48 | class SubscriptionsFeatureTests { |
41 | 49 |
|
@@ -171,6 +179,77 @@ void listenAfterEarlierListenEndedReceivesChanges() { |
171 | 179 | .verifyComplete(); |
172 | 180 | } |
173 | 181 |
|
| 182 | + @Test |
| 183 | + void listenWithoutNotificationsIsRejected() { |
| 184 | + McpServer server = toolsServer(new McpChangeFeed()); |
| 185 | + |
| 186 | + Map<String, Object> params = new HashMap<>(); |
| 187 | + params.put("_meta", meta()); |
| 188 | + McpTransportResponse response = server |
| 189 | + .handle(McpTransportContext.EMPTY, new JSONRPCRequest("subscriptions/listen", 1, params)) |
| 190 | + .block(); |
| 191 | + |
| 192 | + assertNotificationsRequired(response); |
| 193 | + } |
| 194 | + |
| 195 | + @Test |
| 196 | + void listenWithNullNotificationsIsRejected() { |
| 197 | + McpServer server = toolsServer(new McpChangeFeed()); |
| 198 | + |
| 199 | + Map<String, Object> params = new HashMap<>(); |
| 200 | + params.put("_meta", meta()); |
| 201 | + params.put("notifications", null); |
| 202 | + McpTransportResponse response = server |
| 203 | + .handle(McpTransportContext.EMPTY, new JSONRPCRequest("subscriptions/listen", 1, params)) |
| 204 | + .block(); |
| 205 | + |
| 206 | + assertNotificationsRequired(response); |
| 207 | + } |
| 208 | + |
| 209 | + @ParameterizedTest |
| 210 | + @MethodSource("nonObjectNotifications") |
| 211 | + void listenWithNonObjectNotificationsIsRejected(Object notifications) { |
| 212 | + McpServer server = toolsServer(new McpChangeFeed()); |
| 213 | + |
| 214 | + Map<String, Object> params = new HashMap<>(); |
| 215 | + params.put("_meta", meta()); |
| 216 | + params.put("notifications", notifications); |
| 217 | + McpTransportResponse response = server |
| 218 | + .handle(McpTransportContext.EMPTY, new JSONRPCRequest("subscriptions/listen", 1, params)) |
| 219 | + .block(); |
| 220 | + |
| 221 | + assertNotificationsRequired(response); |
| 222 | + } |
| 223 | + |
| 224 | + static Stream<Object> nonObjectNotifications() { |
| 225 | + // "" and [] matter in particular: a mapper configured with |
| 226 | + // ACCEPT_EMPTY_STRING_AS_NULL_OBJECT / ACCEPT_EMPTY_ARRAY_AS_NULL_OBJECT would |
| 227 | + // coerce them to null, which SubscriptionsListenRequest turns into an empty |
| 228 | + // filter. |
| 229 | + return Stream.of("", List.of(), true, 1); |
| 230 | + } |
| 231 | + |
| 232 | + @Test |
| 233 | + void emptyNotificationsIsAcknowledgedAndStaysOpen() { |
| 234 | + McpChangeFeed feed = new McpChangeFeed(); |
| 235 | + McpServer server = toolsServer(feed); |
| 236 | + |
| 237 | + Map<String, Object> params = new HashMap<>(); |
| 238 | + params.put("_meta", meta()); |
| 239 | + params.put("notifications", Map.of()); |
| 240 | + var invocation = (McpTransportResponse.Streaming) server |
| 241 | + .handle(McpTransportContext.EMPTY, new JSONRPCRequest("subscriptions/listen", 1, params)) |
| 242 | + .block(); |
| 243 | + |
| 244 | + StepVerifier.create(invocation.messages()) |
| 245 | + .assertNext(msg -> assertThat(acknowledgedNotifications((JSONRPCNotification) msg)) |
| 246 | + .isEqualTo(SubscriptionFilter.EMPTY)) |
| 247 | + .then(() -> feed.broadcast(new ServerChange.ToolsListChanged())) |
| 248 | + .then(server::closeGracefully) |
| 249 | + .expectNextMatches(msg -> msg instanceof JSONRPCResponse) |
| 250 | + .verifyComplete(); |
| 251 | + } |
| 252 | + |
174 | 253 | @Test |
175 | 254 | void resourceSubscribeIsAdvertisedAndHonouredWhenRepositorySupportsIt() { |
176 | 255 | McpServer server = resourcesServer(true, new McpChangeFeed()); |
@@ -268,6 +347,15 @@ private static McpTransportResponse.Streaming listen(McpServer server, int id) { |
268 | 347 | return (McpTransportResponse.Streaming) server.handle(McpTransportContext.EMPTY, request).block(); |
269 | 348 | } |
270 | 349 |
|
| 350 | + private static void assertNotificationsRequired(McpTransportResponse response) { |
| 351 | + // An error response, not a stream: no acknowledgement is ever sent. |
| 352 | + assertThat(response).asInstanceOf(type(McpTransportResponse.Error.class)) |
| 353 | + .extracting(error -> error.response().error()) |
| 354 | + .returns(ErrorCodes.INVALID_PARAMS, JSONRPCError::code) |
| 355 | + .extracting(JSONRPCError::message, as(STRING)) |
| 356 | + .contains("notifications", "required"); |
| 357 | + } |
| 358 | + |
271 | 359 | private static Object subscriptionIdOf(JSONRPCNotification notification) { |
272 | 360 | return ((SubscriptionsAcknowledgedParams) notification.params()).meta().get(MetaKeys.SUBSCRIPTION_ID); |
273 | 361 | } |
|
0 commit comments