Spring/spring-graphql
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor GraphQlWebSocketHandler tests

Browse Source
This commit is contained in:
Rossen Stoyanchev
2021-05-31 21:49:06 +01:00
parent ad8242e5e1
commit 266a2ddcda
9 changed files with 301 additions and 388 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
spring-graphql/src/main/java/org/springframework/graphql/execution/SingleErrorExceptionResolver.java
View File

@@ -1,42 +0,0 @@
/*
* Copyright 2002-2021 the original author or authors.
*
* Licensed 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
*
* https://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.springframework.graphql.execution;
import java.util.Collections;
import java.util.List;
import graphql.GraphQLError;
import graphql.schema.DataFetchingEnvironment;
import reactor.core.publisher.Mono;
/**
* Simple adapter for {@link DataFetcherExceptionResolver} implementations that
* resolve exceptions to a single error only.
*/
public abstract class SingleErrorExceptionResolver implements DataFetcherExceptionResolver {
@Override
public final Mono<List<GraphQLError>> resolveException(Throwable exception, DataFetchingEnvironment environment) {
return doResolve(exception, environment).map(Collections::singletonList);
}
/**
* Implement this method to resolve the exception to an error.
*/
protected abstract Mono<GraphQLError> doResolve(Throwable exception, DataFetchingEnvironment environment);
}

27
spring-graphql/src/test/java/org/springframework/graphql/web/Book.java
View File

@@ -2,50 +2,41 @@ package org.springframework.graphql.web;
public class Book {
String id;
Long id;
String name;
int pageCount;
String author;
public Book() {
}
public Book(String id, String name, int pageCount, String author) {
public Book(Long id, String name, String author) {
this.id = id;
this.name = name;
this.pageCount = pageCount;
this.author = author;
}
public String getId() {
return id;
public Long getId() {
return this.id;
}
public void setId(String id) {
public void setId(Long id) {
this.id = id;
}
public String getName() {
return name;
return this.name;
}
public void setName(String name) {
this.name = name;
}
public int getPageCount() {
return pageCount;
}
public void setPageCount(int pageCount) {
this.pageCount = pageCount;
}
public String getAuthor() {
return author;
return this.author;
}
public void setAuthor(String author) {

91
spring-graphql/src/test/java/org/springframework/graphql/web/BookTestUtils.java Normal file
View File

@@ -0,0 +1,91 @@
/*
* Copyright 2002-2021 the original author or authors.
*
* Licensed 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
*
* https://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.springframework.graphql.web;
import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import graphql.schema.idl.RuntimeWiring;
import reactor.core.publisher.Flux;
import org.springframework.core.io.ClassPathResource;
import org.springframework.graphql.execution.ExecutionGraphQlService;
import org.springframework.graphql.execution.GraphQlSource;
import static graphql.schema.idl.TypeRuntimeWiring.newTypeWiring;
public abstract class BookTestUtils {
public static final String SUBSCRIPTION_ID = "1";
public static final String BOOK_QUERY = "{" +
"\"id\":\"" + BookTestUtils.SUBSCRIPTION_ID + "\"," +
"\"type\":\"subscribe\"," +
"\"payload\":{\"query\": \"" +
" query TestQuery {" +
" bookById(id: \\\"1\\\"){ " +
" id" +
" name" +
" author" +
" }}\"}" +
"}";
public static final String BOOK_SUBSCRIPTION = "{" +
"\"id\":\"" + SUBSCRIPTION_ID + "\"," +
"\"type\":\"subscribe\"," +
"\"payload\":{\"query\": \"" +
" subscription TestSubscription {" +
" bookSearch(author: \\\"George\\\") {" +
" id" +
" name" +
" author" +
" }}\"}" +
"}";
private static final Map<Long, Book> booksMap = new HashMap<>(4);
static {
booksMap.put(1L, new Book(1L, "Nineteen Eighty-Four", "George Orwell"));
booksMap.put(2L, new Book(2L, "The Great Gatsby", "F. Scott Fitzgerald"));
booksMap.put(3L, new Book(3L, "Catch-22", "Joseph Heller"));
booksMap.put(4L, new Book(4L, "To The Lighthouse", "Virginia Woolf"));
booksMap.put(5L, new Book(5L, "Animal Farm", "George Orwell"));
}
public static WebGraphQlHandler initWebGraphQlHandler(WebInterceptor... interceptors) {
return WebGraphQlHandler.builder(new ExecutionGraphQlService(graphQlSource()))
.interceptors(Arrays.asList(interceptors))
.build();
}
private static GraphQlSource graphQlSource() {
RuntimeWiring.Builder builder = RuntimeWiring.newRuntimeWiring();
builder.type(newTypeWiring("Query").dataFetcher("bookById", env -> {
Long id = Long.parseLong(env.getArgument("id"));
return booksMap.get(id);
}));
builder.type(newTypeWiring("Subscription").dataFetcher("bookSearch", env -> {
String author = env.getArgument("author");
return Flux.fromIterable(booksMap.values()).filter(book -> book.getAuthor().contains(author));
}));
return GraphQlSource.builder()
.schemaResource(new ClassPathResource("books/schema.graphqls"))
.runtimeWiring(builder.build())
.build();
}
}

30
spring-graphql/src/test/java/org/springframework/graphql/web/GraphQlDataFetchers.java
View File

@@ -1,30 +0,0 @@
package org.springframework.graphql.web;
import java.util.Arrays;
import java.util.List;
import graphql.schema.DataFetcher;
import reactor.core.publisher.Flux;
public class GraphQlDataFetchers {
private static List<Book> books = Arrays.asList(
new Book("book-1", "GraphQL for beginners", 100, "John GraphQL"),
new Book("book-2", "Harry Potter and the Philosopher's Stone", 223, "Joanne Rowling"),
new Book("book-3", "Moby Dick", 635, "Moby Dick"),
new Book("book-3", "Moby Dick", 635, "Moby Dick"));
public static DataFetcher getBookByIdDataFetcher() {
return env -> books.stream()
.filter(book -> book.getId().equals(env.getArgument("id")))
.findFirst()
.orElse(null);
}
public static DataFetcher getBooksOnSale() {
return env -> Flux.fromIterable(books)
.filter(book -> book.getPageCount() >= (int) env.getArgument("minPages"));
}
}

223
spring-graphql/src/test/java/org/springframework/graphql/web/webflux/GraphQlWebSocketHandlerTests.java
View File

@@ -15,7 +15,6 @@
*/
package org.springframework.graphql.web.webflux;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
@@ -24,34 +23,22 @@ import java.util.List;
import java.util.Map;
import java.util.function.BiConsumer;
import graphql.schema.idl.RuntimeWiring;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.test.StepVerifier;
import org.springframework.core.io.ClassPathResource;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.graphql.execution.ExecutionGraphQlService;
import org.springframework.graphql.execution.GraphQlSource;
import org.springframework.graphql.web.BookTestUtils;
import org.springframework.graphql.web.ConsumeOneAndNeverCompleteInterceptor;
import org.springframework.graphql.web.GraphQlDataFetchers;
import org.springframework.graphql.web.WebGraphQlHandler;
import org.springframework.graphql.web.WebInterceptor;
import org.springframework.http.HttpHeaders;
import org.springframework.http.codec.ServerCodecConfigurer;
import org.springframework.http.codec.json.Jackson2JsonDecoder;
import org.springframework.lang.Nullable;
import org.springframework.web.reactive.socket.CloseStatus;
import org.springframework.web.reactive.socket.HandshakeInfo;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.adapter.AbstractWebSocketSession;
import static graphql.schema.idl.TypeRuntimeWiring.newTypeWiring;
import static org.assertj.core.api.Assertions.as;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.InstanceOfAssertFactories.map;
@@ -63,72 +50,38 @@ public class GraphQlWebSocketHandlerTests {
private static final Jackson2JsonDecoder decoder = new Jackson2JsonDecoder();
private static final String SUBSCRIPTION_ID = "123";
private static final String BOOK_SEARCH_QUERY = "{" +
"\"id\":\"" + SUBSCRIPTION_ID + "\"," +
"\"type\":\"subscribe\"," +
"\"payload\":{\"query\": \"" +
" subscription TestSubscription {" +
" bookSearch(minPages: 200) {" +
" id" +
" name" +
" pageCount" +
" author" +
" }}\"}" +
"}";
@Test
void query() throws Exception {
String bookQuery = "{" +
"\"id\":\"" + SUBSCRIPTION_ID + "\"," +
"\"type\":\"subscribe\"," +
"\"payload\":{\"query\": \"" +
" query TestQuery {" +
" bookById(id: \\\"book-1\\\"){ " +
" id" +
" name" +
" pageCount" +
" author" +
" }}\"}" +
"}";
Flux<WebSocketMessage> input = Flux.just(
void query() {
TestWebSocketSession session = handle(Flux.just(
toWebSocketMessage("{\"type\":\"connection_init\"}"),
toWebSocketMessage(bookQuery));
TestWebSocketSession session = new TestWebSocketSession(input);
initWebSocketHandler().handle(session).block();
toWebSocketMessage(BookTestUtils.BOOK_QUERY)));
StepVerifier.create(session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.consumeNextWith(message ->
assertThat(decode(message))
.hasSize(3)
.containsEntry("id", SUBSCRIPTION_ID)
.containsEntry("id", BookTestUtils.SUBSCRIPTION_ID)
.containsEntry("type", "next")
.extractingByKey("payload", as(map(String.class, Object.class)))
.extractingByKey("data", as(map(String.class, Object.class)))
.extractingByKey("bookById", as(map(String.class, Object.class)))
.containsEntry("name", "GraphQL for beginners"))
.containsEntry("name", "Nineteen Eighty-Four"))
.consumeNextWith(message -> assertMessageType(message, "complete"))
.verifyComplete();
}
@Test
void subscription() throws Exception {
Flux<WebSocketMessage> input = Flux.just(
void subscription() {
TestWebSocketSession session = handle(Flux.just(
toWebSocketMessage("{\"type\":\"connection_init\"}"),
toWebSocketMessage(BOOK_SEARCH_QUERY));
TestWebSocketSession session = new TestWebSocketSession(input);
initWebSocketHandler().handle(session).block();
toWebSocketMessage(BookTestUtils.BOOK_SUBSCRIPTION)));
BiConsumer<WebSocketMessage, String> bookPayloadAssertion = (message, bookId) ->
assertThat(decode(message))
.hasSize(3)
.containsEntry("id", SUBSCRIPTION_ID)
.containsEntry("id", BookTestUtils.SUBSCRIPTION_ID)
.containsEntry("type", "next")
.extractingByKey("payload", as(map(String.class, Object.class)))
.extractingByKey("data", as(map(String.class, Object.class)))
@@ -137,21 +90,17 @@ public class GraphQlWebSocketHandlerTests {
StepVerifier.create(session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-2"))
.consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-3"))
.consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-3"))
.consumeNextWith(message -> bookPayloadAssertion.accept(message, "1"))
.consumeNextWith(message -> bookPayloadAssertion.accept(message, "5"))
.consumeNextWith(message -> assertMessageType(message, "complete"))
.verifyComplete();
}
@Test
void unauthorizedWithoutMessageType() throws Exception {
Flux<WebSocketMessage> input = Flux.just(
void unauthorizedWithoutMessageType() {
TestWebSocketSession session = handle(Flux.just(
toWebSocketMessage("{\"type\":\"connection_init\"}"),
toWebSocketMessage("{\"id\":\"" + SUBSCRIPTION_ID + "\"}")); // No message type
TestWebSocketSession session = new TestWebSocketSession(input);
initWebSocketHandler().handle(session).block();
toWebSocketMessage("{\"id\":\"" + BookTestUtils.SUBSCRIPTION_ID + "\"}")));
StepVerifier.create(session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
@@ -163,13 +112,12 @@ public class GraphQlWebSocketHandlerTests {
}
@Test
void invalidMessageWithoutId() throws Exception {
void invalidMessageWithoutId() {
Flux<WebSocketMessage> input = Flux.just(
toWebSocketMessage("{\"type\":\"connection_init\"}"),
toWebSocketMessage("{\"type\":\"subscribe\"}")); // No message id
TestWebSocketSession session = new TestWebSocketSession(input);
initWebSocketHandler().handle(session).block();
TestWebSocketSession session = handle(input);
StepVerifier.create(session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
@@ -181,9 +129,8 @@ public class GraphQlWebSocketHandlerTests {
}
@Test
void unauthorizedWithoutConnectionInit() throws Exception {
TestWebSocketSession session = new TestWebSocketSession(Flux.just(toWebSocketMessage(BOOK_SEARCH_QUERY)));
initWebSocketHandler().handle(session).block();
void unauthorizedWithoutConnectionInit() {
TestWebSocketSession session = handle(Flux.just(toWebSocketMessage(BookTestUtils.BOOK_SUBSCRIPTION)));
StepVerifier.create(session.getOutput()).verifyComplete();
StepVerifier.create(session.closeStatus())
@@ -192,13 +139,10 @@ public class GraphQlWebSocketHandlerTests {
}
@Test
void tooManyConnectionInitRequests() throws Exception {
Flux<WebSocketMessage> input = Flux.just(
void tooManyConnectionInitRequests() {
TestWebSocketSession session = handle(Flux.just(
toWebSocketMessage("{\"type\":\"connection_init\"}"),
toWebSocketMessage("{\"type\":\"connection_init\"}"));
TestWebSocketSession session = new TestWebSocketSession(input);
initWebSocketHandler().handle(session).block();
toWebSocketMessage("{\"type\":\"connection_init\"}")));
StepVerifier.create(session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
@@ -210,9 +154,12 @@ public class GraphQlWebSocketHandlerTests {
}
@Test
void connectionInitTimeout() throws Exception {
void connectionInitTimeout() {
GraphQlWebSocketHandler handler = new GraphQlWebSocketHandler(
BookTestUtils.initWebGraphQlHandler(), ServerCodecConfigurer.create(), Duration.ofMillis(50));
TestWebSocketSession session = new TestWebSocketSession(Flux.empty());
initWebSocketHandler(Collections.emptyList(), Duration.ofMillis(50)).handle(session).block();
handler.handle(session).block();
StepVerifier.create(session.closeStatus())
.expectNext(new CloseStatus(4408, "Connection initialisation timeout"))
@@ -220,23 +167,18 @@ public class GraphQlWebSocketHandlerTests {
}
@Test
void subscriptionExists() throws Exception {
Flux<WebSocketMessage> input = Flux.just(
void subscriptionExists() {
TestWebSocketSession session = handle(Flux.just(
toWebSocketMessage("{\"type\":\"connection_init\"}"),
toWebSocketMessage(BOOK_SEARCH_QUERY),
toWebSocketMessage(BOOK_SEARCH_QUERY));
List<WebInterceptor> interceptors = Collections.singletonList(new ConsumeOneAndNeverCompleteInterceptor());
TestWebSocketSession session = new TestWebSocketSession(input);
initWebSocketHandler(interceptors, null).handle(session).block();
toWebSocketMessage(BookTestUtils.BOOK_SUBSCRIPTION),
toWebSocketMessage(BookTestUtils.BOOK_SUBSCRIPTION)), new ConsumeOneAndNeverCompleteInterceptor());
// Collect messages until session closed
List<Map<String, Object>> messages = new ArrayList<>();
session.getOutput().subscribe(message -> messages.add(decode(message)));
StepVerifier.create(session.closeStatus())
.expectNext(new CloseStatus(4409, "Subscriber for " + SUBSCRIPTION_ID + " already exists"))
.expectNext(new CloseStatus(4409, "Subscriber for " + BookTestUtils.SUBSCRIPTION_ID + " already exists"))
.verifyComplete();
assertThat(messages.size()).isEqualTo(2);
@@ -245,55 +187,36 @@ public class GraphQlWebSocketHandlerTests {
}
@Test
void clientCompletion() throws Exception {
void clientCompletion() {
Sinks.Many<WebSocketMessage> input = Sinks.many().unicast().onBackpressureBuffer();
input.tryEmitNext(toWebSocketMessage("{\"type\":\"connection_init\"}"));
input.tryEmitNext(toWebSocketMessage(BOOK_SEARCH_QUERY));
input.tryEmitNext(toWebSocketMessage(BookTestUtils.BOOK_SUBSCRIPTION));
List<WebInterceptor> interceptors = Collections.singletonList(new ConsumeOneAndNeverCompleteInterceptor());
TestWebSocketSession session = new TestWebSocketSession(input.asFlux());
initWebSocketHandler(interceptors, null).handle(session).block();
TestWebSocketSession session =
handle(input.asFlux(), new ConsumeOneAndNeverCompleteInterceptor());
String completeMessage = "{\"id\":\"" + SUBSCRIPTION_ID + "\",\"type\":\"complete\"}";
String completeMessage = "{\"id\":\"" + BookTestUtils.SUBSCRIPTION_ID + "\",\"type\":\"complete\"}";
StepVerifier.create(session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.consumeNextWith(message -> assertMessageType(message, "next"))
.then(() -> input.tryEmitNext(toWebSocketMessage(completeMessage)))
.as("Second subscription with same id is possible only if the first was properly removed")
.then(() -> input.tryEmitNext(toWebSocketMessage(BOOK_SEARCH_QUERY)))
.then(() -> input.tryEmitNext(toWebSocketMessage(BookTestUtils.BOOK_SUBSCRIPTION)))
.consumeNextWith(message -> assertMessageType(message, "next"))
.then(() -> input.tryEmitNext(toWebSocketMessage(completeMessage)))
.verifyTimeout(Duration.ofMillis(500));
}
private GraphQlWebSocketHandler initWebSocketHandler() throws Exception {
return initWebSocketHandler(Collections.emptyList(), Duration.ofSeconds(60));
}
private GraphQlWebSocketHandler initWebSocketHandler(
@Nullable List<WebInterceptor> interceptors, @Nullable Duration initTimeoutDuration) {
WebGraphQlHandler graphQlHandler =
WebGraphQlHandler.builder(new ExecutionGraphQlService(graphQlSource()))
.interceptors(interceptors != null ? interceptors : Collections.emptyList())
.build();
return new GraphQlWebSocketHandler(graphQlHandler,
private TestWebSocketSession handle(Flux<WebSocketMessage> input, WebInterceptor... interceptors) {
GraphQlWebSocketHandler handler = new GraphQlWebSocketHandler(
BookTestUtils.initWebGraphQlHandler(interceptors),
ServerCodecConfigurer.create(),
(initTimeoutDuration != null ? initTimeoutDuration : Duration.ofSeconds(60)));
}
Duration.ofSeconds(60));
private static GraphQlSource graphQlSource() {
RuntimeWiring.Builder builder = RuntimeWiring.newRuntimeWiring();
builder.type(newTypeWiring("Query").dataFetcher("bookById", GraphQlDataFetchers.getBookByIdDataFetcher()));
builder.type(newTypeWiring("Subscription").dataFetcher("bookSearch", GraphQlDataFetchers.getBooksOnSale()));
RuntimeWiring runtimeWiring = builder.build();
return GraphQlSource.builder()
.schemaResource(new ClassPathResource("books/schema.graphqls"))
.runtimeWiring(runtimeWiring)
.build();
TestWebSocketSession session = new TestWebSocketSession(input);
handler.handle(session).block();
return session;
}
private static WebSocketMessage toWebSocketMessage(String data) {
@@ -312,61 +235,7 @@ public class GraphQlWebSocketHandlerTests {
Map<String, Object> map = decode(message);
assertThat(map).containsEntry("type", messageType);
if (!messageType.equals("connection_ack")) {
assertThat(map).containsEntry("id", SUBSCRIPTION_ID);
}
}
private static class TestWebSocketSession extends AbstractWebSocketSession<Object> {
private final Flux<WebSocketMessage> input;
private Flux<WebSocketMessage> output = Flux.empty();
private final Sinks.One<CloseStatus> closeStatusSink = Sinks.one();
public TestWebSocketSession(Flux<WebSocketMessage> input) {
this("1", URI.create("https://example.org/graphql"), input);
}
public TestWebSocketSession(String id, URI uri, Flux<WebSocketMessage> input) {
super(new Object(), id,
new HandshakeInfo(uri, new HttpHeaders(), Mono.empty(), null),
DefaultDataBufferFactory.sharedInstance);
this.input = input;
}
@Override
public Flux<WebSocketMessage> receive() {
return this.input;
}
@Override
public Mono<Void> send(Publisher<WebSocketMessage> messages) {
this.output = Flux.from(messages);
return Mono.empty();
}
public Flux<WebSocketMessage> getOutput() {
return this.output;
}
@Override
public boolean isOpen() {
throw new java.lang.UnsupportedOperationException();
}
@Override
public Mono<Void> close(CloseStatus status) {
this.closeStatusSink.tryEmitValue(status);
return Mono.empty();
}
@Override
public Mono<CloseStatus> closeStatus() {
return this.closeStatusSink.asMono();
assertThat(map).containsEntry("id", BookTestUtils.SUBSCRIPTION_ID);
}
}

87
spring-graphql/src/test/java/org/springframework/graphql/web/webflux/TestWebSocketSession.java Normal file
View File

@@ -0,0 +1,87 @@
/*
* Copyright 2002-2021 the original author or authors.
*
* Licensed 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
*
* https://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.springframework.graphql.web.webflux;
import java.net.URI;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.web.reactive.socket.CloseStatus;
import org.springframework.web.reactive.socket.HandshakeInfo;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.adapter.AbstractWebSocketSession;
/**
* {@link org.springframework.web.reactive.socket.WebSocketSession} that is given
* an input Flux of messages and exposes a Flux of published output messages.
*/
class TestWebSocketSession extends AbstractWebSocketSession<Object> {
private final Flux<WebSocketMessage> input;
private Flux<WebSocketMessage> output = Flux.empty();
private final Sinks.One<CloseStatus> closeStatusSink = Sinks.one();
public TestWebSocketSession(Flux<WebSocketMessage> input) {
this("1", URI.create("https://example.org/graphql"), input);
}
public TestWebSocketSession(String id, URI uri, Flux<WebSocketMessage> input) {
super(new Object(), id,
new HandshakeInfo(uri, new HttpHeaders(), Mono.empty(), null),
DefaultDataBufferFactory.sharedInstance);
this.input = input;
}
@Override
public Flux<WebSocketMessage> receive() {
return this.input;
}
@Override
public Mono<Void> send(Publisher<WebSocketMessage> messages) {
this.output = Flux.from(messages);
return Mono.empty();
}
public Flux<WebSocketMessage> getOutput() {
return this.output;
}
@Override
public boolean isOpen() {
throw new UnsupportedOperationException();
}
@Override
public Mono<Void> close(CloseStatus status) {
this.closeStatusSink.tryEmitValue(status);
return Mono.empty();
}
@Override
public Mono<CloseStatus> closeStatus() {
return this.closeStatusSink.asMono();
}
}

183
spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/GraphQlWebSocketHandlerTests.java
View File

@@ -20,33 +20,25 @@ import java.io.IOException;
import java.io.InputStream;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import graphql.schema.idl.RuntimeWiring;
import org.junit.jupiter.api.Test;
import reactor.test.StepVerifier;
import org.springframework.core.io.ClassPathResource;
import org.springframework.graphql.execution.ExecutionGraphQlService;
import org.springframework.graphql.execution.GraphQlSource;
import org.springframework.graphql.web.BookTestUtils;
import org.springframework.graphql.web.ConsumeOneAndNeverCompleteInterceptor;
import org.springframework.graphql.web.GraphQlDataFetchers;
import org.springframework.graphql.web.WebGraphQlHandler;
import org.springframework.graphql.web.WebInterceptor;
import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpInputMessage;
import org.springframework.http.converter.HttpMessageConverter;
import org.springframework.http.converter.json.MappingJackson2HttpMessageConverter;
import org.springframework.lang.Nullable;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketMessage;
import static graphql.schema.idl.TypeRuntimeWiring.newTypeWiring;
import static org.assertj.core.api.Assertions.as;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.InstanceOfAssertFactories.map;
@@ -56,166 +48,134 @@ import static org.assertj.core.api.InstanceOfAssertFactories.map;
*/
public class GraphQlWebSocketHandlerTests {
private static final String SUBSCRIPTION_ID = "123";
private static final String BOOK_SEARCH_QUERY = "{" +
"\"id\":\"" + SUBSCRIPTION_ID + "\"," +
"\"type\":\"subscribe\"," +
"\"payload\":{\"query\": \"" +
" subscription TestSubscription {" +
" bookSearch(minPages: 200) {" +
" id" +
" name" +
" pageCount" +
" author" +
" }}\"}" +
"}";
private static final HttpMessageConverter<?> converter = new MappingJackson2HttpMessageConverter();
private final TestWebSocketSession session = new TestWebSocketSession();
private final GraphQlWebSocketHandler handler =
initWebSocketHandler(Collections.emptyList(), Duration.ofSeconds(60));
private final GraphQlWebSocketHandler handler = initWebSocketHandler();
@Test
void query() throws Exception {
String bookQuery = "{" +
"\"id\":\"" + SUBSCRIPTION_ID + "\"," +
"\"type\":\"subscribe\"," +
"\"payload\":{\"query\": \"" +
" query TestQuery {" +
" bookById(id: \\\"book-1\\\"){ " +
" id" +
" name" +
" pageCount" +
" author" +
" }}\"}" +
"}";
handle(this.handler,
new TextMessage("{\"type\":\"connection_init\"}"),
new TextMessage(BookTestUtils.BOOK_QUERY));
this.handler.afterConnectionEstablished(session);
this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}"));
this.handler.handleTextMessage(session, new TextMessage(bookQuery));
StepVerifier.create(session.getOutput())
StepVerifier.create(this.session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.consumeNextWith(message ->
assertThat(decode(message))
.hasSize(3)
.containsEntry("id", SUBSCRIPTION_ID)
.containsEntry("id", BookTestUtils.SUBSCRIPTION_ID)
.containsEntry("type", "next")
.extractingByKey("payload", as(map(String.class, Object.class)))
.extractingByKey("data", as(map(String.class, Object.class)))
.extractingByKey("bookById", as(map(String.class, Object.class)))
.containsEntry("name", "GraphQL for beginners"))
.containsEntry("name", "Nineteen Eighty-Four"))
.consumeNextWith(message -> assertMessageType(message, "complete"))
.then(session::close) // Complete output Flux
.then(this.session::close) // Complete output Flux
.verifyComplete();
}
@Test
void subscription() throws Exception {
this.handler.afterConnectionEstablished(session);
this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}"));
this.handler.handleTextMessage(session, new TextMessage(BOOK_SEARCH_QUERY));
handle(this.handler,
new TextMessage("{\"type\":\"connection_init\"}"),
new TextMessage(BookTestUtils.BOOK_SUBSCRIPTION));
BiConsumer<WebSocketMessage<?>, String> bookPayloadAssertion = (message, bookId) ->
assertThat(decode(message))
.hasSize(3)
.containsEntry("id", SUBSCRIPTION_ID)
.containsEntry("id", BookTestUtils.SUBSCRIPTION_ID)
.containsEntry("type", "next")
.extractingByKey("payload", as(map(String.class, Object.class)))
.extractingByKey("data", as(map(String.class, Object.class)))
.extractingByKey("bookSearch", as(map(String.class, Object.class)))
.containsEntry("id", bookId);
StepVerifier.create(session.getOutput())
StepVerifier.create(this.session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-2"))
.consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-3"))
.consumeNextWith(message -> bookPayloadAssertion.accept(message, "book-3"))
.consumeNextWith(message -> bookPayloadAssertion.accept(message, "1"))
.consumeNextWith(message -> bookPayloadAssertion.accept(message, "5"))
.consumeNextWith(message -> assertMessageType(message, "complete"))
.then(session::close) // Complete output Flux
.then(this.session::close) // Complete output Flux
.verifyComplete();
}
@Test
void unauthorizedWithoutMessageType() throws Exception {
this.handler.afterConnectionEstablished(session);
this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}"));
this.handler.handleTextMessage(session, new TextMessage("{\"id\":\"" + SUBSCRIPTION_ID + "\"}")); // No message type
handle(this.handler,
new TextMessage("{\"type\":\"connection_init\"}"),
new TextMessage("{\"id\":\"" + BookTestUtils.SUBSCRIPTION_ID + "\"}")); // No message type
StepVerifier.create(session.getOutput())
StepVerifier.create(this.session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.verifyComplete();
assertThat(session.getCloseStatus()).isEqualTo(new CloseStatus(4400, "Invalid message"));
assertThat(this.session.getCloseStatus()).isEqualTo(new CloseStatus(4400, "Invalid message"));
}
@Test
void invalidMessageWithoutId() throws Exception {
this.handler.afterConnectionEstablished(session);
this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}"));
this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"subscribe\"}")); // No message id
handle(this.handler,
new TextMessage("{\"type\":\"connection_init\"}"),
new TextMessage("{\"type\":\"subscribe\"}")); // No message id
StepVerifier.create(session.getOutput())
StepVerifier.create(this.session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.verifyComplete();
assertThat(session.getCloseStatus()).isEqualTo(new CloseStatus(4400, "Invalid message"));
assertThat(this.session.getCloseStatus()).isEqualTo(new CloseStatus(4400, "Invalid message"));
}
@Test
void unauthorizedWithoutConnectionInit() throws Exception {
this.handler.afterConnectionEstablished(session);
this.handler.handleTextMessage(session, new TextMessage(BOOK_SEARCH_QUERY));
handle(this.handler, new TextMessage(BookTestUtils.BOOK_SUBSCRIPTION));
StepVerifier.create(session.getOutput()).verifyComplete();
assertThat(session.getCloseStatus()).isEqualTo(new CloseStatus(4401, "Unauthorized"));
StepVerifier.create(this.session.getOutput()).verifyComplete();
assertThat(this.session.getCloseStatus()).isEqualTo(new CloseStatus(4401, "Unauthorized"));
}
@Test
void tooManyConnectionInitRequests() throws Exception {
this.handler.afterConnectionEstablished(session);
this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}"));
this.handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}"));
handle(this.handler,
new TextMessage("{\"type\":\"connection_init\"}"),
new TextMessage("{\"type\":\"connection_init\"}"));
StepVerifier.create(session.getOutput())
StepVerifier.create(this.session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.verifyComplete();
assertThat(session.getCloseStatus())
assertThat(this.session.getCloseStatus())
.isEqualTo(new CloseStatus(4429, "Too many initialisation requests"));
}
@Test
void connectionInitTimeout() {
GraphQlWebSocketHandler handler = initWebSocketHandler(Collections.emptyList(), Duration.ofMillis(50));
handler.afterConnectionEstablished(session);
GraphQlWebSocketHandler handler = new GraphQlWebSocketHandler(
BookTestUtils.initWebGraphQlHandler(), converter, Duration.ofMillis(50));
StepVerifier.create(session.closeStatus())
handler.afterConnectionEstablished(this.session);
StepVerifier.create(this.session.closeStatus())
.expectNext(new CloseStatus(4408, "Connection initialisation timeout"))
.verifyComplete();
}
@Test
void subscriptionExists() throws Exception {
GraphQlWebSocketHandler handler = initWebSocketHandler(
Collections.singletonList(new ConsumeOneAndNeverCompleteInterceptor()), null);
handler.afterConnectionEstablished(session);
handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}"));
handler.handleTextMessage(session, new TextMessage(BOOK_SEARCH_QUERY));
handler.handleTextMessage(session, new TextMessage(BOOK_SEARCH_QUERY));
handle(initWebSocketHandler(new ConsumeOneAndNeverCompleteInterceptor()),
new TextMessage("{\"type\":\"connection_init\"}"),
new TextMessage(BookTestUtils.BOOK_SUBSCRIPTION),
new TextMessage(BookTestUtils.BOOK_SUBSCRIPTION));
// Collect messages until session closed
List<Map<String, Object>> messages = new ArrayList<>();
session.getOutput().subscribe(message -> messages.add(decode(message)));
StepVerifier.create(session.closeStatus())
.expectNext(new CloseStatus(4409, "Subscriber for " + SUBSCRIPTION_ID + " already exists"))
StepVerifier.create(this.session.closeStatus())
.expectNext(new CloseStatus(4409, "Subscriber for " + BookTestUtils.SUBSCRIPTION_ID + " already exists"))
.verifyComplete();
assertThat(messages.size()).isEqualTo(2);
@@ -225,69 +185,56 @@ public class GraphQlWebSocketHandlerTests {
@Test
void clientCompletion() throws Exception {
GraphQlWebSocketHandler handler = initWebSocketHandler(
Collections.singletonList(new ConsumeOneAndNeverCompleteInterceptor()), null);
GraphQlWebSocketHandler handler =
initWebSocketHandler(new ConsumeOneAndNeverCompleteInterceptor());
handler.afterConnectionEstablished(session);
handler.handleTextMessage(session, new TextMessage("{\"type\":\"connection_init\"}"));
handler.handleTextMessage(session, new TextMessage(BOOK_SEARCH_QUERY));
handle(handler,
new TextMessage("{\"type\":\"connection_init\"}"),
new TextMessage(BookTestUtils.BOOK_SUBSCRIPTION));
String completeMessage = "{\"id\":\"" + SUBSCRIPTION_ID + "\",\"type\":\"complete\"}";
String completeMessage = "{\"id\":\"" + BookTestUtils.SUBSCRIPTION_ID + "\",\"type\":\"complete\"}";
Consumer<String> messageSender = body -> {
try {
handler.handleTextMessage(session, new TextMessage(body));
handler.handleTextMessage(this.session, new TextMessage(body));
}
catch (Exception ex) {
throw new IllegalStateException(ex);
}
};
StepVerifier.create(session.getOutput())
StepVerifier.create(this.session.getOutput())
.consumeNextWith(message -> assertMessageType(message, "connection_ack"))
.consumeNextWith(message -> assertMessageType(message, "next"))
.then(() -> messageSender.accept(completeMessage))
.as("Second subscription with same id is possible only if the first was properly removed")
.then(() -> messageSender.accept(BOOK_SEARCH_QUERY))
.then(() -> messageSender.accept(BookTestUtils.BOOK_SUBSCRIPTION))
.consumeNextWith(message -> assertMessageType(message, "next"))
.then(() -> messageSender.accept(completeMessage))
.verifyTimeout(Duration.ofMillis(500));
}
private void handle(GraphQlWebSocketHandler handler, TextMessage... textMessages) throws Exception {
handler.afterConnectionEstablished(this.session);
for (TextMessage message : textMessages) {
handler.handleTextMessage(this.session, message);
}
}
private GraphQlWebSocketHandler initWebSocketHandler(
@Nullable List<WebInterceptor> interceptors, @Nullable Duration initTimeoutDuration) {
private GraphQlWebSocketHandler initWebSocketHandler(WebInterceptor... interceptors) {
try {
WebGraphQlHandler graphQlHandler =
WebGraphQlHandler.builder(new ExecutionGraphQlService(graphQlSource()))
.interceptors(interceptors != null ? interceptors : Collections.emptyList())
.build();
return new GraphQlWebSocketHandler(graphQlHandler, converter,
(initTimeoutDuration != null ? initTimeoutDuration : Duration.ofSeconds(60)));
return new GraphQlWebSocketHandler(
BookTestUtils.initWebGraphQlHandler(interceptors), converter, Duration.ofSeconds(60));
}
catch (Exception ex) {
throw new IllegalStateException(ex);
}
}
private static GraphQlSource graphQlSource() {
RuntimeWiring.Builder builder = RuntimeWiring.newRuntimeWiring();
builder.type(newTypeWiring("Query").dataFetcher("bookById", GraphQlDataFetchers.getBookByIdDataFetcher()));
builder.type(newTypeWiring("Subscription").dataFetcher("bookSearch", GraphQlDataFetchers.getBooksOnSale()));
RuntimeWiring runtimeWiring = builder.build();
return GraphQlSource.builder()
.schemaResource(new ClassPathResource("books/schema.graphqls"))
.runtimeWiring(runtimeWiring)
.build();
}
private void assertMessageType(WebSocketMessage<?> message, String messageType) {
Map<String, Object> map = decode(message, Map.class);
assertThat(map).containsEntry("type", messageType);
if (!messageType.equals("connection_ack")) {
assertThat(map).containsEntry("id", SUBSCRIPTION_ID);
assertThat(map).containsEntry("id", BookTestUtils.SUBSCRIPTION_ID);
}
}

3
spring-graphql/src/test/java/org/springframework/graphql/web/webmvc/TestWebSocketSession.java
View File

@@ -35,7 +35,8 @@ import org.springframework.web.socket.WebSocketMessage;
import org.springframework.web.socket.WebSocketSession;
/**
*
* WebSocketSession that saves sent messages and exposes them as a Flux which
* makes assertions comparable to the same for WebFlux.
*/
public class TestWebSocketSession implements WebSocketSession {

3
spring-graphql/src/test/resources/books/schema.graphqls
View File

@@ -5,10 +5,9 @@ type Query {
type Book {
id: ID
name: String
pageCount: Int
author: String
}
type Subscription {
bookSearch(minPages:Int) : Book!
bookSearch(author: String) : Book!
}