<commit_before>package scray.querying.sync

import java.util.concurrent.atomic.AtomicInteger

import scala.annotation.tailrec
import scala.util.Failure
import scala.util.Success
import scala.util.Try

import org.cassandraunit.utils.EmbeddedCassandraServerHelper
import org.junit.runner.RunWith
import org.scalatest.BeforeAndAfter
import org.scalatest.BeforeAndAfterAll
import org.scalatest.WordSpec
import org.scalatest.junit.JUnitRunner

import com.datastax.driver.core.Cluster
import com.datastax.driver.core.ResultSet
import com.datastax.driver.core.Statement
import com.datastax.driver.core.querybuilder.Insert

import scray.common.serialization.BatchID
import shapeless.ops.hlist._
import shapeless.syntax.singleton._
import scray.cassandra.sync.CassandraJobInfo
import java.util.concurrent.TimeUnit
import scray.querying.sync.ColumnWithValue
import scray.querying.sync.State
import scray.querying.sync.helpers.TestDbSession
import scray.cassandra.sync.OnlineBatchSyncCassandra
import scray.querying.sync.ArbitrarylyTypedRows
import scray.querying.sync.Column
import scray.querying.sync.RowWithValue
import scray.cassandra.sync.CassandraImplementation._

@RunWith(classOf[JUnitRunner])
class StartStopTest extends WordSpec {
  var dbconnection: TestDbSession = new TestDbSession
  var jobNr: AtomicInteger = new AtomicInteger(0)

  def getNextJobName: String = {
    "StartStopTest_Job" + jobNr.getAndIncrement
  }

  /**
   * Test columns
   */
  class SumTestColumns() extends ArbitrarylyTypedRows {
    val sum = new Column[Long]("sum")

    override val columns = sum :: Nil
    override val primaryKey = s"(${sum.name})"
    override val indexes: Option[List[String]] = None
  }
  val batchId = new BatchID(1L, 1L)

  "StartStopTest " should {
    " init client" in {
      val table = new OnlineBatchSyncCassandra("johannes")
      val jobInfo = new CassandraJobInfo(getNextJobName)
      assert(table.initJob[SumTestColumns](jobInfo, new SumTestColumns).isSuccess)
    }
    "get running batch/online version " in {
      val table = new OnlineBatchSyncCassandra(dbconnection)

      val jobInfo = new CassandraJobInfo(getNextJobName)
      val sum = new ColumnWithValue[Long]("sum", 100)
      val columns = sum :: Nil
      val primaryKey = s"(${sum.name})"
      val indexes: Option[List[String]] = None

      table.initJob(jobInfo, new RowWithValue(columns, primaryKey, indexes))
      assert(table.startNextBatchJob(jobInfo).isSuccess)
      assert(table.completeBatchJob(jobInfo).isSuccess)

      assert(table.startNextOnlineJob(jobInfo).isSuccess)
      assert(table.startNextBatchJob(jobInfo).isSuccess)
      
      assert(table.getRunningBatchJobSlot(jobInfo).get == 2)<commit_msg>Fix db connection in test<commit_after>package scray.querying.sync

import java.util.concurrent.atomic.AtomicInteger

import scala.annotation.tailrec
import scala.util.Failure
import scala.util.Success
import scala.util.Try

import org.cassandraunit.utils.EmbeddedCassandraServerHelper
import org.junit.runner.RunWith
import org.scalatest.BeforeAndAfter
import org.scalatest.BeforeAndAfterAll
import org.scalatest.WordSpec
import org.scalatest.junit.JUnitRunner

import com.datastax.driver.core.Cluster
import com.datastax.driver.core.ResultSet
import com.datastax.driver.core.Statement
import com.datastax.driver.core.querybuilder.Insert

import scray.common.serialization.BatchID
import shapeless.ops.hlist._
import shapeless.syntax.singleton._
import scray.cassandra.sync.CassandraJobInfo
import java.util.concurrent.TimeUnit
import scray.querying.sync.ColumnWithValue
import scray.querying.sync.State
import scray.querying.sync.helpers.TestDbSession
import scray.cassandra.sync.OnlineBatchSyncCassandra
import scray.querying.sync.ArbitrarylyTypedRows
import scray.querying.sync.Column
import scray.querying.sync.RowWithValue
import scray.cassandra.sync.CassandraImplementation._

@RunWith(classOf[JUnitRunner])
class StartStopTest extends WordSpec {
  var dbconnection: TestDbSession = new TestDbSession
  var jobNr: AtomicInteger = new AtomicInteger(0)

  def getNextJobName: String = {
    "StartStopTest_Job" + jobNr.getAndIncrement
  }

  /**
   * Test columns
   */
  class SumTestColumns() extends ArbitrarylyTypedRows {
    val sum = new Column[Long]("sum")

    override val columns = sum :: Nil
    override val primaryKey = s"(${sum.name})"
    override val indexes: Option[List[String]] = None
  }
  val batchId = new BatchID(1L, 1L)

  "StartStopTest " should {
    " init client" in {
      val table = new OnlineBatchSyncCassandra(dbconnection)
      val jobInfo = new CassandraJobInfo(getNextJobName)
      assert(table.initJob[SumTestColumns](jobInfo, new SumTestColumns).isSuccess)
    }
    "get running batch/online version " in {
      val table = new OnlineBatchSyncCassandra(dbconnection)

      val jobInfo = new CassandraJobInfo(getNextJobName)
      val sum = new ColumnWithValue[Long]("sum", 100)
      val columns = sum :: Nil
      val primaryKey = s"(${sum.name})"
      val indexes: Option[List[String]] = None

      table.initJob(jobInfo, new RowWithValue(columns, primaryKey, indexes))
      assert(table.startNextBatchJob(jobInfo).isSuccess)
      assert(table.completeBatchJob(jobInfo).isSuccess)

      assert(table.startNextOnlineJob(jobInfo).isSuccess)
      assert(table.startNextBatchJob(jobInfo).isSuccess)
      
      assert(table.getRunningBatchJobSlot(jobInfo).get == 2)
{'fraction_non_alphanumeric': 0.09144542772861357, 'fraction_numerical': 0.0025811209439528023, 'mean_word_length': 4.752916224814422, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, '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': '4045873', 'n_tokens_mistral': 1727, 'n_tokens_neox': 1671, 'n_words': 329}