Add tests for CT + Kafka + DevServices

extensions/smallrye-reactive-messaging-kafka/deployment/pom.xml

@@ -60,14 +60,45 @@
         </dependency>
         <dependency>
             <groupId>io.quarkus</groupId>
-            <artifactId>quarkus-jsonb-deployment</artifactId>
+            <artifactId>quarkus-resteasy-jsonb-deployment</artifactId>
             <scope>test</scope>
         </dependency>
         <dependency>
             <groupId>io.apicurio</groupId>
             <artifactId>apicurio-registry-serdes-avro-serde</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-junit5</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-rest-client-deployment</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-apicurio-registry-avro-deployment</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.rest-assured</groupId>
+            <artifactId>rest-assured</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>io.quarkus</groupId>
+            <artifactId>quarkus-vertx-http-deployment</artifactId>
+            <type>test-jar</type>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
     <build>
@@ -84,7 +115,33 @@
                     </annotationProcessorPaths>
                 </configuration>
             </plugin>
+            <plugin>
+                <artifactId>maven-surefire-plugin</artifactId>
+                <configuration>
+                    <skip>true</skip>
+                </configuration>
+            </plugin>
         </plugins>
     </build>
 
+    <profiles>
+        <profile>
+            <id>test-kafka</id>
+            <activation>
+                <property>
+                    <name>start-containers</name>
+                </property>
+            </activation>
+            <build>
+                <plugins>
+                    <plugin>
+                        <artifactId>maven-surefire-plugin</artifactId>
+                        <configuration>
+                            <skip>false</skip>
+                        </configuration>
+                    </plugin>
+                </plugins>
+            </build>
+        </profile>
+    </profiles>
 </project>

extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/testing/KafkaDevServicesContinuousTestingTestCase.java

@@ -0,0 +1,91 @@
+package io.quarkus.smallrye.reactivemessaging.kafka.deployment.testing;
+
+import java.util.function.Supplier;
+
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.asset.StringAsset;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusDevModeTest;
+import io.quarkus.vertx.http.testrunner.ContinuousTestingTestUtils;
+
+public class KafkaDevServicesContinuousTestingTestCase {
+
+    static final String FINAL_APP_PROPERTIES = ContinuousTestingTestUtils.appProperties(
+            "mp.messaging.outgoing.generated-price.connector=smallrye-kafka",
+            "mp.messaging.outgoing.generated-price.topic=prices",
+            "mp.messaging.outgoing.generated-price.value.serializer=org.apache.kafka.common.serialization.IntegerSerializer",
+            "mp.messaging.incoming.prices.connector=smallrye-kafka",
+            "mp.messaging.incoming.prices.health-readiness-enabled=false",
+            "mp.messaging.incoming.prices.topic=prices",
+            "mp.messaging.incoming.prices.value.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer");
+
+    @RegisterExtension
+    public static QuarkusDevModeTest test = new QuarkusDevModeTest()
+            .setArchiveProducer(new Supplier<JavaArchive>() {
+                @Override
+                public JavaArchive get() {
+                    return ShrinkWrap.create(JavaArchive.class)
+                            .addClasses(PriceConverter.class, PriceResource.class, PriceGenerator.class)
+                            .addAsResource(new StringAsset(ContinuousTestingTestUtils.appProperties("")),
+                                    "application.properties");
+                }
+            }).setTestArchiveProducer(new Supplier<JavaArchive>() {
+                @Override
+                public JavaArchive get() {
+                    return ShrinkWrap.create(JavaArchive.class).addClass(PriceResourceET.class);
+                }
+            });
+
+    //see https://github.com/quarkusio/quarkus/issues/19180
+    @Test
+    public void testContinuousTestingScenario1() {
+        ContinuousTestingTestUtils utils = new ContinuousTestingTestUtils();
+        var result = utils.waitForNextCompletion();
+        Assertions.assertEquals(0, result.getTotalTestsPassed());
+        Assertions.assertEquals(1, result.getTotalTestsFailed());
+        test.modifySourceFile(PriceResource.class, s -> s.replaceAll("//", ""));
+        result = utils.waitForNextCompletion();
+        Assertions.assertEquals(0, result.getTotalTestsPassed());
+        Assertions.assertEquals(1, result.getTotalTestsFailed());
+        test.modifySourceFile(PriceConverter.class, s -> s.replaceAll("//", ""));
+        result = utils.waitForNextCompletion();
+        Assertions.assertEquals(0, result.getTotalTestsPassed());
+        Assertions.assertEquals(1, result.getTotalTestsFailed());
+        test.modifySourceFile(PriceGenerator.class, s -> s.replaceAll("//", ""));
+        result = utils.waitForNextCompletion();
+        Assertions.assertEquals(0, result.getTotalTestsPassed());
+        Assertions.assertEquals(1, result.getTotalTestsFailed());
+        test.modifyResourceFile("application.properties", s -> FINAL_APP_PROPERTIES);
+        result = utils.waitForNextCompletion();
+        Assertions.assertEquals(1, result.getTotalTestsPassed());
+        Assertions.assertEquals(0, result.getTotalTestsFailed());
+    }
+
+    @Test
+    public void testContinuousTestingScenario2() {
+        ContinuousTestingTestUtils utils = new ContinuousTestingTestUtils();
+        var result = utils.waitForNextCompletion();
+        Assertions.assertEquals(0, result.getTotalTestsPassed());
+        Assertions.assertEquals(1, result.getTotalTestsFailed());
+        test.modifyResourceFile("application.properties", s -> FINAL_APP_PROPERTIES);
+        result = utils.waitForNextCompletion();
+        Assertions.assertEquals(0, result.getTotalTestsPassed());
+        Assertions.assertEquals(1, result.getTotalTestsFailed());
+        test.modifySourceFile(PriceConverter.class, s -> s.replaceAll("//", ""));
+        result = utils.waitForNextCompletion();
+        Assertions.assertEquals(0, result.getTotalTestsPassed());
+        Assertions.assertEquals(1, result.getTotalTestsFailed());
+        test.modifySourceFile(PriceGenerator.class, s -> s.replaceAll("//", ""));
+        result = utils.waitForNextCompletion();
+        Assertions.assertEquals(0, result.getTotalTestsPassed());
+        Assertions.assertEquals(1, result.getTotalTestsFailed());
+        test.modifySourceFile(PriceResource.class, s -> s.replaceAll("//", ""));
+        result = utils.waitForNextCompletion();
+        Assertions.assertEquals(1, result.getTotalTestsPassed());
+        Assertions.assertEquals(0, result.getTotalTestsFailed());
+    }
+}

extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/testing/KafkaDevServicesContinuousTestingWorkingAppPropsTestCase.java

@@ -0,0 +1,59 @@
+package io.quarkus.smallrye.reactivemessaging.kafka.deployment.testing;
+
+import java.util.function.Supplier;
+
+import org.jboss.shrinkwrap.api.ShrinkWrap;
+import org.jboss.shrinkwrap.api.asset.StringAsset;
+import org.jboss.shrinkwrap.api.spec.JavaArchive;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+import io.quarkus.test.QuarkusDevModeTest;
+import io.quarkus.vertx.http.testrunner.ContinuousTestingTestUtils;
+
+public class KafkaDevServicesContinuousTestingWorkingAppPropsTestCase {
+
+    @RegisterExtension
+    public static QuarkusDevModeTest test = new QuarkusDevModeTest()
+            .setArchiveProducer(new Supplier<JavaArchive>() {
+                @Override
+                public JavaArchive get() {
+                    return ShrinkWrap.create(JavaArchive.class)
+                            .addClasses(PriceConverter.class, PriceResource.class, PriceGenerator.class)
+                            .addAsResource(new StringAsset(KafkaDevServicesContinuousTestingTestCase.FINAL_APP_PROPERTIES),
+                                    "application.properties");
+                }
+            }).setTestArchiveProducer(new Supplier<JavaArchive>() {
+                @Override
+                public JavaArchive get() {
+                    return ShrinkWrap.create(JavaArchive.class).addClass(PriceResourceET.class);
+                }
+            });
+
+    /**
+     * Similar to {@link KafkaDevServicesContinuousTestingTestCase} however it starts with application.properties configued
+     *
+     * see https://github.com/quarkusio/quarkus/issues/19180
+     */
+    @Test
+    public void testContinuousTestingScenario3() {
+        ContinuousTestingTestUtils utils = new ContinuousTestingTestUtils();
+        var result = utils.waitForNextCompletion();
+        Assertions.assertEquals(0, result.getTotalTestsPassed());
+        Assertions.assertEquals(1, result.getTotalTestsFailed());
+        test.modifySourceFile(PriceResource.class, s -> s.replaceAll("//", ""));
+        result = utils.waitForNextCompletion();
+        Assertions.assertEquals(0, result.getTotalTestsPassed());
+        Assertions.assertEquals(1, result.getTotalTestsFailed());
+        test.modifySourceFile(PriceConverter.class, s -> s.replaceAll("//", ""));
+        result = utils.waitForNextCompletion();
+        Assertions.assertEquals(0, result.getTotalTestsPassed());
+        Assertions.assertEquals(1, result.getTotalTestsFailed());
+        test.modifySourceFile(PriceGenerator.class, s -> s.replaceAll("//", ""));
+        result = utils.waitForNextCompletion();
+        Assertions.assertEquals(1, result.getTotalTestsPassed());
+        Assertions.assertEquals(0, result.getTotalTestsFailed());
+    }
+
+}

extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/testing/PriceConverter.java

@@ -0,0 +1,20 @@
+package io.quarkus.smallrye.reactivemessaging.kafka.deployment.testing;
+
+//
+//import io.smallrye.reactive.messaging.annotations.Broadcast;
+//import org.eclipse.microprofile.reactive.messaging.Incoming;
+//import org.eclipse.microprofile.reactive.messaging.Outgoing;
+//
+//import javax.enterprise.context.ApplicationScoped;
+//
+//@ApplicationScoped
+public class PriceConverter {
+    //    private static final double CONVERSION_RATE = 0.88;
+    //
+    //    @Incoming("prices")
+    //    @Outgoing("processed-prices")
+    //    @Broadcast
+    //    public double process(int priceInUsd) {
+    //        return priceInUsd * CONVERSION_RATE;
+    //    }
+}

extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/testing/PriceGenerator.java

@@ -0,0 +1,20 @@
+package io.quarkus.smallrye.reactivemessaging.kafka.deployment.testing;
+
+//import io.smallrye.mutiny.Multi;
+//import org.eclipse.microprofile.reactive.messaging.Outgoing;
+//
+//import javax.enterprise.context.ApplicationScoped;
+//import java.time.Duration;
+//import java.util.Random;
+//
+//@ApplicationScoped
+public class PriceGenerator {
+    //        private final Random random = new Random();
+    //
+    //        @Outgoing("generated-price")
+    //        public Multi<Integer> generate() {
+    //            return Multi.createFrom().ticks().every(Duration.ofMillis(10))
+    //                    .onOverflow().drop()
+    //                    .map(tick -> this.random.nextInt(100));
+    //        }
+}

extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/testing/PriceResource.java

@@ -0,0 +1,26 @@
+package io.quarkus.smallrye.reactivemessaging.kafka.deployment.testing;
+
+//
+//import org.eclipse.microprofile.reactive.messaging.Channel;
+//import org.reactivestreams.Publisher;
+//
+//import javax.ws.rs.GET;
+//import javax.ws.rs.Path;
+//import javax.ws.rs.Produces;
+//import javax.ws.rs.core.MediaType;
+//
+//@Path("/prices")
+public class PriceResource {
+    //    private final Publisher<Double> processedPrices;
+    //
+    //    public PriceResource(@Channel("processed-prices") Publisher<Double> processedPrices) {
+    //        this.processedPrices = processedPrices;
+    //    }
+    //
+    //    @GET
+    //    @Path("/stream")
+    //    @Produces(MediaType.SERVER_SENT_EVENTS)
+    //    public Publisher<Double> ssePrices() {
+    //        return this.processedPrices;
+    //    }
+}

extensions/smallrye-reactive-messaging-kafka/deployment/src/test/java/io/quarkus/smallrye/reactivemessaging/kafka/deployment/testing/PriceResourceET.java

@@ -0,0 +1,47 @@
+package io.quarkus.smallrye.reactivemessaging.kafka.deployment.testing;
+
+import java.net.URI;
+import java.time.Duration;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.client.WebTarget;
+import javax.ws.rs.sse.SseEventSource;
+
+import org.assertj.core.api.Assertions;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+import io.quarkus.test.common.http.TestHTTPEndpoint;
+import io.quarkus.test.common.http.TestHTTPResource;
+import io.quarkus.test.junit.QuarkusTest;
+
+@QuarkusTest
+class PriceResourceET {
+    @TestHTTPEndpoint(PriceResource.class)
+    @TestHTTPResource("/stream")
+    URI uri;
+
+    @Test
+    public void sseStream() {
+        Client client = ClientBuilder.newClient();
+        WebTarget target = client.target(this.uri);
+
+        List<Double> received = new CopyOnWriteArrayList<>();
+
+        try (SseEventSource source = SseEventSource.target(target).build()) {
+            source.register(inboundSseEvent -> received.add(Double.valueOf(inboundSseEvent.readData())));
+            source.open();
+
+            Awaitility.await()
+                    .atMost(Duration.ofSeconds(1))
+                    .until(() -> received.size() >= 2);
+        }
+
+        Assertions.assertThat(received)
+                .hasSizeGreaterThan(2)
+                .allMatch(value -> (value >= 0) && (value < 100));
+    }
+}