You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ta...@apache.org on 2023/11/24 06:52:30 UTC
(flink) 01/02: [FLINK-31385] Introduce extended Assertj matchers for completable futures, that don't rely on timeouts.
This is an automated email from the ASF dual-hosted git repository.
tangyun pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 2a83c910eee711f8b5f9dd4697de60221f21fb9d
Author: David Moravek <dm...@apache.org>
AuthorDate: Mon Feb 27 18:46:58 2023 +0100
[FLINK-31385] Introduce extended Assertj matchers for completable futures, that don't rely on timeouts.
Signed-off-by: David Moravek <dm...@apache.org>
---
.../flink/core/testutils/FlinkAssertions.java | 27 ++++
.../testutils/FlinkCompletableFutureAssert.java | 136 +++++++++++++++++++++
.../flink/core/testutils/FlinkAssertionsTest.java | 121 ++++++++++++++++++
3 files changed, 284 insertions(+)
diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkAssertions.java b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkAssertions.java
index c2f51b0295a..ea11627cdf1 100644
--- a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkAssertions.java
+++ b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkAssertions.java
@@ -25,6 +25,8 @@ import org.assertj.core.api.InstanceOfAssertFactory;
import org.assertj.core.api.ListAssert;
import org.assertj.core.api.ThrowingConsumer;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
import java.util.function.Function;
import java.util.stream.Stream;
@@ -139,4 +141,29 @@ public final class FlinkAssertions {
}
return Stream.concat(Stream.of(throwable), chainOfCauses(throwable.getCause()));
}
+
+ /**
+ * Create assertion for {@link java.util.concurrent.CompletableFuture}.
+ *
+ * @param actual the actual value.
+ * @param <T> the type of the value contained in the {@link
+ * java.util.concurrent.CompletableFuture}.
+ * @return the created assertion object.
+ */
+ public static <T> FlinkCompletableFutureAssert<T> assertThatFuture(
+ CompletableFuture<T> actual) {
+ return new FlinkCompletableFutureAssert<>(actual);
+ }
+
+ /**
+ * Create assertion for {@link java.util.concurrent.CompletionStage}.
+ *
+ * @param actual the actual value.
+ * @param <T> the type of the value contained in the {@link
+ * java.util.concurrent.CompletionStage}.
+ * @return the created assertion object.
+ */
+ public static <T> FlinkCompletableFutureAssert<T> assertThatFuture(CompletionStage<T> actual) {
+ return new FlinkCompletableFutureAssert<>(actual);
+ }
}
diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkCompletableFutureAssert.java b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkCompletableFutureAssert.java
new file mode 100644
index 00000000000..c0833daa9ea
--- /dev/null
+++ b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/core/testutils/FlinkCompletableFutureAssert.java
@@ -0,0 +1,136 @@
+/*
+ * 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.flink.core.testutils;
+
+import org.assertj.core.api.AbstractCompletableFutureAssert;
+import org.assertj.core.api.AssertionInfo;
+import org.assertj.core.api.ObjectAssert;
+import org.assertj.core.api.ThrowableAssertAlternative;
+import org.assertj.core.error.BasicErrorMessageFactory;
+import org.assertj.core.internal.Failures;
+import org.assertj.core.internal.Objects;
+
+import java.time.Duration;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+
+/**
+ * Enhanced version of {@link org.assertj.core.api.CompletableFutureAssert}, that allows asserting
+ * futures without relying on timeouts.
+ *
+ * @param <T> type of the value contained in the {@link CompletableFuture}.
+ */
+public class FlinkCompletableFutureAssert<T>
+ extends AbstractCompletableFutureAssert<FlinkCompletableFutureAssert<T>, T> {
+
+ private static final String SHOULD_HAVE_SUCCEEDED = "%nExpecting%n <%s>%nto have succeeded.%n";
+
+ private static final String SHOULD_HAVE_FAILED = "%nExpecting%n <%s>%nto have failed.%n";
+
+ /** A strongly typed alternative to {@link org.assertj.core.api.WithThrowable}. */
+ public static class WithThrowable {
+
+ private final Throwable throwable;
+
+ private WithThrowable(Throwable throwable) {
+ this.throwable = throwable;
+ }
+
+ /**
+ * Checks that the underlying throwable is of the given type and returns a {@link
+ * ThrowableAssertAlternative} to chain further assertions on the underlying throwable.
+ *
+ * @param type the expected {@link Throwable} type
+ * @param <T> the expected {@link Throwable} type
+ * @return a {@link ThrowableAssertAlternative} built with underlying throwable.
+ */
+ public <T extends Throwable> ThrowableAssertAlternative<T> withThrowableOfType(
+ Class<T> type) {
+ final ThrowableAssertAlternative<Throwable> throwableAssert =
+ new ThrowableAssertAlternative<>(throwable).isInstanceOf(type);
+ @SuppressWarnings("unchecked")
+ final ThrowableAssertAlternative<T> cast =
+ (ThrowableAssertAlternative<T>) throwableAssert;
+ return cast;
+ }
+ }
+
+ FlinkCompletableFutureAssert(CompletableFuture<T> actual) {
+ super(actual, FlinkCompletableFutureAssert.class);
+ }
+
+ FlinkCompletableFutureAssert(CompletionStage<T> actual) {
+ super(actual.toCompletableFuture(), FlinkCompletableFutureAssert.class);
+ }
+
+ /**
+ * An equivalent of {@link #succeedsWithin(Duration)}, that doesn't rely on timeouts.
+ *
+ * @return a new assertion object on the future's result
+ */
+ public ObjectAssert<T> eventuallySucceeds() {
+ final T object = assertEventuallySucceeds(info, actual);
+ return new ObjectAssert<>(object);
+ }
+
+ /**
+ * An equivalent of {@link #failsWithin(Duration)}, that doesn't rely on timeouts.
+ *
+ * @return a new assertion instance on the future's exception.
+ */
+ public WithThrowable eventuallyFails() {
+ return new WithThrowable(assertEventuallyFails(info, actual));
+ }
+
+ /**
+ * An equivalent of {@link #failsWithin(Duration)}, that doesn't rely on timeouts.
+ *
+ * @param exceptionClass type of the exception we expect the future to complete with
+ * @return a new assertion instance on the future's exception.
+ * @param <E> type of the exception we expect the future to complete with
+ */
+ public <E extends Throwable> ThrowableAssertAlternative<E> eventuallyFailsWith(
+ Class<E> exceptionClass) {
+ return eventuallyFails().withThrowableOfType(exceptionClass);
+ }
+
+ private T assertEventuallySucceeds(AssertionInfo info, Future<T> actual) {
+ Objects.instance().assertNotNull(info, actual);
+ try {
+ return actual.get();
+ } catch (InterruptedException | ExecutionException | CancellationException e) {
+ throw Failures.instance()
+ .failure(info, new BasicErrorMessageFactory(SHOULD_HAVE_SUCCEEDED, actual));
+ }
+ }
+
+ private Exception assertEventuallyFails(AssertionInfo info, Future<?> actual) {
+ Objects.instance().assertNotNull(info, actual);
+ try {
+ actual.get();
+ throw Failures.instance()
+ .failure(info, new BasicErrorMessageFactory(SHOULD_HAVE_FAILED, actual));
+ } catch (InterruptedException | ExecutionException | CancellationException e) {
+ return e;
+ }
+ }
+}
diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/test/java/org/apache/flink/core/testutils/FlinkAssertionsTest.java b/flink-test-utils-parent/flink-test-utils-junit/src/test/java/org/apache/flink/core/testutils/FlinkAssertionsTest.java
new file mode 100644
index 00000000000..97ae1a72511
--- /dev/null
+++ b/flink-test-utils-parent/flink-test-utils-junit/src/test/java/org/apache/flink/core/testutils/FlinkAssertionsTest.java
@@ -0,0 +1,121 @@
+/*
+ * 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.flink.core.testutils;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import static org.apache.flink.core.testutils.FlinkAssertions.assertThatFuture;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Test suite for {@link FlinkAssertions}. */
+class FlinkAssertionsTest {
+
+ @Test
+ void testEventuallySucceedsWithCompletedFuture() {
+ final CompletableFuture<String> completedFuture =
+ CompletableFuture.completedFuture("Apache Flink");
+ assertThatFuture(completedFuture).eventuallySucceeds().isEqualTo("Apache Flink");
+ }
+
+ @Test
+ void testEventuallySucceedsWithFailedFuture() {
+ final CompletableFuture<String> failedFuture = new CompletableFuture<>();
+ failedFuture.completeExceptionally(new IllegalStateException("Squirrel"));
+ assertThatThrownBy(() -> assertThatFuture(failedFuture).eventuallySucceeds())
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("Squirrel");
+ }
+
+ @Test
+ void testEventuallySucceedsWithIncompleteFutureTimesOut() throws Exception {
+ assertWaitingForInterrupt(
+ () ->
+ assertThatThrownBy(
+ () ->
+ assertThatFuture(new CompletableFuture<>())
+ .eventuallySucceeds())
+ .isInstanceOf(AssertionError.class),
+ Duration.ofMillis(10));
+ }
+
+ @Test
+ void testEventuallyFailsWithCompletedFuture() {
+ final CompletableFuture<String> completedFuture =
+ CompletableFuture.completedFuture("Apache Flink");
+ assertThatThrownBy(
+ () ->
+ assertThatFuture(completedFuture)
+ .eventuallyFailsWith(IllegalStateException.class))
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("Apache Flink");
+ }
+
+ @Test
+ void testEventuallyFailsWithFailedFuture() {
+ final CompletableFuture<String> failedFuture = new CompletableFuture<>();
+ failedFuture.completeExceptionally(new IllegalStateException());
+ assertThatFuture(failedFuture)
+ .eventuallyFailsWith(ExecutionException.class)
+ .withCauseInstanceOf(IllegalStateException.class);
+ }
+
+ @Test
+ void testEventuallyFailsWithFailedFutureWithDifferentException() {
+ final CompletableFuture<String> failedFuture = new CompletableFuture<>();
+ failedFuture.completeExceptionally(new IllegalStateException());
+ assertThatThrownBy(
+ () ->
+ assertThatFuture(failedFuture)
+ .eventuallyFailsWith(CancellationException.class))
+ .isInstanceOf(AssertionError.class)
+ .hasMessageContaining("CancellationException");
+ }
+
+ @Test
+ void testEventuallyFailsWithIncompleteFutureTimesOut() throws Exception {
+ assertWaitingForInterrupt(
+ () ->
+ assertThatFuture(new CompletableFuture<>())
+ .eventuallyFails()
+ .withThrowableOfType(InterruptedException.class),
+ Duration.ofMillis(10));
+ }
+
+ private static void assertWaitingForInterrupt(Runnable runnable, Duration timeout)
+ throws Exception {
+ final CheckedThread thread =
+ new CheckedThread() {
+ @Override
+ public void go() {
+ runnable.run();
+ }
+ };
+ thread.start();
+ Thread.sleep(timeout.toMillis());
+ assertThat(thread.isAlive()).isTrue();
+ thread.interrupt();
+ thread.sync();
+ }
+}