Testkit: introduce log capturing (akka#2040)

Touches all modules to introduce log capturing which makes the logs show when tests fail.

amqp/src/test/java/akka/stream/alpakka/amqp/javadsl/AmqpConnectionProvidersTest.java

@@ -7,15 +7,19 @@
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.junit.Assert.*;
 
-import akka.japi.Pair;
 import akka.stream.alpakka.amqp.*;
+import akka.stream.alpakka.testkit.javadsl.LogCapturingJunit4;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
+import org.junit.Rule;
 import org.junit.Test;
 
 import java.net.ConnectException;
 
 public class AmqpConnectionProvidersTest {
+
+  @Rule public final LogCapturingJunit4 logCapturing = new LogCapturingJunit4();
+
   @Test
   public void LocalAmqpConnectionCreatesNewConnection() throws Exception {
     AmqpConnectionProvider connectionProvider = AmqpLocalConnectionProvider.getInstance();

amqp/src/test/java/akka/stream/alpakka/amqp/javadsl/AmqpConnectorsTest.java

@@ -13,6 +13,7 @@
 import akka.stream.Materializer;
 import akka.stream.UniqueKillSwitch;
 import akka.stream.alpakka.amqp.*;
+import akka.stream.alpakka.testkit.javadsl.LogCapturingJunit4;
 import akka.stream.javadsl.Flow;
 import akka.stream.javadsl.Keep;
 import akka.stream.javadsl.Sink;
@@ -23,10 +24,7 @@
 import akka.testkit.javadsl.TestKit;
 import akka.util.ByteString;
 import com.rabbitmq.client.AuthenticationFailureException;
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
+import org.junit.*;
 import scala.collection.JavaConverters;
 import scala.concurrent.duration.Duration;
 
@@ -45,6 +43,8 @@
 /** Needs a local running AMQP server on the default port with no password. */
 public class AmqpConnectorsTest {
 
+  @Rule public final LogCapturingJunit4 logCapturing = new LogCapturingJunit4();
+
   private static ActorSystem system;
   private static Materializer materializer;
 

amqp/src/test/java/akka/stream/alpakka/amqp/javadsl/AmqpFlowTest.java

@@ -10,7 +10,9 @@
 import java.util.concurrent.CompletionStage;
 import java.util.stream.Collectors;
 
+import akka.stream.alpakka.testkit.javadsl.LogCapturingJunit4;
 import org.junit.BeforeClass;
+import org.junit.Rule;
 import org.junit.Test;
 
 import akka.Done;
@@ -35,6 +37,8 @@
 /** Needs a local running AMQP server on the default port with no password. */
 public class AmqpFlowTest {
 
+  @Rule public final LogCapturingJunit4 logCapturing = new LogCapturingJunit4();
+
   private static ActorSystem system;
   private static Materializer materializer;
 

amqp/src/test/java/docs/javadsl/AmqpDocsTest.java

@@ -18,6 +18,7 @@
 import akka.stream.alpakka.amqp.javadsl.AmqpSink;
 import akka.stream.alpakka.amqp.javadsl.AmqpSource;
 import akka.stream.alpakka.amqp.javadsl.CommittableReadResult;
+import akka.stream.alpakka.testkit.javadsl.LogCapturingJunit4;
 import akka.stream.javadsl.Flow;
 import akka.stream.javadsl.Keep;
 import akka.stream.javadsl.Sink;
@@ -28,10 +29,7 @@
 import akka.testkit.javadsl.TestKit;
 import akka.util.ByteString;
 
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
+import org.junit.*;
 
 import java.time.Duration;
 import java.util.*;
@@ -45,6 +43,8 @@
 /** Needs a local running AMQP server on the default port with no password. */
 public class AmqpDocsTest {
 
+  @Rule public final LogCapturingJunit4 logCapturing = new LogCapturingJunit4();
+
   private static ActorSystem system;
   private static Materializer materializer;
 

amqp/src/test/resources/logback-test.xml

@@ -7,7 +7,22 @@
         </encoder>
     </appender>
 
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} %-5level [%-20.20thread] %-36.36logger{36}  %msg%n%rEx</pattern>
+        </encoder>
+    </appender>
+
+    <appender name="CapturingAppender" class="akka.stream.alpakka.testkit.CapturingAppender"/>
+
+    <logger name="akka.stream.alpakka.testkit.CapturingAppenderDelegate">
+        <appender-ref ref="STDOUT"/>
+    </logger>
+
+    <logger name="akka" level="DEBUG"/>
+
     <root level="debug">
+        <appender-ref ref="CapturingAppender"/>
         <appender-ref ref="FILE" />
     </root>
-</configuration>
+</configuration>

amqp/src/test/scala/akka/stream/alpakka/amqp/AmqpSpec.scala

@@ -7,12 +7,13 @@ package akka.stream.alpakka.amqp
 import akka.actor.ActorSystem
 import akka.dispatch.ExecutionContexts
 import akka.stream.ActorMaterializer
+import akka.stream.alpakka.testkit.scaladsl.LogCapturing
 import org.scalatest.concurrent.ScalaFutures
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.wordspec.AnyWordSpec
 
-abstract class AmqpSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with ScalaFutures {
+abstract class AmqpSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll with ScalaFutures with LogCapturing {
 
   implicit val system = ActorSystem(this.getClass.getSimpleName)
   implicit val materializer = ActorMaterializer()

amqp/src/test/scala/akka/stream/alpakka/amqp/scaladsl/AmqpGraphStageLogicConnectionShutdownSpec.scala

@@ -17,6 +17,7 @@ import akka.stream.alpakka.amqp.{
   AmqpWriteSettings,
   QueueDeclaration
 }
+import akka.stream.alpakka.testkit.scaladsl.LogCapturing
 import akka.stream.scaladsl.Source
 import akka.util.ByteString
 import com.rabbitmq.client.{AddressResolver, Connection, ConnectionFactory, ShutdownListener}
@@ -37,7 +38,8 @@ class AmqpGraphStageLogicConnectionShutdownSpec
     extends AnyWordSpec
     with Matchers
     with ScalaFutures
-    with BeforeAndAfterEach {
+    with BeforeAndAfterEach
+    with LogCapturing {
 
   override implicit val patienceConfig = PatienceConfig(10.seconds)
   private implicit val executionContext = ExecutionContexts.sameThreadExecutionContext

avroparquet/src/test/java/docs/javadsl/AvroParquetSinkTest.java

@@ -9,6 +9,7 @@
 import akka.stream.ActorMaterializer;
 import akka.stream.Materializer;
 import akka.stream.alpakka.avroparquet.javadsl.AvroParquetSink;
+import akka.stream.alpakka.testkit.javadsl.LogCapturingJunit4;
 import akka.stream.javadsl.Sink;
 import akka.stream.javadsl.Source;
 import akka.stream.testkit.javadsl.StreamTestKit;
@@ -21,10 +22,10 @@
 import org.apache.parquet.hadoop.ParquetReader;
 import org.junit.After;
 import org.junit.Before;
+import org.junit.Rule;
 import org.junit.Test;
 import java.io.File;
 import java.io.IOException;
-import java.nio.file.Files;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CompletionStage;
@@ -33,7 +34,6 @@
 import java.util.concurrent.TimeoutException;
 
 import static junit.framework.TestCase.assertEquals;
-import static junit.framework.TestCase.fail;
 
 // #init-writer
 import org.apache.parquet.hadoop.ParquetWriter;
@@ -47,6 +47,8 @@
 
 public class AvroParquetSinkTest {
 
+  @Rule public final LogCapturingJunit4 logCapturing = new LogCapturingJunit4();
+
   private final Schema schema =
       new Schema.Parser()
           .parse(

avroparquet/src/test/resources/logback-test.xml

@@ -7,7 +7,22 @@
         </encoder>
     </appender>
 
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} %-5level [%-20.20thread] %-36.36logger{36}  %msg%n%rEx</pattern>
+        </encoder>
+    </appender>
+
+    <appender name="CapturingAppender" class="akka.stream.alpakka.testkit.CapturingAppender"/>
+
+    <logger name="akka.stream.alpakka.testkit.CapturingAppenderDelegate">
+        <appender-ref ref="STDOUT"/>
+    </logger>
+
+    <logger name="akka" level="DEBUG"/>
+
     <root level="debug">
+        <appender-ref ref="CapturingAppender"/>
         <appender-ref ref="FILE" />
     </root>
-</configuration>
+</configuration>

awslambda/src/test/java/docs/javadsl/AwsLambdaFlowTest.java

@@ -8,32 +8,31 @@
 import akka.actor.ActorSystem;
 import akka.stream.ActorMaterializer;
 import akka.stream.alpakka.awslambda.javadsl.AwsLambdaFlow;
+import akka.stream.alpakka.testkit.javadsl.LogCapturingJunit4;
 import akka.stream.javadsl.Flow;
 import akka.stream.javadsl.Sink;
 import akka.stream.javadsl.Source;
 import akka.stream.testkit.javadsl.StreamTestKit;
 import akka.testkit.javadsl.TestKit;
+import org.junit.*;
 import software.amazon.awssdk.services.lambda.LambdaAsyncClient;
 import software.amazon.awssdk.services.lambda.model.InvokeRequest;
 import software.amazon.awssdk.services.lambda.model.InvokeResponse;
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
 
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.Assert.assertEquals;
-import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 public class AwsLambdaFlowTest {
 
+  @Rule public final LogCapturingJunit4 logCapturing = new LogCapturingJunit4();
+
   private static ActorSystem system;
   private static ActorMaterializer materializer;
   private static LambdaAsyncClient awsLambdaClient;

awslambda/src/test/resources/logback-test.xml

@@ -7,7 +7,22 @@
         </encoder>
     </appender>
 
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} %-5level [%-20.20thread] %-36.36logger{36}  %msg%n%rEx</pattern>
+        </encoder>
+    </appender>
+
+    <appender name="CapturingAppender" class="akka.stream.alpakka.testkit.CapturingAppender"/>
+
+    <logger name="akka.stream.alpakka.testkit.CapturingAppenderDelegate">
+        <appender-ref ref="STDOUT"/>
+    </logger>
+
+    <logger name="akka" level="DEBUG"/>
+
     <root level="debug">
+        <appender-ref ref="CapturingAppender"/>
         <appender-ref ref="FILE" />
     </root>
-</configuration>
+</configuration>

awslambda/src/test/scala/docs/scaladsl/AwsLambdaFlowSpec.scala

@@ -9,6 +9,7 @@ import java.util.concurrent.CompletableFuture
 import akka.actor.ActorSystem
 import akka.stream.ActorMaterializer
 import akka.stream.alpakka.awslambda.scaladsl.AwsLambdaFlow
+import akka.stream.alpakka.testkit.scaladsl.LogCapturing
 import akka.stream.scaladsl.{Keep, Sink}
 import akka.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
 import akka.stream.testkit.scaladsl.TestSource
@@ -36,7 +37,8 @@ class AwsLambdaFlowSpec
     with BeforeAndAfterEach
     with ScalaFutures
     with Matchers
-    with MockitoSugar {
+    with MockitoSugar
+    with LogCapturing {
 
   implicit val mat = ActorMaterializer()
 

azure-storage-queue/src/test/java/docs/javadsl/JavaDslTest.java

@@ -10,6 +10,7 @@
 import akka.stream.ActorMaterializer;
 import akka.stream.alpakka.azure.storagequeue.*;
 import akka.stream.alpakka.azure.storagequeue.javadsl.*;
+import akka.stream.alpakka.testkit.javadsl.LogCapturingJunit4;
 import akka.stream.javadsl.Sink;
 import akka.stream.javadsl.Source;
 import akka.stream.testkit.javadsl.StreamTestKit;
@@ -28,6 +29,8 @@
 import org.junit.*;
 
 public class JavaDslTest {
+  @Rule public final LogCapturingJunit4 logCapturing = new LogCapturingJunit4();
+
   private static ActorSystem system;
   private static ActorMaterializer materializer;
   private static final String storageConnectionString = System.getenv("AZURE_CONNECTION_STRING");

azure-storage-queue/src/test/resources/logback-test.xml

@@ -7,7 +7,22 @@
         </encoder>
     </appender>
 
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} %-5level [%-20.20thread] %-36.36logger{36}  %msg%n%rEx</pattern>
+        </encoder>
+    </appender>
+
+    <appender name="CapturingAppender" class="akka.stream.alpakka.testkit.CapturingAppender"/>
+
+    <logger name="akka.stream.alpakka.testkit.CapturingAppenderDelegate">
+        <appender-ref ref="STDOUT"/>
+    </logger>
+
+    <logger name="akka" level="DEBUG"/>
+
     <root level="debug">
+        <appender-ref ref="CapturingAppender"/>
         <appender-ref ref="FILE" />
     </root>
-</configuration>
+</configuration>

build.sbt

@@ -263,6 +263,7 @@ lazy val orientdb = alpakkaProject("orientdb",
                                    fatalWarnings := false)
 
 lazy val reference = internalProject("reference", Dependencies.Reference)
+  .dependsOn(testkit % Test)
 
 lazy val s3 = alpakkaProject("s3", "aws.s3", Dependencies.S3)
 
@@ -363,6 +364,8 @@ lazy val docs = project
     apidocRootPackage := "akka"
   )
 
+lazy val testkit = internalProject("testkit", Dependencies.testkit)
+
 lazy val whitesourceSupported = project
   .in(file("tmp"))
   .settings(whitesourceGroup := Whitesource.Group.Supported)
@@ -400,6 +403,7 @@ def alpakkaProject(projectId: String, moduleName: String, additionalSettings: sb
       mimaBinaryIssueFilters += ProblemFilters.exclude[Problem]("*.impl.*")
     )
     .settings(additionalSettings: _*)
+    .dependsOn(testkit % Test)
 }
 
 def internalProject(projectId: String, additionalSettings: sbt.Def.SettingsDefinition*): Project =

cassandra/src/test/java/docs/javadsl/CassandraSourceTest.java

@@ -10,15 +10,13 @@
 import akka.stream.alpakka.cassandra.javadsl.CassandraFlow;
 import akka.stream.alpakka.cassandra.javadsl.CassandraSink;
 import akka.stream.alpakka.cassandra.javadsl.CassandraSource;
+import akka.stream.alpakka.testkit.javadsl.LogCapturingJunit4;
 import akka.stream.javadsl.Flow;
 import akka.stream.javadsl.Source;
 import akka.stream.testkit.javadsl.StreamTestKit;
 import akka.testkit.javadsl.TestKit;
 import com.datastax.driver.core.*;
-import org.junit.After;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
+import org.junit.*;
 
 import static org.junit.Assert.*;
 
@@ -39,6 +37,8 @@
 /** All the tests must be run with a local Cassandra running on default port 9042. */
 public class CassandraSourceTest {
 
+  @Rule public final LogCapturingJunit4 logCapturing = new LogCapturingJunit4();
+
   // #element-to-insert
   private class ToInsert {
     Integer id;