<commit_before>package uk.ac.wellcome.platform.reindex.reindex_worker.services

import com.gu.scanamo.Scanamo
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.mockito.MockitoSugar
import org.scalatest.{FunSpec, Matchers}
import uk.ac.wellcome.messaging.test.fixtures.SNS.Topic
import uk.ac.wellcome.messaging.test.fixtures.SQS.QueuePair
import uk.ac.wellcome.messaging.test.fixtures.{SNS, SQS}
import uk.ac.wellcome.platform.reindex.reindex_worker.fixtures.{
  DynamoFixtures,
  ReindexableTable,
  WorkerServiceFixture
}
import uk.ac.wellcome.test.fixtures._
import uk.ac.wellcome.json.JsonUtil._
import uk.ac.wellcome.platform.reindex.reindex_worker.models.CompleteReindexParameters
import uk.ac.wellcome.storage.ObjectLocation
import uk.ac.wellcome.storage.fixtures.LocalDynamoDb.Table
import uk.ac.wellcome.storage.vhs.HybridRecord

class ReindexWorkerServiceTest
    extends FunSpec
    with Matchers
    with MockitoSugar
    with Akka
    with DynamoFixtures
    with ReindexableTable
    with SNS
    with SQS
    with ScalaFutures
    with WorkerServiceFixture {

  val exampleRecord = HybridRecord(
    id = "id",
    version = 1,
    location = ObjectLocation(
      namespace = "s3://example-bukkit",
      key = "<KEY>"
    )
  )

  it("completes a reindex") {
    withLocalDynamoDbTable { table =>
      withLocalSnsTopic { topic =>
        withLocalSqsQueueAndDlq { case QueuePair(queue, dlq) =>
          withWorkerService(queue, table, topic) { _ =>
            val reindexParameters = CompleteReindexParameters(
              segment = 0,
              totalSegments = 1
            )

            Scanamo.put(dynamoDbClient)(table.name)(exampleRecord)

            sendNotificationToSQS(
              queue = queue,
              message = createReindexRequestWith(parameters = reindexParameters)
            )

            eventually {
              val actualRecords: Seq[HybridRecord] =
                listMessagesReceivedFromSNS(topic)
                  .map {
                    _.message
                  }
                  .map {
                    fromJson[HybridRecord](_).get
                  }
                  .distinct

              actualRecords shouldBe List(exampleRecord)
              assertQueueEmpty(queue)
              assertQueueEmpty(dlq)
            }
          }
        }
      }
    }
  }

  it("fails if it cannot parse the SQS message as a ReindexJob") {
    withLocalDynamoDbTable { table =>
      withLocalSnsTopic { topic =>
        withLocalSqsQueueAndDlq {
          case QueuePair(queue, dlq) =>
            withWorkerService(queue, table, topic) { _ =>
              sendNotificationToSQS(
                queue = queue,
                body = "<xml>What is JSON.</xl?>"
              )

              eventually {
                assertQueueEmpty(queue)
                assertQueueHasSize(dlq, 1)
              }
            }
        }
      }
    }
  }

  it("fails if the reindex job fails") {
    val badTable = Table(name = "doesnotexist", index = "whatindex")
    val badTopic = Topic("does-not-exist")

    withLocalSqsQueueAndDlq { case QueuePair(queue, dlq) =>
      withWorkerService(queue, badTable, badTopic) { _ =>
        sendNotificationToSQS(queue = queue, message = createReindexRequest)

        eventually {
          assertQueueEmpty(queue)
          assertQueueHasSize(dlq, 1)
        }
      }
    }
  }
}<commit_msg>Add a test for passing an invalid job config ID<commit_after>package uk.ac.wellcome.platform.reindex.reindex_worker.services

import com.gu.scanamo.Scanamo
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.mockito.MockitoSugar
import org.scalatest.{FunSpec, Matchers}
import uk.ac.wellcome.messaging.test.fixtures.SNS.Topic
import uk.ac.wellcome.messaging.test.fixtures.SQS.QueuePair
import uk.ac.wellcome.messaging.test.fixtures.{SNS, SQS}
import uk.ac.wellcome.platform.reindex.reindex_worker.fixtures.{
  DynamoFixtures,
  ReindexableTable,
  WorkerServiceFixture
}
import uk.ac.wellcome.test.fixtures._
import uk.ac.wellcome.json.JsonUtil._
import uk.ac.wellcome.platform.reindex.reindex_worker.models.CompleteReindexParameters
import uk.ac.wellcome.storage.ObjectLocation
import uk.ac.wellcome.storage.fixtures.LocalDynamoDb.Table
import uk.ac.wellcome.storage.vhs.HybridRecord

class ReindexWorkerServiceTest
    extends FunSpec
    with Matchers
    with MockitoSugar
    with Akka
    with DynamoFixtures
    with ReindexableTable
    with SNS
    with SQS
    with ScalaFutures
    with WorkerServiceFixture {

  val exampleRecord = HybridRecord(
    id = "id",
    version = 1,
    location = ObjectLocation(
      namespace = "s3://example-bukkit",
      key = "<KEY>"
    )
  )

  it("completes a reindex") {
    withLocalDynamoDbTable { table =>
      withLocalSnsTopic { topic =>
        withLocalSqsQueueAndDlq { case QueuePair(queue, dlq) =>
          withWorkerService(queue, table, topic) { _ =>
            val reindexParameters = CompleteReindexParameters(
              segment = 0,
              totalSegments = 1
            )

            Scanamo.put(dynamoDbClient)(table.name)(exampleRecord)

            sendNotificationToSQS(
              queue = queue,
              message = createReindexRequestWith(parameters = reindexParameters)
            )

            eventually {
              val actualRecords: Seq[HybridRecord] =
                listMessagesReceivedFromSNS(topic)
                  .map {
                    _.message
                  }
                  .map {
                    fromJson[HybridRecord](_).get
                  }
                  .distinct

              actualRecords shouldBe List(exampleRecord)
              assertQueueEmpty(queue)
              assertQueueEmpty(dlq)
            }
          }
        }
      }
    }
  }

  it("fails if it cannot parse the SQS message as a ReindexJob") {
    withLocalDynamoDbTable { table =>
      withLocalSnsTopic { topic =>
        withLocalSqsQueueAndDlq {
          case QueuePair(queue, dlq) =>
            withWorkerService(queue, table, topic) { _ =>
              sendNotificationToSQS(
                queue = queue,
                body = "<xml>What is JSON.</xl?>"
              )

              eventually {
                assertQueueEmpty(queue)
                assertQueueHasSize(dlq, 1)
              }
            }
        }
      }
    }
  }

  it("fails if the reindex job fails") {
    val badTable = Table(name = "doesnotexist", index = "whatindex")
    val badTopic = Topic("does-not-exist")

    withLocalSqsQueueAndDlq { case QueuePair(queue, dlq) =>
      withWorkerService(queue, badTable, badTopic) { _ =>
        sendNotificationToSQS(queue = queue, message = createReindexRequest)

        eventually {
          assertQueueEmpty(queue)
          assertQueueHasSize(dlq, 1)
        }
      }
    }
  }

  it("fails if passed an invalid job ID") {
    withLocalDynamoDbTable { table =>
      withLocalSnsTopic { topic =>
        withLocalSqsQueueAndDlq { case QueuePair(queue, dlq) =>
          withWorkerService(queue, configMap = Map("foo" -> ((table, topic)))) { _ =>
            sendNotificationToSQS(
              queue = queue,
              message = createReindexRequestWith(jobConfigId = "bar")
            )

            eventually {
              assertQueueEmpty(queue)
              assertQueueHasSize(dlq, 1)
            }
          }
        }
      }
    }
  }
}
{'fraction_non_alphanumeric': 0.08654233194686703, 'fraction_numerical': 0.0017442640547430566, 'mean_word_length': 2.0524160524160524, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 34, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, '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': '3116479', 'n_tokens_mistral': 2249, 'n_tokens_neox': 2164, 'n_words': 393}