Integrated Pipeline: CalculationParsingFlow test refactoring

integrated-pipeline/src/test/scala/eu/nomad_lab/integrated_pipeline_tests/CalculationParsingFlowSpec.scala

@@ -29,34 +29,25 @@ class CalculationParsingFlowSpec extends WordSpec with MockitoSugar with TestDat
     relativePath = Paths.get(fileName),
     parserName = "dummyParser"
   )
-  val unusedPath = Paths.get("/tmp/extracted/magic")
 
-  private class GraphWithDummyWorkers(numWorkers: Int) extends StreamAssertions[FileParsingResultSignal] {
+  private class Fixture(numWorkers: Int) extends StreamAssertions[FileParsingResultSignal] {
     require(numWorkers >= 1, "need at least one dummy worker")
     val dummyWorkers = (1 to numWorkers).map { index =>
       val myMock = mock[CalculationParsingEngine]
       when(myMock.processRequest(any(), any())).thenAnswer(new Answer[FileParsingResult] {
         override def answer(invocation: InvocationOnMock): FileParsingResult = {
-          val task = invocation.getArgument[FileParsingTask](0)
-          InMemoryResult(
-            task = task,
-            result = ParseResult.ParseSuccess,
-            start = None,
-            events = Seq(AddValue("dummy_worker_index", JInt(index))),
-            end = None
-          )
+          anInMemoryResult().
+            withTask(invocation.getArgument[FileParsingTask](0)).
+            withParseResult(ParseResult.ParseSuccess).
+            build()
         }
       })
       when(myMock.failParseRequest(any(), any())).thenAnswer(new Answer[FileParsingResult] {
         override def answer(invocation: InvocationOnMock): FileParsingResult = {
-          val task = invocation.getArgument[FileParsingTask](0)
-          InMemoryResult(
-            task = task,
-            result = ParseResult.ParseFailure,
-            start = None,
-            events = Seq(AddValue("dummy_worker_index", JInt(index))),
-            end = None
-          )
+          anInMemoryResult().
+            withTask(invocation.getArgument[FileParsingTask](0)).
+            withParseResult(ParseResult.ParseFailure).
+            build()
         }
       })
       myMock
@@ -65,7 +56,7 @@ class CalculationParsingFlowSpec extends WordSpec with MockitoSugar with TestDat
 
     private val testInput = TestSource.probe[FileParsingTaskSignal]
     private val testRequests = TestSink.probe[FileParsingResultSignal]
-    val testGraph = RunnableGraph.fromGraph(
+    private val testGraph = RunnableGraph.fromGraph(
       GraphDSL.create(testInput, testRequests)((_, _)) { implicit builder => (source, sink) =>
         import GraphDSL.Implicits._
 
@@ -84,19 +75,19 @@ class CalculationParsingFlowSpec extends WordSpec with MockitoSugar with TestDat
   "a CalculationParsingFlow" when {
     "managing a single worker" should {
 
-      "expect a file tree started signal before any incoming parsing requests" in new GraphWithDummyWorkers(1) {
+      "expect a file tree started signal before any incoming parsing requests" in new Fixture(1) {
         source.sendNext(input1)
         sink.ensureSubscription().request(1)
         sink.expectError()
       }
 
-      "forward a file tree started signal after processing it" in new GraphWithDummyWorkers(1) {
+      "forward a file tree started signal after processing it" in new Fixture(1) {
         source.sendNext(FileParsingTaskStartTree(sampleTreeScan))
         sink.ensureSubscription().request(1)
         sink.expectNext(FileParsingResultStartTree(sampleTreeScan))
       }
 
-      "emit parsing results for every incoming parsing request in order of input" in new GraphWithDummyWorkers(1) {
+      "emit parsing results for every incoming parsing request in order of input" in new Fixture(1) {
         source.sendNext(FileParsingTaskStartTree(sampleTreeScan))
         source.sendNext(input1).sendNext(input2).sendNext(input3)
         sink.ensureSubscription().request(4)
@@ -112,7 +103,7 @@ class CalculationParsingFlowSpec extends WordSpec with MockitoSugar with TestDat
         }
       }
 
-      "forward a file tree finished signal after processing it" in new GraphWithDummyWorkers(1) {
+      "forward a file tree finished signal after processing it" in new Fixture(1) {
         source.sendNext(FileParsingTaskStartTree(sampleTreeScan))
         source.sendNext(FileParsingTaskEndTree(sampleTreeScan))
         sink.ensureSubscription().request(2)
@@ -120,7 +111,7 @@ class CalculationParsingFlowSpec extends WordSpec with MockitoSugar with TestDat
         sink.expectNext(FileParsingResultEndTree(sampleTreeScan))
       }
 
-      "reject parsing requests after receiving the tree finished signal" in new GraphWithDummyWorkers(1) {
+      "reject parsing requests after receiving the tree finished signal" in new Fixture(1) {
         source.sendNext(FileParsingTaskStartTree(sampleTreeScan))
         source.sendNext(FileParsingTaskEndTree(sampleTreeScan)).sendNext(input1)
         sink.ensureSubscription().request(3)
@@ -128,7 +119,7 @@ class CalculationParsingFlowSpec extends WordSpec with MockitoSugar with TestDat
         sink.expectError()
       }
 
-      "complete the stage if upstream signals completion before all elements are processed" in new GraphWithDummyWorkers(1) {
+      "complete the stage if upstream signals completion before all elements are processed" in new Fixture(1) {
         source.sendNext(FileParsingTaskStartTree(sampleTreeScan))
         source.sendNext(input1).sendNext(input2).sendNext(input3).sendComplete()
         sink.ensureSubscription().request(4)
@@ -136,7 +127,7 @@ class CalculationParsingFlowSpec extends WordSpec with MockitoSugar with TestDat
         sink.expectComplete()
       }
 
-      "complete the stage if upstream signals completion after all elements are processed" in new GraphWithDummyWorkers(1) {
+      "complete the stage if upstream signals completion after all elements are processed" in new Fixture(1) {
         source.sendNext(FileParsingTaskStartTree(sampleTreeScan))
         source.sendNext(input1).sendNext(input2).sendNext(input3)
         sink.ensureSubscription().request(4)
@@ -145,14 +136,14 @@ class CalculationParsingFlowSpec extends WordSpec with MockitoSugar with TestDat
         sink.expectComplete()
       }
 
-      "unpack a zip archive when the start tree signal arrives" in new GraphWithDummyWorkers(1) {
+      "unpack a zip archive when the start tree signal arrives" in new Fixture(1) {
         source.sendNext(FileParsingTaskStartTree(sampleTreeScan.copy(treeType = TreeType.Zip)))
         sink.ensureSubscription().request(1)
         sink.expectNext()
         verify(dummyArchiveHandler).extractZipArchive(sampleTreeScan.treeBasePath)
       }
 
-      "delete the extracted files from a zip archive when the end tree signal arrives" in new GraphWithDummyWorkers(1) {
+      "delete the extracted files from a zip archive when the end tree signal arrives" in new Fixture(1) {
         source.sendNext(FileParsingTaskStartTree(sampleTreeScan.copy(treeType = TreeType.Zip)))
         source.sendNext(FileParsingTaskEndTree(sampleTreeScan.copy(treeType = TreeType.Zip)))
         sink.ensureSubscription().request(2)
@@ -160,7 +151,7 @@ class CalculationParsingFlowSpec extends WordSpec with MockitoSugar with TestDat
         verify(dummyArchiveHandler).cleanUpExtractedArchive(sampleTreeScan.treeBasePath)
       }
 
-      "not interact with the ArchiveHandler when processing directory file trees" in new GraphWithDummyWorkers(1) {
+      "not interact with the ArchiveHandler when processing directory file trees" in new Fixture(1) {
         source.sendNext(FileParsingTaskStartTree(sampleTreeScan))
         source.sendNext(FileParsingTaskEndTree(sampleTreeScan))
         sink.ensureSubscription().request(2)
@@ -168,7 +159,7 @@ class CalculationParsingFlowSpec extends WordSpec with MockitoSugar with TestDat
         verifyZeroInteractions(dummyArchiveHandler)
       }
 
-      "specify the location of the main file to the parsing engine when handling directories" in new GraphWithDummyWorkers(1) {
+      "specify the location of the main file to the parsing engine when handling directories" in new Fixture(1) {
         source.sendNext(FileParsingTaskStartTree(sampleTreeScan))
         source.sendNext(input1)
         sink.ensureSubscription().request(2).expectNextN(2)
@@ -176,10 +167,9 @@ class CalculationParsingFlowSpec extends WordSpec with MockitoSugar with TestDat
         verify(dummyWorkers.head).processRequest(any(), ArgumentMatchers.eq(expectedPath))
       }
 
-      "specify the temporarily extracted main file to the parsing engine when handling zip archives" in new GraphWithDummyWorkers(1) {
+      "specify the temporarily extracted main file to the parsing engine when handling zip archives" in new Fixture(1) {
         val zipTreeTask = sampleTreeScan.copy(treeType = TreeType.Zip)
         val extractedPath = Paths.get("/tmp/extracted/magic")
-        //        val task = aFileParsingTask().withExtractedPath(Some(extractedPath)).build()
         when(dummyArchiveHandler.extractZipArchive(any())).thenReturn(extractedPath)
         source.sendNext(FileParsingTaskStartTree(zipTreeTask))
         source.sendNext(input1.copy(treeTask = zipTreeTask))
@@ -188,7 +178,7 @@ class CalculationParsingFlowSpec extends WordSpec with MockitoSugar with TestDat
         verify(dummyWorkers.head).processRequest(any(), ArgumentMatchers.eq(expectedPath))
       }
 
-      "gracefully fail parsing requests with unknown or not supported file tree types" in new GraphWithDummyWorkers(1) {
+      "gracefully fail parsing requests with unknown or not supported file tree types" in new Fixture(1) {
         sink.ensureSubscription().request(6)
         Seq(TreeType.Unknown, TreeType.File, TreeType.Tar).foreach { treeType =>
           val anotherTreeScan = sampleTreeScan.copy(treeType = treeType)
@@ -203,8 +193,8 @@ class CalculationParsingFlowSpec extends WordSpec with MockitoSugar with TestDat
         }
       }
 
-      "handle parsing requests with supported file tree types" in new GraphWithDummyWorkers(1) {
-        when(dummyArchiveHandler.extractZipArchive(any())).thenReturn(unusedPath)
+      "handle parsing requests with supported file tree types" in new Fixture(1) {
+        when(dummyArchiveHandler.extractZipArchive(any())).thenReturn(Paths.get("/tmp/extracted/magic"))
         sink.ensureSubscription().request(4)
         Seq(TreeType.Directory, TreeType.Zip).foreach { treeType =>
           val anotherTreeScan = sampleTreeScan.copy(treeType = treeType)