Adding assertion to sse test

Author: edeandrea

chapter-5/chapter-5-quarkus-kafka-streams/src/test/java/org/acme/rest/PriceResourceTest.java

@@ -1,5 +1,6 @@
 package org.acme.rest;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 
 import java.net.URI;
@@ -31,10 +32,17 @@ void testPricesEventStream() {
 
         List<Double> received = new CopyOnWriteArrayList<>();
 
-        SseEventSource source = SseEventSource.target(target).build();
-        source.register(inboundSseEvent -> received.add(Double.valueOf(inboundSseEvent.readData())));
-        source.open();
-        await().atMost(Duration.ofSeconds(20)).until(() -> received.size() == 3);
-        source.close();
+        try (SseEventSource source = SseEventSource.target(target).build()) {
+            source.register(inboundSseEvent -> received.add(Double.valueOf(inboundSseEvent.readData())));
+            source.open();
+
+            await()
+              .atMost(Duration.ofSeconds(20))
+              .until(() -> received.size() == 3);
+        }
+
+        assertThat(received)
+          .hasSize(3)
+          .allMatch(value -> (value >= 0) && (value < 100));
     }
 }

chapter-5/chapter-5-quarkus-kafka-streams/src/test/java/org/acme/service/PriceGeneratorTests.java

@@ -4,7 +4,6 @@
 
 import java.time.Duration;
 import java.util.List;
-import java.util.function.Predicate;
 
 import org.junit.jupiter.api.Test;
 
@@ -13,23 +12,20 @@
 class PriceGeneratorTests {
 	PriceGenerator priceGenerator = new PriceGenerator();
 
-	private static final Predicate<Integer> VALUE_TEST =
-		value -> (value >= 0) && (value < 100);
-
 	@Test
 	public void generatesProperly() {
 		List<Integer> prices = this.priceGenerator.generate()
 			.select().first(2)
 			.subscribe().withSubscriber(AssertSubscriber.create(2))
 			.assertSubscribed()
-			.awaitNextItem(Duration.ofSeconds(7)) // Needs to be a second or 2 more than the timing of actual events
-			.awaitNextItem(Duration.ofSeconds(7))
+			.awaitNextItem(Duration.ofSeconds(8)) // Needs to be a few seconds more than the timing of actual events
+			.awaitNextItem(Duration.ofSeconds(8))
 			.awaitCompletion(Duration.ofSeconds(15))
 			.assertCompleted()
 			.getItems();
 
 		assertThat(prices)
 			.hasSize(2)
-			.allMatch(VALUE_TEST);
+			.allMatch(value -> (value >= 0) && (value < 100));
 	}
 }

chapter-5/chapter-5-spring-kafka-streams/src/test/java/org/acme/Chapter5SpringKafkaStreamsApplicationTests.java

@@ -36,12 +36,7 @@ void sseWorks() {
 
 		assertThat(emittedPrices)
 			.isNotNull()
-			.hasSize(3);
-
-		emittedPrices.forEach(price ->
-			assertThat(price)
-				.isNotNull()
-				.isGreaterThan(0)
-		);
+			.hasSize(3)
+			.allMatch(value -> (value >= 0) && (value < 100));
 	}
 }