<commit_before>package build.unstable.sonicd.service.source

import java.sql.{Connection, DriverManager, Statement}

import akka.actor.{ActorContext, ActorRef, ActorSystem, Props}
import akka.stream.actor.{ActorPublisher, ActorPublisherMessage}
import akka.testkit.{CallingThreadDispatcher, ImplicitSender, TestActorRef, TestKit}
import build.unstable.sonicd.auth.{ApiKey, ApiUser}
import build.unstable.sonicd.model.JsonProtocol._
import build.unstable.sonicd.model._
import build.unstable.sonicd.service.{Fixture, ImplicitSubscriber}
import build.unstable.sonicd.source.{JdbcConnectionsHandler, JdbcExecutor}
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
import spray.json._

class JdbcSourceSpec(_system: ActorSystem)
  extends TestKit(_system) with WordSpecLike
  with Matchers with BeforeAndAfterAll with ImplicitSender
  with ImplicitSubscriber {

  import Fixture._

  override protected def afterAll(): Unit = {
    TestKit.shutdownActorSystem(system)
    runQuery(s"DROP ALL OBJECTS")()
    testConnection.close()
  }

  def this() = this(ActorSystem("JdbcSourceSpec"))

  val controller: TestActorRef[TestController] =
    TestActorRef(Props(classOf[TestController], self).withDispatcher(CallingThreadDispatcher.Id))

  Class.forName(H2Driver)
  val testConnection = DriverManager.getConnection(H2Url, "SONICD", "")

  def runQuery(q: String)(validation: (Statement) ⇒ Unit = stmt ⇒ ()) = {
    val stmt = testConnection.createStatement()
    stmt.execute(q)
    validation(stmt)
    if (!testConnection.getAutoCommit) testConnection.commit()
    stmt.close()
  }

  def testConnectionOpen() {
    runQuery("select count(*) from information_schema.sessions;") { stmt ⇒
      val rs = stmt.getResultSet
      rs.next()
      rs.getInt(1) shouldBe 1 //test connections
    }
  }

  def newPublisher(q: String, context: RequestContext = testCtx): ActorRef = {
    val query = new Query(Some(1L), Some("traceId"), None, q, H2Config)
    val src = new JdbcSource(query, controller.underlyingActor.context, context)
    val ref = controller.underlyingActor.context.actorOf(src.handlerProps.withDispatcher(CallingThreadDispatcher.Id))
    ActorPublisher(ref).subscribe(subs)
    watch(ref)
    ref
  }

  def expectTypeMetadata() = {
    expectMsgAnyClassOf(classOf[TypeMetadata])
  }

  def expectDone(pub: ActorRef) = {
    expectMsg(DoneWithQueryExecution.success)
    expectMsg("complete") //sent by ImplicitSubscriber
    expectTerminated(pub)
  }

  "JdbcSource" should {

    "run a simple statement" in {
      val pub = newPublisher("create table users(id VARCHAR);")
      pub ! ActorPublisherMessage.Request(1)
      expectTypeMetadata()
      pub ! ActorPublisherMessage.Request(1)
      expectDone(pub)
      runQuery("show tables") { stmt ⇒
        val rs = stmt.getResultSet
        rs.next()
        rs.getString(1).toUpperCase shouldBe "USERS"
      }
      runQuery("drop table users;")()
      testConnectionOpen()
    }

    "commit changes only if user is authenticated and has write access" in {
      val pub = newPublisher("create table nowrite(id VARCHAR);",
        RequestContext("a", Some(ApiUser("", 1, ApiKey.Mode.Read, None))))
      pub ! ActorPublisherMessage.Request(1)
      expectTypeMetadata()
      pub ! ActorPublisherMessage.Request(1)
      expectDone(pub)
      runQuery("show tables") { stmt ⇒
        val rs = stmt.getResultSet
        rs.next()
        rs.getString(1).toUpperCase shouldBe "USERS"
        assert(!rs.next())
      }
      testConnectionOpen()
    }

    "run multiple statements" in {
      val pub = newPublisher(
        "create table one(id VARCHAR);" +
          "create table two(id VARCHAR);" +
          "--create table two(no VARCHAR);" +
          "create table three(id VARCHAR);" +
          "create table four(id VARCHAR);" +
          "create table five(id VARCHAR); ")
      pub ! ActorPublisherMessage.Request(1)
      expectTypeMetadata()<commit_msg>fixed jdbc commit tests<commit_after>package build.unstable.sonicd.service.source

import java.sql.{Connection, DriverManager, Statement}

import akka.actor.{ActorContext, ActorRef, ActorSystem, Props}
import akka.stream.actor.{ActorPublisher, ActorPublisherMessage}
import akka.testkit.{CallingThreadDispatcher, ImplicitSender, TestActorRef, TestKit}
import build.unstable.sonicd.auth.{ApiKey, ApiUser}
import build.unstable.sonicd.model.JsonProtocol._
import build.unstable.sonicd.model._
import build.unstable.sonicd.service.{Fixture, ImplicitSubscriber}
import build.unstable.sonicd.source.{JdbcConnectionsHandler, JdbcExecutor}
import org.scalatest.{BeforeAndAfterAll, Matchers, WordSpecLike}
import spray.json._

class JdbcSourceSpec(_system: ActorSystem)
  extends TestKit(_system) with WordSpecLike
  with Matchers with BeforeAndAfterAll with ImplicitSender
  with ImplicitSubscriber {

  import Fixture._

  override protected def afterAll(): Unit = {
    TestKit.shutdownActorSystem(system)
    runQuery(s"DROP ALL OBJECTS")()
    testConnection.close()
  }

  def this() = this(ActorSystem("JdbcSourceSpec"))

  val controller: TestActorRef[TestController] =
    TestActorRef(Props(classOf[TestController], self).withDispatcher(CallingThreadDispatcher.Id))

  Class.forName(H2Driver)
  val testConnection = DriverManager.getConnection(H2Url, "SONICD", "")

  def runQuery(q: String)(validation: (Statement) ⇒ Unit = stmt ⇒ ()) = {
    val stmt = testConnection.createStatement()
    stmt.execute(q)
    validation(stmt)
    if (!testConnection.getAutoCommit) testConnection.commit()
    stmt.close()
  }

  def testConnectionOpen() {
    runQuery("select count(*) from information_schema.sessions;") { stmt ⇒
      val rs = stmt.getResultSet
      rs.next()
      rs.getInt(1) shouldBe 1 //test connections
    }
  }

  def newPublisher(q: String, context: RequestContext = testCtx): ActorRef = {
    val query = new Query(Some(1L), Some("traceId"), None, q, H2Config)
    val src = new JdbcSource(query, controller.underlyingActor.context, context)
    val ref = controller.underlyingActor.context.actorOf(src.handlerProps.withDispatcher(CallingThreadDispatcher.Id))
    ActorPublisher(ref).subscribe(subs)
    watch(ref)
    ref
  }

  def expectTypeMetadata() = {
    expectMsgAnyClassOf(classOf[TypeMetadata])
  }

  def expectDone(pub: ActorRef) = {
    expectMsg(DoneWithQueryExecution.success)
    expectMsg("complete") //sent by ImplicitSubscriber
    expectTerminated(pub)
  }

  "JdbcSource" should {

    "run a simple statement" in {
      val pub = newPublisher("create table users(id VARCHAR);")
      pub ! ActorPublisherMessage.Request(1)
      expectTypeMetadata()
      pub ! ActorPublisherMessage.Request(1)
      expectDone(pub)
      runQuery("show tables") { stmt ⇒
        val rs = stmt.getResultSet
        rs.next()
        rs.getString(1).toUpperCase shouldBe "USERS"
      }
      runQuery("drop table users;")()
      testConnectionOpen()
    }

    //http://www.h2database.com/html/advanced.html
    //Please note that most data definition language (DDL) statements,
    //such as "create table", commit the current transaction. See the Grammar for details.
    "commit changes only if user is authenticated and has write access" in {
      runQuery("CREATE TABLE test_commit(id VARCHAR)")()
      runQuery("INSERT INTO test_commit (id) VALUES ('1234')")()
      //try to delete all data
      val pub = newPublisher("delete from test_commit;",
        RequestContext("a", Some(ApiUser("", 1, ApiKey.Mode.Read, None))))
      pub ! ActorPublisherMessage.Request(1)
      expectTypeMetadata()
      pub ! ActorPublisherMessage.Request(1)
      expectDone(pub)
      runQuery("select count(*) from test_commit") { stmt ⇒
        val rs = stmt.getResultSet
        rs.next()
        rs.getInt(1) shouldBe 1
      }
      runQuery("drop table test_commit;")()
      testConnectionOpen()
    }

    "run multiple statements" in {
      val pub = newPublisher(
        "create table one(id VARCHAR);" +
          "create table two(id VARCHAR);" +
          "--create table two(no VARCHAR);" +
          "create table three(id VARCHAR);" +
          "create table four(id VARCHAR);" +
          "create table five(id VARCHAR); ")
      pub ! ActorPublisherMessage.Request(1)
      expectTypeMetadata()
{'fraction_non_alphanumeric': 0.11093674553355867, 'fraction_numerical': 0.004104297440849831, 'mean_word_length': 3.756027554535017, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 1, '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': '4826854', 'n_tokens_mistral': 2585, 'n_tokens_neox': 2450, 'n_words': 565}