diff --git a/build.gradle b/build.gradle index 8c70ad80..18570b7d 100644 --- a/build.gradle +++ b/build.gradle @@ -65,10 +65,11 @@ subprojects { subproject -> targetCompatibility = 1.7 ext { + assertjVersion = '3.3.0' avroVersion = '1.7.6' gsCollectionsVersion = '5.0.0' hamcrestVersion = '1.3' - junitVersion = '4.11' + junitVersion = '4.12' kafkaVersion = '0.9.0.1' log4jVersion = '1.2.17' mockitoVersion = '1.9.5' @@ -146,6 +147,7 @@ project ('spring-kafka') { compile "org.apache.kafka:kafka-clients:$kafkaVersion" testCompile project (":spring-kafka-test") + testCompile "org.assertj:assertj-core:$assertjVersion" } } @@ -167,8 +169,8 @@ project ('spring-kafka-test') { compile ("org.mockito:mockito-core:$mockitoVersion") { exclude group: 'org.hamcrest' } - compile "org.hamcrest:hamcrest-all:$hamcrestVersion" - + compile ("org.hamcrest:hamcrest-all:$hamcrestVersion", optional) + compile ("org.assertj:assertj-core:$assertjVersion", optional) } } diff --git a/spring-kafka-test/src/main/java/org/springframework/kafka/test/assertj/KafkaConditions.java b/spring-kafka-test/src/main/java/org/springframework/kafka/test/assertj/KafkaConditions.java new file mode 100644 index 00000000..ced1ad25 --- /dev/null +++ b/spring-kafka-test/src/main/java/org/springframework/kafka/test/assertj/KafkaConditions.java @@ -0,0 +1,108 @@ +/* + * Copyright 2016 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 + * + * 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.springframework.kafka.test.assertj; + +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.assertj.core.api.Condition; + +/** + * The AssertJ custom {@link Condition}s. + * + * @author Artem Bilan + */ +public final class KafkaConditions { + + private KafkaConditions() { + // private ctor + } + + /** + * @param key the key + * @param the type. + * @return a Condition that matches the key in a consumer record. + */ + public static Condition> key(K key) { + return new ConsumerRecordKeyCondition<>(key); + } + + /** + * @param value the value. + * @param the type. + * @return a Condition that matches the value in a consumer record. + */ + public static Condition> value(V value) { + return new ConsumerRecordValueCondition<>(value); + } + + /** + * @param partition the partition. + * @return a Condition that matches the partition in a consumer record. + */ + public static Condition> partition(int partition) { + return new ConsumerRecordPartitionCondition(partition); + } + + + public static class ConsumerRecordKeyCondition extends Condition> { + + private final K key; + + public ConsumerRecordKeyCondition(K key) { + super("a ConsumerRecord with 'key' " + key); + this.key = key; + } + + @Override + public boolean matches(ConsumerRecord value) { + return value != null && value.key().equals(this.key); + } + + } + + public static class ConsumerRecordValueCondition extends Condition> { + + private final V payload; + + public ConsumerRecordValueCondition(V payload) { + super("a ConsumerRecord with 'value' " + payload); + this.payload = payload; + } + + @Override + public boolean matches(ConsumerRecord value) { + return value != null && value.value().equals(this.payload); + } + + } + + public static class ConsumerRecordPartitionCondition extends Condition> { + + private final int partition; + + public ConsumerRecordPartitionCondition(int partition) { + super("a ConsumerRecord with 'partition' " + partition); + this.partition = partition; + } + + @Override + public boolean matches(ConsumerRecord value) { + return value != null && value.partition() == this.partition; + } + + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index 2ef31d9a..be1b40d8 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -16,9 +16,7 @@ package org.springframework.kafka.annotation; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertTrue; +import static org.assertj.core.api.Assertions.assertThat; import java.util.Map; import java.util.concurrent.CountDownLatch; @@ -56,6 +54,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Gary Russell + * @author Artem Bilan * */ @ContextConfiguration @@ -81,26 +80,26 @@ public class EnableKafkaIntegrationTests { waitListening("foo"); template.convertAndSend("annotated1", 0, "foo"); template.flush(); - assertTrue(this.listener.latch1.await(10, TimeUnit.SECONDS)); + assertThat(this.listener.latch1.await(10, TimeUnit.SECONDS)).isTrue(); waitListening("bar"); template.convertAndSend("annotated2", 0, "foo"); template.flush(); - assertTrue(this.listener.latch2.await(10, TimeUnit.SECONDS)); - assertNotNull(this.listener.partition); + assertThat(this.listener.latch2.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.listener.partition).isNotNull(); waitListening("baz"); template.convertAndSend("annotated3", 0, "foo"); template.flush(); - assertTrue(this.listener.latch3.await(10, TimeUnit.SECONDS)); - assertEquals("foo", this.listener.record.value()); + assertThat(this.listener.latch3.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.listener.record.value()).isEqualTo("foo"); waitListening("qux"); template.convertAndSend("annotated4", 0, "foo"); template.flush(); - assertTrue(this.listener.latch4.await(10, TimeUnit.SECONDS)); - assertEquals("foo", this.listener.record.value()); - assertNotNull(this.listener.ack); + assertThat(this.listener.latch4.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.listener.record.value()).isEqualTo("foo"); + assertThat(this.listener.ack).isNotNull(); waitListening("fiz"); template.convertAndSend("annotated5", 0, 0, "foo"); @@ -108,7 +107,7 @@ public class EnableKafkaIntegrationTests { template.convertAndSend("annotated6", 0, 0, "baz"); template.convertAndSend("annotated6", 1, 0, "qux"); template.flush(); - assertTrue(this.listener.latch5.await(10, TimeUnit.SECONDS)); + assertThat(this.listener.latch5.await(10, TimeUnit.SECONDS)).isTrue(); } private void waitListening(String id) throws InterruptedException { @@ -120,7 +119,7 @@ public class EnableKafkaIntegrationTests { while (n++ < 6000 && (kmlc.getAssignedPartitions() == null || kmlc.getAssignedPartitions().size() == 0)) { Thread.sleep(100); } - assertTrue(kmlc.getAssignedPartitions().size() > 0); + assertThat(kmlc.getAssignedPartitions().size()).isGreaterThan(0); } @Configuration diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java index 35c5f7ef..8180d412 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java @@ -16,10 +16,10 @@ package org.springframework.kafka.core; -import static org.junit.Assert.assertThat; -import static org.springframework.kafka.test.hamcrest.KafkaMatchers.hasKey; -import static org.springframework.kafka.test.hamcrest.KafkaMatchers.hasPartition; -import static org.springframework.kafka.test.hamcrest.KafkaMatchers.hasValue; +import static org.assertj.core.api.Assertions.assertThat; +import static org.springframework.kafka.test.assertj.KafkaConditions.key; +import static org.springframework.kafka.test.assertj.KafkaConditions.partition; +import static org.springframework.kafka.test.assertj.KafkaConditions.value; import java.util.Map; import java.util.concurrent.BlockingQueue; @@ -39,6 +39,7 @@ import org.springframework.kafka.test.utils.KafkaTestUtils; /** * @author Gary Russell + * @author Artem Bilan * */ public class KafkaTemplateTests { @@ -73,17 +74,17 @@ public class KafkaTemplateTests { KafkaTemplate template = new KafkaTemplate<>(pf); template.setDefaultTopic(TEMPLATE_TOPIC); template.syncConvertAndSend("foo"); - assertThat(records.poll(10, TimeUnit.SECONDS), hasValue("foo")); + assertThat(records.poll(10, TimeUnit.SECONDS)).has(value("foo")); template.syncConvertAndSend(0, 2, "bar"); ConsumerRecord received = records.poll(10, TimeUnit.SECONDS); - assertThat(received, hasKey(2)); - assertThat(received, hasPartition(0)); - assertThat(received, hasValue("bar")); + assertThat(received).has(key(2)); + assertThat(received).has(partition(0)); + assertThat(received).has(value("bar")); template.syncConvertAndSend(TEMPLATE_TOPIC, 0, 2, "baz"); received = records.poll(10, TimeUnit.SECONDS); - assertThat(received, hasKey(2)); - assertThat(received, hasPartition(0)); - assertThat(received, hasValue("baz")); + assertThat(received).has(key(2)); + assertThat(received).has(partition(0)); + assertThat(received).has(value("baz")); } } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java index 9d002c57..aeddbeb5 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java @@ -16,11 +16,7 @@ package org.springframework.kafka.listener; -import static org.hamcrest.Matchers.anyOf; -import static org.hamcrest.Matchers.equalTo; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertThat; -import static org.junit.Assert.assertTrue; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Matchers.anyLong; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.mock; @@ -110,7 +106,7 @@ public class ConcurrentMessageListenerContainerTests { template.convertAndSend(0, "baz"); template.convertAndSend(2, "qux"); template.flush(); - assertTrue(latch.await(60, TimeUnit.SECONDS)); + assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue(); container.stop(); logger.info("Stop auto"); } @@ -144,7 +140,7 @@ public class ConcurrentMessageListenerContainerTests { template.convertAndSend(0, "baz"); template.convertAndSend(2, "qux"); template.flush(); - assertTrue(latch.await(60, TimeUnit.SECONDS)); + assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue(); container.stop(); logger.info("Stop manual"); } @@ -209,7 +205,7 @@ public class ConcurrentMessageListenerContainerTests { container2.setBeanName("b2"); container2.start(); - assertTrue(initialConsumersLatch.await(20, TimeUnit.SECONDS)); + assertThat(initialConsumersLatch.await(20, TimeUnit.SECONDS)).isTrue(); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); @@ -221,8 +217,8 @@ public class ConcurrentMessageListenerContainerTests { template.convertAndSend(2, "qux"); template.flush(); - assertTrue(latch1.await(60, TimeUnit.SECONDS)); - assertTrue(latch2.await(60, TimeUnit.SECONDS)); + assertThat(latch1.await(60, TimeUnit.SECONDS)).isTrue(); + assertThat(latch2.await(60, TimeUnit.SECONDS)).isTrue(); container1.stop(); container2.stop(); @@ -242,9 +238,9 @@ public class ConcurrentMessageListenerContainerTests { } }); resettingContainer.start(); - assertTrue(latch3.await(60, TimeUnit.SECONDS)); + assertThat(latch3.await(60, TimeUnit.SECONDS)).isTrue(); resettingContainer.stop(); - assertThat(latch3.getCount(), equalTo(0L)); + assertThat(latch3.getCount()).isEqualTo(0L); props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); cf = new DefaultKafkaConsumerFactory<>(props); @@ -264,10 +260,10 @@ public class ConcurrentMessageListenerContainerTests { } }); resettingContainer.start(); - assertTrue(latch4.await(60, TimeUnit.SECONDS)); + assertThat(latch4.await(60, TimeUnit.SECONDS)).isTrue(); resettingContainer.stop(); - assertThat(receivedMessage.get(), anyOf(equalTo("baz"), equalTo("qux"))); - assertThat(latch4.getCount(), equalTo(0L)); + assertThat(receivedMessage.get()).isIn("baz", "qux"); + assertThat(latch4.getCount()).isEqualTo(0L); logger.info("Stop auto parts"); } @@ -312,7 +308,7 @@ public class ConcurrentMessageListenerContainerTests { template.convertAndSend(0, "baz"); template.convertAndSend(2, "qux"); template.flush(); - assertTrue(latch.await(60, TimeUnit.SECONDS)); + assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue(); container.stop(); logger.info("Stop " + ackMode); } @@ -356,10 +352,10 @@ public class ConcurrentMessageListenerContainerTests { List> containers = (List>) new DirectFieldAccessor( container).getPropertyValue("containers"); - assertEquals(3, containers.size()); + assertThat(containers.size()).isEqualTo(3); for (int i = 0; i < 3; i++) { - assertEquals(i < 2 ? 2 : 3, ((TopicPartition[]) new DirectFieldAccessor(containers.get(i)) - .getPropertyValue("partitions")).length); + assertThat(((TopicPartition[]) new DirectFieldAccessor(containers.get(i)) + .getPropertyValue("partitions")).length).isEqualTo(i < 2 ? 2 : 3); } container.stop(); } @@ -394,7 +390,7 @@ public class ConcurrentMessageListenerContainerTests { template.convertAndSend(0, "baz"); template.convertAndSend(2, "qux"); template.flush(); - assertTrue(latch.await(60, TimeUnit.SECONDS)); + assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue(); container.stop(); logger.info("Stop exception"); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerTestUtils.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerTestUtils.java index d329e070..8fc39bff 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerTestUtils.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerTestUtils.java @@ -16,8 +16,7 @@ package org.springframework.kafka.listener; -import static org.hamcrest.Matchers.equalTo; -import static org.junit.Assert.assertThat; +import static org.assertj.core.api.Assertions.assertThat; import java.util.List; @@ -47,7 +46,7 @@ public final class ContainerTestUtils { Thread.sleep(100); } } - assertThat(count, equalTo(partitions)); + assertThat(count).isEqualTo(partitions); } public static void waitForAssignment(KafkaMessageListenerContainer container, int partitions) @@ -63,7 +62,7 @@ public final class ContainerTestUtils { Thread.sleep(100); } } - assertThat(count, equalTo(partitions)); + assertThat(count).isEqualTo(partitions); } } diff --git a/src/checkstyle/checkstyle.xml b/src/checkstyle/checkstyle.xml index 06474a4a..c2354582 100644 --- a/src/checkstyle/checkstyle.xml +++ b/src/checkstyle/checkstyle.xml @@ -81,16 +81,10 @@ org.junit.internal.matchers.ThrowableMessageMatcher.*, org.hamcrest.CoreMatchers.*, org.hamcrest.Matchers.*, - org.springframework.boot.configurationprocessor.ConfigurationMetadataMatchers.*, - org.springframework.boot.configurationprocessor.TestCompiler.*, org.mockito.Mockito.*, org.mockito.Matchers.*, - org.springframework.test.web.servlet.request.MockMvcRequestBuilders.*, - org.springframework.test.web.servlet.result.MockMvcResultMatchers.*, - org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.*, - org.springframework.security.test.web.servlet.setup.SecurityMockMvcConfigurers.*, - org.springframework.hateoas.mvc.ControllerLinkBuilder.linkTo, - org.springframework.kafka.test.hamcrest.KafkaMatchers.*" /> + org.springframework.kafka.test.hamcrest.KafkaMatchers.*, + org.springframework.kafka.test.assertj.KafkaConditions.*" /> @@ -106,10 +100,10 @@ - - - - + + + + @@ -157,13 +151,13 @@ - - - - - - - + + + + + + diff --git a/src/reference/asciidoc/testing.adoc b/src/reference/asciidoc/testing.adoc index b0ce4f11..8898309e 100644 --- a/src/reference/asciidoc/testing.adoc +++ b/src/reference/asciidoc/testing.adoc @@ -60,12 +60,14 @@ The `o.s.kafka.test.hamcrest.KafkaMatchers` provides the following matchers: ---- /** * @param key the key + * @param the type. * @return a Matcher that matches the key in a consumer record. */ public static Matcher> hasKey(K key) { ... } /** * @param value the value. + * @param the type. * @return a Matcher that matches the value in a consumer record. */ public static Matcher> hasValue(V value) { ... } @@ -77,6 +79,31 @@ public static Matcher> hasValue(V value) { ... } public static Matcher> hasPartition(int partition) { ... } ---- +==== AssertJ Conditions + +[source, java] +---- +/** + * @param key the key + * @param the type. + * @return a Condition that matches the key in a consumer record. + */ +public static Condition> key(K key) { ... } + +/** + * @param value the value. + * @param the type. + * @return a Condition that matches the value in a consumer record. + */ +public static Condition> value(V value) { ... } + +/** + * @param partition the partition. + * @return a Condition that matches the partition in a consumer record. + */ +public static Condition> partition(int partition) { ... } +---- + ==== Example Putting it all together: @@ -131,3 +158,23 @@ public class KafkaTemplateTests { } ---- + +The above uses the hamcrest matchers; with `AssertJ`, the final part looks like this... + +[source, java] +---- +... + assertThat(records.poll(10, TimeUnit.SECONDS)).has(value("foo")); + template.syncConvertAndSend(0, 2, "bar"); + ConsumerRecord received = records.poll(10, TimeUnit.SECONDS); + assertThat(received).has(key(2)); + assertThat(received).has(partition(0)); + assertThat(received).has(value("bar")); + template.syncConvertAndSend(TEMPLATE_TOPIC, 0, 2, "baz"); + received = records.poll(10, TimeUnit.SECONDS); + assertThat(received).has(key(2)); + assertThat(received).has(partition(0)); + assertThat(received).has(value("baz")); + } +} +----