<commit_before>package cromwell.engine.db.slick

import java.nio.file.{FileSystem, FileSystems}
import java.sql.SQLException
import java.time.OffsetDateTime

import better.files._
import com.google.api.client.util.ExponentialBackOff
import com.typesafe.config.ConfigFactory
import cromwell.CromwellSpec.DbmsTest
import cromwell.CromwellTestkitSpec.TestWorkflowManagerSystem
import cromwell.backend.wdl.{OldCallEngineFunctions, OldWorkflowEngineFunctions}
import cromwell.backend.{ExecutionEventEntry, JobKey}
import cromwell.core._
import cromwell.database.SqlConverters._
import cromwell.database.obj.{Execution, WorkflowMetadataKeys}
import cromwell.database.slick.SlickDatabase
import cromwell.engine._
import cromwell.engine.backend._
import cromwell.engine.backend.io._
import cromwell.engine.backend.local.OldStyleLocalBackend
import cromwell.engine.backend.local.OldStyleLocalBackend.InfoKeys
import cromwell.engine.db.{DataAccess, ExecutionDatabaseKey}
import cromwell.engine.workflow.{BackendCallKey, ScatterKey}
import cromwell.services.{MetadataEvent, MetadataKey, MetadataValue}
import cromwell.util.SampleWdl
import cromwell.webservice.{CallCachingParameters, WorkflowQueryKey, WorkflowQueryParameters}
import cromwell.{CromwellTestkitSpec, webservice}
import org.scalactic.StringNormalizations._
import org.scalatest.PartialFunctionValues._
import org.scalatest.concurrent.PatienceConfiguration.Timeout
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.prop.TableDrivenPropertyChecks
import org.scalatest.time.SpanSugar._
import org.scalatest.time.{Millis, Seconds, Span}
import org.scalatest.{BeforeAndAfterAll, FlatSpec, Matchers}
import org.specs2.mock.Mockito
import wdl4s._
import wdl4s.types.{WdlArrayType, WdlStringType}
import wdl4s.values.{WdlFile, _}

import scala.concurrent.{ExecutionContext, Future}
import scala.util.Try

object SlickDataAccessSpec {
  val AllowFalse = Seq(webservice.QueryParameter("allow", "false"))
  val AllowTrue = Seq(webservice.QueryParameter("allow", "true"))

  val Workflow1Name = "test1"
  val Workflow2Name = "test2"

  implicit class EnhancedEngineWorkflowDescriptor(val workflowDescriptor: EngineWorkflowDescriptor) extends AnyVal {
    def fileHasher: FileHasher = { wdlFile: WdlFile =>
        SymbolHash(wdlFile.value.toPath(FileSystems.getDefault).hash)
    }
  }
}

class SlickDataAccessSpec extends FlatSpec with Matchers with ScalaFutures with BeforeAndAfterAll with Mockito
  with WorkflowDescriptorBuilder {
  import SlickDataAccessSpec._

  behavior of "SlickDataAccess"

  import TableDrivenPropertyChecks._

  val workflowManagerSystem = new TestWorkflowManagerSystem
  override implicit val actorSystem = workflowManagerSystem.actorSystem

  override protected def afterAll() = {
    workflowManagerSystem.shutdownTestActorSystem()
    super.afterAll()
  }

  implicit val ec = ExecutionContext.global

  implicit val defaultPatience = PatienceConfig(timeout = Span(5, Seconds), interval = Span(100, Millis))

  lazy val localBackend = OldStyleLocalBackend(CromwellTestkitSpec.DefaultLocalBackendConfigEntry, workflowManagerSystem.actorSystem)

  val test1Sources = WorkflowSourceFiles("workflow test1 {}", "{}", "{}")
  val test2Sources = WorkflowSourceFiles("workflow test2 {}", "{}", "{}")

  object UnknownOldStyleBackend$ extends OldStyleBackend {
    def engineFunctions(fileSystems: List[FileSystem], workflowContext: OldWorkflowContext): OldWorkflowEngineFunctions = throw new NotImplementedError
    override val actorSystem = workflowManagerSystem.actorSystem
    override val backendConfigEntry = CromwellTestkitSpec.DefaultLocalBackendConfigEntry
    override def adjustInputPaths(backendCallJobDescriptor: OldStyleBackendCallJobDescriptor) = throw new NotImplementedError()
    override def stdoutStderr(jobDescriptor: OldStyleBackendCallJobDescriptor): CallLogs = throw new NotImplementedError
    override def initializeForWorkflow(workflow: OldStyleWorkflowDescriptor) = throw new NotImplementedError
    override def backendType: BackendType = throw new NotImplementedError
    override def rootPath(workflowOptions: WorkflowOptions): String = throw new NotImplementedError
    override def pollBackoff: ExponentialBackOff = throw new NotImplementedError
    override def executionInfoKeys: List[String] = List.empty
    override def instantiateCommand(descriptor: OldStyleBackendCallJobDescriptor): Try[String] = throw new NotImplementedError
    override def poll(jobDescriptor: OldStyleBackendCallJobDescriptor, previous: OldStyleExecutionHandle)(implicit ec: ExecutionContext): Future[OldStyleExecutionHandle] = throw new NotImplementedError()
    override def callEngineFunctions(descriptor: OldStyleBackendCallJobDescriptor): OldCallEngineFunctions = throw new NotImplementedError()
    override def useCachedCall(cachedCall: OldStyleBackendCallJobDescriptor, backendCall: OldStyleBackendCallJobDescriptor)(implicit ec: ExecutionContext): Future[OldStyleExecutionHandle] = throw new NotImplementedError()
    override def execute(jobDescriptor: OldStyleBackendCallJobDescriptor)(implicit ec: ExecutionContext): Future[OldStyleExecutionHandle] = throw new NotImplementedError()
    override def resume(descriptor: OldStyleBackendCallJobDescriptor, executionInfos: Map[String, Option[String]])(implicit ec: ExecutionContext): Future[OldStyleExecutionHandle] = throw new NotImplementedError()
    override def fileSystems(options: WorkflowOptions): List[FileSystem] = List(FileSystems.getDefault)
    override def isResumable(key: JobKey, executionInfo: Map[String, Option[String]]) = false
    override def isRestartable(key: JobKey, executionInfo: Map[String, Option[String]]) = false
  }

  it should "not deadlock" in {
    // Test based on https://github.com/kwark/slick-deadlock/blob/82525fc/src/main/scala/SlickDeadlock.scala
    val databaseConfig = ConfigFactory.parseString(
      s"""
         |db.url = "jdbc:hsqldb:mem:$${slick.uniqueSchema};shutdown=false;hsqldb.tx=mvcc"
         |db.driver = "org.hsqldb.jdbcDriver"
         |db.numThreads = 2
         |driver = "slick.driver.HsqldbDriver$$"
         |""".stripMargin)

    for {
      dataAccess <- (new SlickDatabase(databaseConfig) with DataAccess).autoClosed
    } {
      val futures = 1 to 20 map { _ =>
        val workflowId = WorkflowId.randomId()
        val workflowInfo = createMaterializedEngineWorkflowDescriptor(id = workflowId, workflowSources = test1Sources)
        for {
          _ <- dataAccess.createWorkflow(workflowInfo, test1Sources, Nil, Nil, localBackend)
          _ <- dataAccess.getWorkflowExecutionAndAux(workflowInfo.id) map { result =>
            result.execution.workflowExecutionUuid should be(workflowId.toString)
          }
        } yield ()
      }
      Future.sequence(futures).futureValue(Timeout(10.seconds))
    }
  }

  // Test the no-args constructor
  "SlickDataAccess (main.hsqldb)" should behave like testWith(new SlickDatabase() with DataAccess)

  "SlickDataAccess (test.mysql)" should behave like
    testWith(new SlickDatabase(SlickDatabase.getDatabaseConfig("test.mysql")) with DataAccess)

  /*
  If the above tests fail, one may be accidentally talking to the singleton.
  Uncomment the following test, and see if the suite works. If so, have any code talking to services etc. talk to
  the DataAccess instead.
  */
  //"SlickDataAccess (global.singleton)" should behave like
  //  testWith(DataAccess.globalDataAccess.asInstanceOf[SlickDatabase with DataAccess])

  def testWith(dataAccess: SlickDatabase with DataAccess): Unit = {

    /**
      * Assert that the specified workflow has the appropriate number of calls in appropriately terminal states per
      * the `ExpectedTerminal` function.  This function maps from an FQN + index pair to a Boolean expectation.
      */
    type ExpectedTerminal = (FullyQualifiedName, Int) => Boolean
    def assertTerminalExecution(id: WorkflowId, expectedCount: Int, expectedTerminal: ExpectedTerminal): Future[Unit] = {
      for {
        _ <- dataAccess.getExecutions(id.toString) map { executions =>
          executions should have size expectedCount
          executions foreach { execution =>
            execution.endDt.isDefined shouldBe expectedTerminal(execution.callFqn, execution.index)
          }
        }
      } yield ()
    }

    val id = WorkflowId.randomId()
    import SlickDataAccessSpec.EnhancedEngineWorkflowDescriptor
    implicit lazy val hasher = createMaterializedEngineWorkflowDescriptor(id, test1Sources).fileHasher
    lazy val testWdlString = WdlString("testStringvalue")
    lazy val testWdlStringHash = testWdlString.computeHash
    lazy val testWdlStringShard = WdlString("testStringValueShard")
    lazy val testWdlStringShardHash = testWdlStringShard.computeHash

    it should "(if hsqldb) have transaction isolation mvcc" taggedAs DbmsTest in {
      import dataAccess.dataAccess.driver.api._
<commit_msg>Re-lazied the `SlickDataAccessSpec` database configs.<commit_after>package cromwell.engine.db.slick

import java.nio.file.{FileSystem, FileSystems}
import java.sql.SQLException
import java.time.OffsetDateTime

import better.files._
import com.google.api.client.util.ExponentialBackOff
import com.typesafe.config.ConfigFactory
import cromwell.CromwellSpec.DbmsTest
import cromwell.CromwellTestkitSpec.TestWorkflowManagerSystem
import cromwell.backend.wdl.{OldCallEngineFunctions, OldWorkflowEngineFunctions}
import cromwell.backend.{ExecutionEventEntry, JobKey}
import cromwell.core._
import cromwell.database.SqlConverters._
import cromwell.database.obj.{Execution, WorkflowMetadataKeys}
import cromwell.database.slick.SlickDatabase
import cromwell.engine._
import cromwell.engine.backend._
import cromwell.engine.backend.io._
import cromwell.engine.backend.local.OldStyleLocalBackend
import cromwell.engine.backend.local.OldStyleLocalBackend.InfoKeys
import cromwell.engine.db.{DataAccess, ExecutionDatabaseKey}
import cromwell.engine.workflow.{BackendCallKey, ScatterKey}
import cromwell.services.{MetadataEvent, MetadataKey, MetadataValue}
import cromwell.util.SampleWdl
import cromwell.webservice.{CallCachingParameters, WorkflowQueryKey, WorkflowQueryParameters}
import cromwell.{CromwellTestkitSpec, webservice}
import org.scalactic.StringNormalizations._
import org.scalatest.PartialFunctionValues._
import org.scalatest.concurrent.PatienceConfiguration.Timeout
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.prop.TableDrivenPropertyChecks
import org.scalatest.time.SpanSugar._
import org.scalatest.time.{Millis, Seconds, Span}
import org.scalatest.{BeforeAndAfterAll, FlatSpec, Matchers}
import org.specs2.mock.Mockito
import wdl4s._
import wdl4s.types.{WdlArrayType, WdlStringType}
import wdl4s.values.{WdlFile, _}

import scala.concurrent.{ExecutionContext, Future}
import scala.util.Try

object SlickDataAccessSpec {
  val AllowFalse = Seq(webservice.QueryParameter("allow", "false"))
  val AllowTrue = Seq(webservice.QueryParameter("allow", "true"))

  val Workflow1Name = "test1"
  val Workflow2Name = "test2"

  implicit class EnhancedEngineWorkflowDescriptor(val workflowDescriptor: EngineWorkflowDescriptor) extends AnyVal {
    def fileHasher: FileHasher = { wdlFile: WdlFile =>
        SymbolHash(wdlFile.value.toPath(FileSystems.getDefault).hash)
    }
  }
}

class SlickDataAccessSpec extends FlatSpec with Matchers with ScalaFutures with BeforeAndAfterAll with Mockito
  with WorkflowDescriptorBuilder {
  import SlickDataAccessSpec._

  behavior of "SlickDataAccess"

  import TableDrivenPropertyChecks._

  val workflowManagerSystem = new TestWorkflowManagerSystem
  override implicit val actorSystem = workflowManagerSystem.actorSystem

  override protected def afterAll() = {
    workflowManagerSystem.shutdownTestActorSystem()
    super.afterAll()
  }

  implicit val ec = ExecutionContext.global

  implicit val defaultPatience = PatienceConfig(timeout = Span(5, Seconds), interval = Span(100, Millis))

  lazy val localBackend = OldStyleLocalBackend(CromwellTestkitSpec.DefaultLocalBackendConfigEntry, workflowManagerSystem.actorSystem)

  val test1Sources = WorkflowSourceFiles("workflow test1 {}", "{}", "{}")
  val test2Sources = WorkflowSourceFiles("workflow test2 {}", "{}", "{}")

  object UnknownOldStyleBackend$ extends OldStyleBackend {
    def engineFunctions(fileSystems: List[FileSystem], workflowContext: OldWorkflowContext): OldWorkflowEngineFunctions = throw new NotImplementedError
    override val actorSystem = workflowManagerSystem.actorSystem
    override val backendConfigEntry = CromwellTestkitSpec.DefaultLocalBackendConfigEntry
    override def adjustInputPaths(backendCallJobDescriptor: OldStyleBackendCallJobDescriptor) = throw new NotImplementedError()
    override def stdoutStderr(jobDescriptor: OldStyleBackendCallJobDescriptor): CallLogs = throw new NotImplementedError
    override def initializeForWorkflow(workflow: OldStyleWorkflowDescriptor) = throw new NotImplementedError
    override def backendType: BackendType = throw new NotImplementedError
    override def rootPath(workflowOptions: WorkflowOptions): String = throw new NotImplementedError
    override def pollBackoff: ExponentialBackOff = throw new NotImplementedError
    override def executionInfoKeys: List[String] = List.empty
    override def instantiateCommand(descriptor: OldStyleBackendCallJobDescriptor): Try[String] = throw new NotImplementedError
    override def poll(jobDescriptor: OldStyleBackendCallJobDescriptor, previous: OldStyleExecutionHandle)(implicit ec: ExecutionContext): Future[OldStyleExecutionHandle] = throw new NotImplementedError()
    override def callEngineFunctions(descriptor: OldStyleBackendCallJobDescriptor): OldCallEngineFunctions = throw new NotImplementedError()
    override def useCachedCall(cachedCall: OldStyleBackendCallJobDescriptor, backendCall: OldStyleBackendCallJobDescriptor)(implicit ec: ExecutionContext): Future[OldStyleExecutionHandle] = throw new NotImplementedError()
    override def execute(jobDescriptor: OldStyleBackendCallJobDescriptor)(implicit ec: ExecutionContext): Future[OldStyleExecutionHandle] = throw new NotImplementedError()
    override def resume(descriptor: OldStyleBackendCallJobDescriptor, executionInfos: Map[String, Option[String]])(implicit ec: ExecutionContext): Future[OldStyleExecutionHandle] = throw new NotImplementedError()
    override def fileSystems(options: WorkflowOptions): List[FileSystem] = List(FileSystems.getDefault)
    override def isResumable(key: JobKey, executionInfo: Map[String, Option[String]]) = false
    override def isRestartable(key: JobKey, executionInfo: Map[String, Option[String]]) = false
  }

  it should "not deadlock" in {
    // Test based on https://github.com/kwark/slick-deadlock/blob/82525fc/src/main/scala/SlickDeadlock.scala
    val databaseConfig = ConfigFactory.parseString(
      s"""
         |db.url = "jdbc:hsqldb:mem:$${slick.uniqueSchema};shutdown=false;hsqldb.tx=mvcc"
         |db.driver = "org.hsqldb.jdbcDriver"
         |db.numThreads = 2
         |driver = "slick.driver.HsqldbDriver$$"
         |""".stripMargin)

    for {
      dataAccess <- (new SlickDatabase(databaseConfig) with DataAccess).autoClosed
    } {
      val futures = 1 to 20 map { _ =>
        val workflowId = WorkflowId.randomId()
        val workflowInfo = createMaterializedEngineWorkflowDescriptor(id = workflowId, workflowSources = test1Sources)
        for {
          _ <- dataAccess.createWorkflow(workflowInfo, test1Sources, Nil, Nil, localBackend)
          _ <- dataAccess.getWorkflowExecutionAndAux(workflowInfo.id) map { result =>
            result.execution.workflowExecutionUuid should be(workflowId.toString)
          }
        } yield ()
      }
      Future.sequence(futures).futureValue(Timeout(10.seconds))
    }
  }

  "SlickDataAccess (main.hsqldb)" should behave like testWith("main.hsqldb")

  "SlickDataAccess (test.mysql)" should behave like testWith("test.mysql")

  /*
  If the above tests fail, one may be accidentally talking to the singleton.
  Uncomment the following test, and see if the suite works. If so, have any code talking to services etc. talk to
  the DataAccess instead.
  */
  //"SlickDataAccess (global.singleton)" should behave like testWith("global.singleton")

  def testWith(configPath: String): Unit = {

    lazy val dataAccess: SlickDatabase with DataAccess = configPath match {
      case "global.singleton" => DataAccess.globalDataAccess.asInstanceOf[SlickDatabase with DataAccess]
      case "main.hsqldb" => new SlickDatabase() with DataAccess // Test the no-args constructor
      case _ => new SlickDatabase(SlickDatabase.getDatabaseConfig(configPath)) with DataAccess
    }

    /**
      * Assert that the specified workflow has the appropriate number of calls in appropriately terminal states per
      * the `ExpectedTerminal` function.  This function maps from an FQN + index pair to a Boolean expectation.
      */
    type ExpectedTerminal = (FullyQualifiedName, Int) => Boolean
    def assertTerminalExecution(id: WorkflowId, expectedCount: Int, expectedTerminal: ExpectedTerminal): Future[Unit] = {
      for {
        _ <- dataAccess.getExecutions(id.toString) map { executions =>
          executions should have size expectedCount
          executions foreach { execution =>
            execution.endDt.isDefined shouldBe expectedTerminal(execution.callFqn, execution.index)
          }
        }
      } yield ()
    }

    val id = WorkflowId.randomId()
    import SlickDataAccessSpec.EnhancedEngineWorkflowDescriptor
    implicit lazy val hasher = createMaterializedEngineWorkflowDescriptor(id, test1Sources).fileHasher
    lazy val testWdlString = WdlString("testStringvalue")
    lazy val testWdlStringHash = testWdlString.computeHash
    lazy val testWdlStringShard = WdlString("testStringValueShard")
    lazy val testWdlStringShardHash = testWdlStringShard.computeHash

    it should "(if hsqldb) have transaction isolation mvcc" taggedAs DbmsTest in {
      import dataAccess.dataAccess.driver.api._

{'fraction_non_alphanumeric': 0.07985822672647727, 'fraction_numerical': 0.003322811098189068, 'mean_word_length': 5.302966841186737, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 18, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6080364', 'n_tokens_mistral': 5093, 'n_tokens_neox': 5042, 'n_words': 1166}