use CSVPrinter to generate DRS localizer manifest (and add a test)

Author: breilly2

supportedBackends/google/pipelines/v2beta/src/main/scala/cromwell/backend/google/pipelines/v2beta/PipelinesApiAsyncBackendJobExecutionActor.scala

@@ -16,11 +16,15 @@ import cromwell.filesystems.drs.DrsPath
 import cromwell.filesystems.gcs.GcsPathBuilder.ValidFullGcsPath
 import cromwell.filesystems.gcs.{GcsPath, GcsPathBuilder}
 import org.apache.commons.codec.digest.DigestUtils
+import org.apache.commons.csv.{CSVFormat, CSVPrinter}
+import org.apache.commons.io.output.ByteArrayOutputStream
 import wom.core.FullyQualifiedName
 import wom.expression.FileEvaluation
 import wom.values.{GlobFunctions, WomFile, WomGlobFile, WomMaybeListedDirectory, WomMaybePopulatedFile, WomSingleFile, WomUnlistedDirectory}
 
-import java.io.FileNotFoundException
+import java.nio.charset.Charset
+
+import java.io.{FileNotFoundException, OutputStreamWriter}
 import scala.concurrent.Future
 import scala.io.Source
 import scala.language.postfixOps
@@ -175,16 +179,6 @@ class PipelinesApiAsyncBackendJobExecutionActor(standardParams: StandardAsyncExe
 
   import mouse.all._
 
-  private def generateDrsLocalizerManifest(inputs: List[PipelinesApiInput]): String = {
-    inputs.collect {
-      case PipelinesApiFileInput(name, drsPath: DrsPath, relativeHostPath, mount) => {
-        val drsPathStr = drsPath.pathAsString
-        val containerPathStr = PipelinesApiFileInput(name, drsPath, relativeHostPath, mount).containerPath.pathAsString
-        s"\"$drsPathStr\",\"$containerPathStr\""
-      }
-    }.mkString("\n")
-  }
-
   override def uploadDrsLocalizationManifest(createPipelineParameters: CreatePipelineParameters, cloudPath: Path): Future[Unit] = {
     val content = generateDrsLocalizerManifest(createPipelineParameters.inputOutputParameters.fileInputParameters)
     asyncIo.writeAsync(cloudPath, content, Seq(CloudStorageOptions.withMimeType("text/plain")))
@@ -412,4 +406,17 @@ object PipelinesApiAsyncBackendJobExecutionActor {
       }
     } combineAll
   }
+
+  private [v2beta] def generateDrsLocalizerManifest(inputs: List[PipelinesApiInput]): String = {
+    val outputStream = new ByteArrayOutputStream()
+    val csvPrinter = new CSVPrinter(new OutputStreamWriter(outputStream), CSVFormat.DEFAULT)
+    val drsFileInputs = inputs collect {
+      case drsInput@PipelinesApiFileInput(_, drsPath: DrsPath, _, _) => (drsInput, drsPath)
+    }
+    drsFileInputs foreach { case (drsInput, drsPath) =>
+      csvPrinter.printRecord(drsPath.pathAsString, drsInput.containerPath.pathAsString)
+    }
+    csvPrinter.close(true)
+    outputStream.toString(Charset.defaultCharset())
+  }
 }

supportedBackends/google/pipelines/v2beta/src/test/scala/cromwell/backend/google/pipelines/v2beta/PipelinesApiAsyncBackendJobExecutionActorSpec.scala

@@ -2,14 +2,22 @@ package cromwell.backend.google.pipelines.v2beta
 
 import java.nio.file.Paths
 import cats.data.NonEmptyList
+import cloud.nio.impl.drs.DrsCloudNioFileProvider.DrsReadInterpreter
+import cloud.nio.impl.drs.{DrsCloudNioFileSystemProvider, GoogleDrsCredentials}
+import com.google.cloud.NoCredentials
+import com.typesafe.config.{Config, ConfigFactory}
 import common.assertion.CromwellTimeoutSpec
 import common.mock.MockSugar
 import cromwell.backend.google.pipelines.common.PipelinesApiFileInput
+import cromwell.backend.google.pipelines.common.io.{DiskType, PipelinesApiWorkingDisk}
 import cromwell.core.path.DefaultPathBuilder
+import cromwell.filesystems.drs.DrsPathBuilder
 import org.mockito.Mockito._
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 
+import scala.concurrent.duration.DurationInt
+
 class PipelinesApiAsyncBackendJobExecutionActorSpec extends AnyFlatSpec with CromwellTimeoutSpec with Matchers
   with MockSugar {
   behavior of "PipelinesParameterConversions"
@@ -44,4 +52,45 @@ class PipelinesApiAsyncBackendJobExecutionActorSpec extends AnyFlatSpec with Cro
 
     PipelinesApiAsyncBackendJobExecutionActor.groupParametersByGcsBucket(inputs) shouldEqual expected
   }
+
+  it should "generate a CSV manifest for DRS inputs, ignoring non-DRS inputs" in {
+    def makeDrsPathBuilder: DrsPathBuilder = {
+      val marthaConfig: Config = ConfigFactory.parseString(
+        """martha {
+          |   url = "http://martha-url"
+          |}
+          |""".stripMargin
+      )
+
+      val fakeCredentials = NoCredentials.getInstance
+
+      val drsReadInterpreter: DrsReadInterpreter = (_, _) =>
+        throw new UnsupportedOperationException("PipelinesApiAsyncBackendJobExecutionActorSpec doesn't need to use drs read interpreter.")
+
+      DrsPathBuilder(
+        new DrsCloudNioFileSystemProvider(marthaConfig, GoogleDrsCredentials(fakeCredentials, 1.minutes), drsReadInterpreter),
+        None,
+      )
+    }
+
+    val mount = PipelinesApiWorkingDisk(DiskType.LOCAL, 1)
+
+    def makeDrsInput(name: String, drsUri: String, containerPath: String): PipelinesApiFileInput = {
+      val drsPath = makeDrsPathBuilder.build(drsUri).get
+      val containerRelativePath = DefaultPathBuilder.get(containerPath)
+      PipelinesApiFileInput(name, drsPath, containerRelativePath, mount)
+    }
+
+    val nonDrsInput: PipelinesApiFileInput = PipelinesApiFileInput("nnn",
+      DefaultPathBuilder.get("/local/nnn.bai"), DefaultPathBuilder.get("/path/to/nnn.bai"), mount)
+
+    val inputs = List(
+      makeDrsInput("aaa", "drs://drs.example.org/aaa", "path/to/aaa.bai"),
+      nonDrsInput,
+      makeDrsInput("bbb", "drs://drs.example.org/bbb", "path/to/bbb.bai")
+    )
+
+    PipelinesApiAsyncBackendJobExecutionActor.generateDrsLocalizerManifest(inputs) shouldEqual
+      "drs://drs.example.org/aaa,/cromwell_root/path/to/aaa.bai\r\ndrs://drs.example.org/bbb,/cromwell_root/path/to/bbb.bai\r\n"
+  }
 }