Integrated Pipeline: further refactored tests for ParsingResultsProcessingManager

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

 package eu.nomad_lab.integrated_pipeline_tests
 
-import java.nio.file.Paths
-
 import eu.nomad_lab.integrated_pipeline.messages._
 import eu.nomad_lab.integrated_pipeline.{ OutputType, ParsingResultsProcessingManager, ParsingResultsProcessor }
 import org.mockito.ArgumentMatchers.any
@@ -51,48 +49,43 @@ class ParsingResultsProcessingManagerSpec extends WordSpec with MockitoSugar wit
     f
   }
 
-  "a ParsingResultsProcessorFlow" when {
-
-    "not having received any events" should {
-      behave like processorWithNoSignalReady(
-        () => noResultsNoEndReceived().manager,
-        Seq(anInMemoryResult(), aFileParsingSignalEndTree())
-      )
-
-      "start processing a file tree when the first signal associated with it arrives (result arrives first)" in {
+  def fullyProcessedATree(): Fixture = {
     val f = new Fixture
     f.manager.processSignal(anInMemoryResult())
-        verify(f.processingMock).startProcessingTreeResults(any())
     f.manager.processSignal(anInMemoryResult())
-        verify(f.processingMock).startProcessingTreeResults(any())
+    f.manager.processSignal(aFileParsingSignalEndTree().withTaskCount(2))
+    f.manager.getNextSignalToEmit()
+    f
   }
 
-      "start processing a file tree when the first signal associated with it arrives (end signal arrives first)" in {
-        val f = new Fixture
-        f.manager.processSignal(aFileParsingSignalEndTree())
-        verify(f.processingMock).startProcessingTreeResults(any())
-        f.manager.processSignal(anInMemoryResult())
-        verify(f.processingMock).startProcessingTreeResults(any())
-      }
+  "a ParsingResultsProcessorFlow" when {
 
-      "process every incoming file parsing result" in {
-        val f = new Fixture
-        f.manager.processSignal(anInMemoryResult().withRelativePath("foo"))
-        f.manager.processSignal(anInMemoryResult().withRelativePath("magic"))
-        val argument: ArgumentCaptor[FileParsingResult] = ArgumentCaptor.forClass(classOf[FileParsingResult])
-        verify(f.processingMock, times(2)).processFileParsingResult(argument.capture())
-        val results = argument.getAllValues.asScala.map(x => x.task.relativePath)
-        results should contain allOf (Paths.get("foo"), Paths.get("magic"))
-      }
+    "not having received any events" should {
+      val createFixture = () => noResultsNoEndReceived()
 
+      behave like processorWithNoSignalReady(
+        () => createFixture().manager,
+        Seq(anInMemoryResult(), aFileParsingSignalEndTree())
+      )
+
+      behave like readyToAcceptParsingResults(createFixture)
+      behave like readyToAcceptParsingResultsFromAnotherTree(createFixture)
+      behave like interleaveAnotherFileTree(createFixture)
+      behave like startProcessingANewTree(createFixture)
     }
 
     "having received all parsing tasks but no end signal" should {
+      val createFixture = () => twoResultsNoEndReceived()
+
       behave like processorWithNoSignalReady(
-        () => twoResultsNoEndReceived().manager,
+        () => createFixture().manager,
         Seq(anInMemoryResult(), aFileParsingSignalEndTree())
       )
 
+      behave like readyToAcceptParsingResultsFromAnotherTree(createFixture)
+      behave like interleaveAnotherFileTree(createFixture)
+      behave like startProcessingANewTree(createFixture)
+
       "finish processing a file tree when the associated end signal arrives" in {
         val f = twoResultsNoEndReceived()
         f.manager.hasSignalToEmit() should be(false)
@@ -104,11 +97,18 @@ class ParsingResultsProcessingManagerSpec extends WordSpec with MockitoSugar wit
     }
 
     "having received the end signal and all but one parsing results" should {
+      val createFixture = () => oneResultEndWithTwoEntriesReceived()
+
       behave like processorWithNoSignalReady(
-        () => oneResultEndWithTwoEntriesReceived().manager,
+        () => createFixture().manager,
         Seq(anInMemoryResult(), aFileParsingSignalEndTree())
       )
 
+      behave like readyToAcceptParsingResults(createFixture)
+      behave like readyToAcceptParsingResultsFromAnotherTree(createFixture)
+      behave like interleaveAnotherFileTree(createFixture)
+      behave like startProcessingANewTree(createFixture)
+
       "finish processing a file tree when the last associated parsing result arrived" in {
         val f = oneResultEndWithTwoEntriesReceived()
         f.manager.hasSignalToEmit() should be(false)
@@ -135,47 +135,85 @@ class ParsingResultsProcessingManagerSpec extends WordSpec with MockitoSugar wit
         f.orderingTest.verify(f.processingMock).finishProcessingTreeResults(aFileTreeScanTask())
       }
 
+    }
+
+    "having fully processed a file tree and emitted the result" should {
+      val createFixture = () => fullyProcessedATree()
+
+      behave like processorWithNoSignalReady(
+        () => createFixture().manager,
+        Seq(anInMemoryResult(), aFileParsingSignalEndTree())
+      )
+
+      behave like readyToAcceptParsingResults(createFixture)
+      behave like readyToAcceptParsingResultsFromAnotherTree(createFixture)
+      behave like interleaveAnotherFileTree(createFixture)
+      behave like startProcessingANewTree(createFixture)
+
       "be able to process the same file tree again after finishing it once" in {
-        val f = twoResultsEndWithTwoEntriesReceived()
-        f.manager.getNextSignalToEmit()
+        val f = fullyProcessedATree()
         f.manager.processSignal(anInMemoryResult())
         f.manager.processSignal(anInMemoryResult())
         f.manager.processSignal(aFileParsingSignalEndTree().withTaskCount(2))
         verify(f.processingMock, times(2)).finishProcessingTreeResults(aFileTreeScanTask())
       }
     }
+  }
 
-    "handling multiple, possibly interleaved, tree results at the same time" should {
-      "process all file parsing results, regardless of their inbound ordering" in {
-        val f = new Fixture
-        val treeTask1 = aFileTreeScanTask().withBasePath("/path1")
-        val treeTask2 = aFileTreeScanTask().withBasePath("/path2")
-        val inputs1 = (1 to 3).map(x => anInMemoryResult().withTreeTask(treeTask1).withRelativePath(s"foo$x").build())
-        val inputs2 = (1 to 3).map(x => anInMemoryResult().withTreeTask(treeTask2).withRelativePath(s"bar$x").build())
-        val chaos = Seq(inputs1(0), inputs1(1), inputs2(0), inputs2(1), inputs1(2), inputs2(2))
-        chaos.foreach(f.manager.processSignal)
-        chaos.foreach(x => verify(f.processingMock, times(1)).processFileParsingResult(x))
+  def readyToAcceptParsingResults(createFixture: () => Fixture): Unit = {
+    "process a file parsing result" in {
+      import FileParsingResultMatchers._
+      val f = createFixture()
+      f.manager.processSignal(anInMemoryResult().withRelativePath("foo"))
+      val argument: ArgumentCaptor[FileParsingResult] = ArgumentCaptor.forClass(classOf[FileParsingResult])
+      verify(f.processingMock, atLeastOnce()).processFileParsingResult(argument.capture())
+      val results = argument.getAllValues.asScala
+      exactly(1, results) should have(relativePath("foo"))
+    }
   }
 
-      "send a result for each tree once all associated file parsing results are processed" in {
+  def readyToAcceptParsingResultsFromAnotherTree(createFixture: () => Fixture): Unit = {
+    "process a file parsing result from another file tree" in {
+      import FileParsingResultMatchers._
+      val f = createFixture()
+      val fileTree = aFileTreeScanTask().withBasePath("/universe/magic")
+      f.manager.processSignal(anInMemoryResult().withTreeTask(fileTree))
+      val argument: ArgumentCaptor[FileParsingResult] = ArgumentCaptor.forClass(classOf[FileParsingResult])
+      verify(f.processingMock, atLeastOnce()).processFileParsingResult(argument.capture())
+      val results = argument.getAllValues.asScala
+      exactly(1, results) should have(treeTask(fileTree))
+    }
+  }
+
+  def interleaveAnotherFileTree(createFixture: () => Fixture): Unit = {
+    "interleave requests from another file tree" in {
       import FileTreeParsingResultMatchers._
-        val f = new Fixture
-        val treeTask1 = aFileTreeScanTask().withBasePath("/path1")
-        val treeTask2 = aFileTreeScanTask().withBasePath("/path2")
-        val inputs1 = (1 to 3).map(x => anInMemoryResult().withTreeTask(treeTask1).withRelativePath(s"foo$x").build())
-        val inputs2 = (1 to 2).map(x => anInMemoryResult().withTreeTask(treeTask2).withRelativePath(s"bar$x").build())
-        val chaos = Seq(inputs1(0), inputs1(1), inputs2(0), inputs1(2), inputs2(1))
-        chaos.foreach(f.manager.processSignal)
-        chaos.foreach(x => verify(f.processingMock, times(1)).processFileParsingResult(x))
-        f.manager.processSignal(aFileParsingSignalEndTree().withTreeTask(treeTask1).withTaskCount(3))
-        f.manager.hasSignalToEmit() should be(true)
-        f.manager.getNextSignalToEmit() should have(treeTask(treeTask1), numParsingTasks(3))
-        f.manager.hasSignalToEmit() should be(false)
-        f.manager.processSignal(aFileParsingSignalEndTree().withTreeTask(treeTask2).withTaskCount(2))
+      val f = createFixture()
+      val fileTree = aFileTreeScanTask().withBasePath("/universe/magic")
+      f.manager.processSignal(anInMemoryResult().withTreeTask(fileTree))
+      f.manager.processSignal(aFileParsingSignalEndTree().withTreeTask(fileTree).withTaskCount(1))
       f.manager.hasSignalToEmit() should be(true)
-        f.manager.getNextSignalToEmit() should have(treeTask(treeTask2), numParsingTasks(2))
-        f.manager.hasSignalToEmit() should be(false)
+      f.manager.getNextSignalToEmit() should have(treeTask(fileTree), numParsingTasks(1))
+    }
   }
+
+  def startProcessingANewTree(createFixture: () => Fixture): Unit = {
+    "start processing a file tree when the first signal associated with it arrives (result arrives first)" in {
+      val f = createFixture()
+      val fileTree = aFileTreeScanTask().withBasePath("/tmp/foo/random")
+      f.manager.processSignal(anInMemoryResult().withTreeTask(fileTree))
+      verify(f.processingMock, times(1)).startProcessingTreeResults(fileTree)
+      f.manager.processSignal(anInMemoryResult().withTreeTask(fileTree))
+      verify(f.processingMock, times(1)).startProcessingTreeResults(fileTree)
+    }
+
+    "start processing a file tree when the first signal associated with it arrives (end signal arrives first)" in {
+      val f = createFixture()
+      val fileTree = aFileTreeScanTask().withBasePath("/tmp/foo/random")
+      f.manager.processSignal(aFileParsingSignalEndTree().withTreeTask(fileTree))
+      verify(f.processingMock, times(1)).startProcessingTreeResults(fileTree)
+      f.manager.processSignal(anInMemoryResult().withTreeTask(fileTree))
+      verify(f.processingMock, times(1)).startProcessingTreeResults(fileTree)
     }
   }