diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java index bd9fa94d09..d0ff48e767 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/reactivestreams/ReactiveStreamsTests.java @@ -31,8 +31,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.stream.Collectors; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.Test; import org.reactivestreams.Publisher; import org.springframework.beans.factory.annotation.Autowired; @@ -51,7 +50,7 @@ import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; -import org.springframework.test.context.junit4.SpringRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import reactor.core.publisher.Flux; @@ -62,7 +61,7 @@ import reactor.core.publisher.Flux; * * @since 5.0 */ -@RunWith(SpringRunner.class) +@SpringJUnitConfig @DirtiesContext public class ReactiveStreamsTests { @@ -98,7 +97,7 @@ public class ReactiveStreamsTests { private Publisher> fixedSubscriberChannelFlow; @Test - public void testReactiveFlow() throws Exception { + void testReactiveFlow() throws Exception { List results = new ArrayList<>(); CountDownLatch latch = new CountDownLatch(6); Flux.from(this.publisher) @@ -115,7 +114,7 @@ public class ReactiveStreamsTests { } @Test - public void testPollableReactiveFlow() throws Exception { + void testPollableReactiveFlow() throws Exception { this.inputChannel.send(new GenericMessage<>("1,2,3,4,5")); CountDownLatch latch = new CountDownLatch(6); @@ -152,7 +151,7 @@ public class ReactiveStreamsTests { } @Test - public void testFromPublisher() { + void testFromPublisher() { Flux> messageFlux = Flux.just("1,2,3,4") .map(v -> v.split(",")) .flatMapIterable(Arrays::asList) @@ -178,11 +177,11 @@ public class ReactiveStreamsTests { } @Test - public void testFluxTransform() { + void testFluxTransform() { QueueChannel resultChannel = new QueueChannel(); IntegrationFlow integrationFlow = f -> f - .split() + .split((splitter) -> splitter.delimiters(",")) .fluxTransform(flux -> flux .map(Message::getPayload) .map(String::toUpperCase)) @@ -212,7 +211,7 @@ public class ReactiveStreamsTests { } @Test - public void singleChannelFlowTest() throws InterruptedException { + void singleChannelFlowTest() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); Flux.from(this.singleChannelFlow) .map(m -> m.getPayload().toUpperCase()) @@ -224,7 +223,7 @@ public class ReactiveStreamsTests { } @Test - public void fixedSubscriberChannelFlowTest() throws InterruptedException { + void fixedSubscriberChannelFlowTest() throws InterruptedException { CountDownLatch latch = new CountDownLatch(1); Flux.from(this.fixedSubscriberChannelFlow) .map(m -> m.getPayload().toUpperCase()) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java index 6e5040c669..452cc9dfaa 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/splitter/DefaultSplitterTests.java @@ -18,13 +18,13 @@ package org.springframework.integration.splitter; import static org.assertj.core.api.Assertions.assertThat; +import java.time.Duration; import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.stream.Stream; -import org.junit.Test; -import org.reactivestreams.Subscriber; +import org.junit.jupiter.api.Test; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.channel.DirectChannel; @@ -32,7 +32,6 @@ import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.endpoint.EventDrivenConsumer; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; @@ -45,10 +44,10 @@ import reactor.test.StepVerifier; * @author Gunnar Hillert * @author Artem Bilan */ -public class DefaultSplitterTests { +class DefaultSplitterTests { @Test - public void splitMessageWithArrayPayload() throws Exception { + void splitMessageWithArrayPayload() { String[] payload = new String[] { "x", "y", "z" }; Message message = MessageBuilder.withPayload(payload).build(); QueueChannel replyChannel = new QueueChannel(); @@ -69,7 +68,7 @@ public class DefaultSplitterTests { } @Test - public void splitMessageWithCollectionPayload() throws Exception { + void splitMessageWithCollectionPayload() { List payload = Arrays.asList("x", "y", "z"); Message> message = MessageBuilder.withPayload(payload).build(); QueueChannel replyChannel = new QueueChannel(); @@ -90,7 +89,7 @@ public class DefaultSplitterTests { } @Test - public void correlationIdCopiedFromMessageId() { + void correlationIdCopiedFromMessageId() { Message message = MessageBuilder.withPayload("test").build(); DirectChannel inputChannel = new DirectChannel(); QueueChannel outputChannel = new QueueChannel(1); @@ -105,7 +104,7 @@ public class DefaultSplitterTests { } @Test - public void splitMessageWithEmptyCollectionPayload() throws Exception { + void splitMessageWithEmptyCollectionPayload() { Message> message = MessageBuilder.withPayload(Collections.emptyList()).build(); QueueChannel replyChannel = new QueueChannel(); DefaultMessageSplitter splitter = new DefaultMessageSplitter(); @@ -116,7 +115,7 @@ public class DefaultSplitterTests { } @Test - public void splitStream() { + void splitStream() { Message message = new GenericMessage<>( Stream.generate(Math::random) .limit(10)); @@ -133,7 +132,7 @@ public class DefaultSplitterTests { } @Test - public void splitFlux() { + void splitFlux() { Message message = new GenericMessage<>( Flux .generate(() -> 0, @@ -159,7 +158,7 @@ public class DefaultSplitterTests { } @Test - public void splitArrayPayloadReactive() { + void splitArrayPayloadReactive() { Message message = new GenericMessage<>(new String[] { "x", "y", "z" }); FluxMessageChannel replyChannel = new FluxMessageChannel(); DefaultMessageSplitter splitter = new DefaultMessageSplitter(); @@ -174,14 +173,13 @@ public class DefaultSplitterTests { StepVerifier.create(testFlux) .expectNext("x", "y", "z") - .then(() -> - ((Subscriber) TestUtils.getPropertyValue(replyChannel, "subscribers", List.class).get(0)) - .onComplete()) - .verifyComplete(); + .expectNoEvent(Duration.ofMillis(100)) + .thenCancel() + .verify(Duration.ofSeconds(1)); } @Test - public void splitStreamReactive() { + void splitStreamReactive() { Message message = new GenericMessage<>(Stream.of("x", "y", "z")); FluxMessageChannel replyChannel = new FluxMessageChannel(); DefaultMessageSplitter splitter = new DefaultMessageSplitter(); @@ -196,14 +194,13 @@ public class DefaultSplitterTests { StepVerifier.create(testFlux) .expectNext("x", "y", "z") - .then(() -> - ((Subscriber) TestUtils.getPropertyValue(replyChannel, "subscribers", List.class).get(0)) - .onComplete()) - .verifyComplete(); + .expectNoEvent(Duration.ofMillis(100)) + .thenCancel() + .verify(Duration.ofSeconds(1)); } @Test - public void splitFluxReactive() { + void splitFluxReactive() { Message message = new GenericMessage<>(Flux.just("x", "y", "z")); FluxMessageChannel replyChannel = new FluxMessageChannel(); DefaultMessageSplitter splitter = new DefaultMessageSplitter(); @@ -218,10 +215,9 @@ public class DefaultSplitterTests { StepVerifier.create(testFlux) .expectNext("x", "y", "z") - .then(() -> - ((Subscriber) TestUtils.getPropertyValue(replyChannel, "subscribers", List.class).get(0)) - .onComplete()) - .verifyComplete(); + .expectNoEvent(Duration.ofMillis(100)) + .thenCancel() + .verify(Duration.ofSeconds(1)); } } diff --git a/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java b/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java index 66e47c5781..7690448c96 100644 --- a/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java +++ b/spring-integration-file/src/test/java/org/springframework/integration/file/splitter/FileSplitterTests.java @@ -17,7 +17,7 @@ package org.springframework.integration.file.splitter; import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.fail; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; import java.io.ByteArrayInputStream; import java.io.File; @@ -29,15 +29,13 @@ import java.io.InputStream; import java.io.Reader; import java.nio.charset.Charset; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.Date; -import java.util.List; -import org.junit.AfterClass; -import org.junit.BeforeClass; -import org.junit.Test; -import org.junit.runner.RunWith; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; import org.mockito.Mockito; -import org.reactivestreams.Subscriber; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; @@ -53,15 +51,15 @@ import org.springframework.integration.file.FileHeaders; import org.springframework.integration.file.splitter.FileSplitter.FileMarker; import org.springframework.integration.support.json.JsonObjectMapper; import org.springframework.integration.support.json.JsonObjectMapperProvider; -import org.springframework.integration.test.util.TestUtils; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.MessagingException; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; -import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import org.springframework.test.context.support.AnnotationConfigContextLoader; import org.springframework.util.FileCopyUtils; @@ -75,7 +73,7 @@ import reactor.test.StepVerifier; * @since 4.1.2 */ @ContextConfiguration(loader = AnnotationConfigContextLoader.class) -@RunWith(SpringJUnit4ClassRunner.class) +@SpringJUnitConfig @DirtiesContext public class FileSplitterTests { @@ -95,20 +93,15 @@ public class FileSplitterTests { @Autowired private PollableChannel output; - @BeforeClass - public static void setup() throws IOException { - file = File.createTempFile("foo", ".txt"); + @BeforeAll + static void setup(@TempDir File tempDir) throws IOException { + file = new File(tempDir, "foo.txt"); FileCopyUtils.copy(SAMPLE_CONTENT.getBytes(StandardCharsets.UTF_8), new FileOutputStream(file, false)); } - @AfterClass - public static void tearDown() { - file.delete(); - } - @Test - public void testFileSplitter() throws Exception { + void testFileSplitter() throws Exception { this.input1.send(new GenericMessage<>(file)); Message receive = this.output.receive(10000); assertThat(receive).isNotNull(); //HelloWorld @@ -156,14 +149,11 @@ public class FileSplitterTests { assertThat(receive).isNotNull(); //äöüß assertThat(this.output.receive(1)).isNull(); - try { - this.input2.send(new GenericMessage<>("bar")); - fail("FileNotFoundException expected"); - } - catch (Exception e) { - assertThat(e.getCause()).isInstanceOf(FileNotFoundException.class); - assertThat(e.getMessage()).contains("Failed to read file [bar]"); - } + assertThatExceptionOfType(MessagingException.class) + .isThrownBy(() -> this.input2.send(new GenericMessage<>("bar"))) + .withCauseInstanceOf(FileNotFoundException.class) + .withMessageContaining("Failed to read file [bar]"); + this.input2.send(new GenericMessage<>(new Date())); receive = this.output.receive(10000); assertThat(receive).isNotNull(); @@ -190,7 +180,7 @@ public class FileSplitterTests { } @Test - public void testMarkers() { + void testMarkers() { QueueChannel outputChannel = new QueueChannel(); FileSplitter splitter = new FileSplitter(true, true); splitter.setOutputChannel(outputChannel); @@ -216,7 +206,7 @@ public class FileSplitterTests { } @Test - public void testMarkersEmptyFile() throws IOException { + void testMarkersEmptyFile() throws IOException { QueueChannel outputChannel = new QueueChannel(); FileSplitter splitter = new FileSplitter(true, true); splitter.setOutputChannel(outputChannel); @@ -244,7 +234,7 @@ public class FileSplitterTests { } @Test - public void testMarkersJson() throws Exception { + void testMarkersJson() throws Exception { JsonObjectMapper objectMapper = JsonObjectMapperProvider.newInstance(); QueueChannel outputChannel = new QueueChannel(); FileSplitter splitter = new FileSplitter(true, true, true); @@ -273,7 +263,7 @@ public class FileSplitterTests { } @Test - public void testFileSplitterReactive() { + void testFileSplitterReactive() { FluxMessageChannel outputChannel = new FluxMessageChannel(); FileSplitter splitter = new FileSplitter(true, true); splitter.setApplySequence(true); @@ -300,14 +290,13 @@ public class FileSplitterTests { assertThat(fileMarker.getFilePath()).isEqualTo(file.getAbsolutePath()); assertThat(fileMarker.getLineCount()).isEqualTo(2); }) - .then(() -> - ((Subscriber) TestUtils.getPropertyValue(outputChannel, "subscribers", List.class).get(0)) - .onComplete()) - .verifyComplete(); + .expectNoEvent(Duration.ofMillis(100)) + .thenCancel() + .verify(Duration.ofSeconds(1)); } @Test - public void testFirstLineAsHeader() { + void testFirstLineAsHeader() { QueueChannel outputChannel = new QueueChannel(); FileSplitter splitter = new FileSplitter(true, true); splitter.setFirstLineAsHeader("firstLine"); @@ -337,7 +326,7 @@ public class FileSplitterTests { } @Test - public void testFirstLineAsHeaderOnlyHeader() throws IOException { + void testFirstLineAsHeaderOnlyHeader() throws IOException { QueueChannel outputChannel = new QueueChannel(); FileSplitter splitter = new FileSplitter(true, true); splitter.setFirstLineAsHeader("firstLine"); @@ -365,7 +354,7 @@ public class FileSplitterTests { } @Test - public void testFileReaderClosedOnException() throws Exception { + void testFileReaderClosedOnException() throws Exception { DirectChannel outputChannel = new DirectChannel(); outputChannel.subscribe(message -> { throw new RuntimeException(); diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java index 64c9b3bd32..129ccacfb9 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/outbound/WebFluxRequestExecutingMessageHandlerTests.java @@ -18,10 +18,9 @@ package org.springframework.integration.webflux.outbound; import static org.assertj.core.api.Assertions.assertThat; -import java.util.List; +import java.time.Duration; -import org.junit.Test; -import org.reactivestreams.Subscriber; +import org.junit.jupiter.api.Test; import org.springframework.core.io.buffer.DataBuffer; import org.springframework.core.io.buffer.DataBufferFactory; @@ -33,7 +32,6 @@ import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.http.HttpHeaders; import org.springframework.integration.support.MessageBuilder; -import org.springframework.integration.test.util.TestUtils; import org.springframework.integration.webflux.support.ClientHttpResponseBodyExtractor; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandlingException; @@ -52,10 +50,10 @@ import reactor.test.StepVerifier; * * @since 5.0 */ -public class WebFluxRequestExecutingMessageHandlerTests { +class WebFluxRequestExecutingMessageHandlerTests { @Test - public void testReactiveReturn() { + void testReactiveReturn() { ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { response.setStatusCode(HttpStatus.OK); @@ -78,14 +76,13 @@ public class WebFluxRequestExecutingMessageHandlerTests { StepVerifier.create(ackChannel, 2) .assertNext(m -> assertThat(m.getHeaders()).containsEntry(HttpHeaders.STATUS_CODE, HttpStatus.OK)) .assertNext(m -> assertThat(m.getHeaders()).containsEntry(HttpHeaders.STATUS_CODE, HttpStatus.OK)) - .then(() -> - ((Subscriber) TestUtils.getPropertyValue(ackChannel, "subscribers", List.class).get(0)) - .onComplete()) - .verifyComplete(); + .expectNoEvent(Duration.ofMillis(100)) + .thenCancel() + .verify(Duration.ofSeconds(1)); } @Test - public void testReactiveErrorOneWay() { + void testReactiveErrorOneWay() { ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { response.setStatusCode(HttpStatus.UNAUTHORIZED); @@ -115,7 +112,7 @@ public class WebFluxRequestExecutingMessageHandlerTests { } @Test - public void testReactiveConnectErrorOneWay() { + void testReactiveConnectErrorOneWay() { ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { throw new RuntimeException("Intentional connection error"); @@ -144,7 +141,7 @@ public class WebFluxRequestExecutingMessageHandlerTests { } @Test - public void testServiceUnavailableWithoutBody() { + void testServiceUnavailableWithoutBody() { ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { response.setStatusCode(HttpStatus.SERVICE_UNAVAILABLE); @@ -185,7 +182,7 @@ public class WebFluxRequestExecutingMessageHandlerTests { @Test @SuppressWarnings("unchecked") - public void testFluxReply() { + void testFluxReply() { ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { response.setStatusCode(HttpStatus.OK); response.getHeaders().setContentType(MediaType.TEXT_PLAIN); @@ -227,7 +224,7 @@ public class WebFluxRequestExecutingMessageHandlerTests { } @Test - public void testClientHttpResponseAsReply() { + void testClientHttpResponseAsReply() { ClientHttpConnector httpConnector = new HttpHandlerConnector((request, response) -> { response.setStatusCode(HttpStatus.OK); response.getHeaders().setContentType(MediaType.TEXT_PLAIN); @@ -276,5 +273,4 @@ public class WebFluxRequestExecutingMessageHandlerTests { .verifyComplete(); } - }