<commit_before>package uk.ac.wellcome.platform.goobi_reader.services

import java.io.{ByteArrayInputStream, InputStream}
import java.time.Instant
import java.time.temporal.ChronoUnit

import com.amazonaws.services.s3.{AmazonS3, AmazonS3Client}
import com.gu.scanamo.Scanamo
import org.mockito.Matchers.any
import org.mockito.Mockito.when
import org.scalatest.{FunSpec, Inside}
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.mockito.MockitoSugar
import uk.ac.wellcome.messaging.sns.NotificationMessage
import uk.ac.wellcome.messaging.test.fixtures.{Messaging, SQS}
import uk.ac.wellcome.messaging.test.fixtures.SQS.{Queue, QueuePair}
import uk.ac.wellcome.monitoring.MetricsSender
import uk.ac.wellcome.monitoring.fixtures.MetricsSenderFixture
import uk.ac.wellcome.platform.goobi_reader.GoobiRecordMetadata
import uk.ac.wellcome.platform.goobi_reader.fixtures.GoobiReaderFixtures
import uk.ac.wellcome.storage.ObjectStore
import uk.ac.wellcome.storage.dynamo._
import uk.ac.wellcome.storage.fixtures.LocalDynamoDb.Table
import uk.ac.wellcome.storage.fixtures.S3.Bucket
import uk.ac.wellcome.storage.vhs.{HybridRecord, VersionedHybridStore}
import uk.ac.wellcome.test.fixtures.{Akka, TestWith}
import uk.ac.wellcome.test.utils.ExtendedPatience

import scala.concurrent.ExecutionContext.Implicits.global

class GoobiReaderWorkerServiceTest
    extends FunSpec
    with Akka
    with MetricsSenderFixture
    with ScalaFutures
    with ExtendedPatience
    with MockitoSugar
    with GoobiReaderFixtures
    with Messaging
    with SQS
    with Inside {

  private val id = "mets-0001"
  private val goobiS3Prefix = "goobi"
  private val sourceKey = s"$id.xml"
  private val contents = "muddling the machinations of morose METS"
  private val updatedContents = "Updated contents"
  private val eventTime = Instant.parse("2018-01-01T01:00:00.000Z")

  it("processes a notification") {
    withGoobiReaderWorkerService(s3Client) {
      case (bucket, QueuePair(queue, _), _, table, _) =>
        s3Client.putObject(bucket.name, sourceKey, contents)

        sendNotificationToSQS(
          queue = queue,
          body = anS3Notification(sourceKey, bucket.name, eventTime)
        )

        eventually {
          assertRecordStored(
            id = id,
            version = 1,
            expectedMetadata = GoobiRecordMetadata(eventTime),
            expectedContents = contents,
            table = table,
            bucket = bucket)
        }
    }
  }

  it("ingests an object with a space in the s3Key") {
    withGoobiReaderWorkerService(s3Client) {
      case (bucket, QueuePair(queue, _), _, table, _) =>
        val sourceKey = s"$id work.xml"
        val urlEncodedSourceKey =
          java.net.URLEncoder.encode(sourceKey, "utf-8")
        s3Client.putObject(bucket.name, sourceKey, contents)

        sendNotificationToSQS(
          queue = queue,
          body = anS3Notification(urlEncodedSourceKey, bucket.name, eventTime)
        )

        eventually {

          assertRecordStored(
            id = s"$id work",
            version = 1,
            expectedMetadata = GoobiRecordMetadata(eventTime),
            expectedContents = contents,
            table = table,
            bucket = bucket)
        }
    }
  }

  it("doesn't overwrite a newer work with an older work") {
    withGoobiReaderWorkerService(s3Client) {
      case (bucket, QueuePair(queue, dlq), _, table, vhs) =>
        val contentStream = new ByteArrayInputStream(contents.getBytes)
        vhs.updateRecord(id = id)(
          ifNotExisting = (contentStream, GoobiRecordMetadata(eventTime)))(
          ifExisting = (t, m) => (t, m))
        eventually {
          val expectedMetadata = GoobiRecordMetadata(eventTime)
          assertRecordStored(
            id = id,
            version = 1,
            expectedMetadata = expectedMetadata,
            expectedContents = contents,
            table = table,
            bucket = bucket)
        }

        s3Client.putObject(bucket.name, sourceKey, contents)

        val olderEventTime = eventTime.minus(1, ChronoUnit.HOURS)

        sendNotificationToSQS(
          queue = queue,
          body = anS3Notification(sourceKey, bucket.name, olderEventTime)
        )

        eventually {

          val expectedMetadata = GoobiRecordMetadata(eventTime)
          assertRecordStored(
            id = id,
            version = 1,
            expectedMetadata,
            contents,
            table,
            bucket)
        }
    }
  }

  it("overwrites an older work with an newer work") {
    withGoobiReaderWorkerService(s3Client) {
      case (bucket, QueuePair(queue, dlq), _, table, vhs) =>
        val contentStream = new ByteArrayInputStream(contents.getBytes)
        vhs.updateRecord(id = id)(
          ifNotExisting = (contentStream, GoobiRecordMetadata(eventTime)))(
          ifExisting = (t, m) => (t, m))
        eventually {

          assertRecordStored(
            id = id,
            version = 1,
            GoobiRecordMetadata(eventTime),
            contents,
            table,
            bucket)
        }

        s3Client.putObject(bucket.name, sourceKey, updatedContents)
        val newerEventTime = eventTime.plus(1, ChronoUnit.HOURS)

        sendNotificationToSQS(
          queue = queue,
          body = anS3Notification(sourceKey, bucket.name, newerEventTime)
        )

        eventually {
          val expectedMetadata = GoobiRecordMetadata(newerEventTime)
          assertRecordStored(
            id = id,
            version = 2,
            expectedMetadata,
            updatedContents,
            table,
            bucket)
        }
    }
  }

  it("fails gracefully if Json cannot be parsed") {
    withGoobiReaderWorkerService(s3Client) {
      case (bucket, QueuePair(queue, dlq), mockMetricsSender, table, _) =>
        sendNotificationToSQS(
          queue = queue,
          body = "NotJson"
        )

        eventually {
          assertMessageSentToDlq(queue, dlq)
          assertUpdateNotSaved(bucket, table)
          assertFailureMetricNotIncremented(mockMetricsSender)
        }
    }
  }

  it("does not fail gracefully if content cannot be fetched") {
    withGoobiReaderWorkerService(s3Client) {
      case (bucket, QueuePair(queue, dlq), mockMetricsSender, table, _) =>
        val sourceKey = "NotThere.xml"

        sendNotificationToSQS(
          queue = queue,
          body = anS3Notification(sourceKey, bucket.name, eventTime)
        )

        eventually {
          assertMessageSentToDlq(queue, dlq)
          assertUpdateNotSaved(bucket, table)
          assertFailureMetricIncremented(mockMetricsSender)
        }
    }
  }

  it("does not fail gracefully when there is an unexpected failure") {
    val mockClient = mock[AmazonS3Client]
    val expectedException = new RuntimeException("Failed!")
    when(mockClient.getObject(any[String], any[String]))
      .thenThrow(expectedException)
    withGoobiReaderWorkerService(mockClient) {
      case (bucket, QueuePair(queue, dlq), mockMetricsSender, table, _) =>
        val sourceKey = "any.xml"
        sendNotificationToSQS(
          queue = queue,
          body = anS3Notification(sourceKey, bucket.name, eventTime)
        )

        eventually {
          assertMessageSentToDlq(queue, dlq)
          assertUpdateNotSaved(bucket, table)
          assertFailureMetricIncremented(mockMetricsSender)
        }
    }
  }

  private def assertRecordStored(id: String,
                                 version: Int,
                                 expectedMetadata: GoobiRecordMetadata,
                                 expectedContents: String,
                                 table: Table,
                                 bucket: Bucket) = {

    inside(getHybridRecord(table, id)) {<commit_msg>Fix the Goobi reader tests<commit_after>package uk.ac.wellcome.platform.goobi_reader.services

import java.io.{ByteArrayInputStream, InputStream}
import java.time.Instant
import java.time.temporal.ChronoUnit

import com.amazonaws.services.s3.{AmazonS3, AmazonS3Client}
import com.gu.scanamo.Scanamo
import org.mockito.Matchers.any
import org.mockito.Mockito.{never, times, verify, when}
import org.scalatest.{FunSpec, Inside}
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.mockito.MockitoSugar
import uk.ac.wellcome.messaging.sns.NotificationMessage
import uk.ac.wellcome.messaging.test.fixtures.{Messaging, SQS}
import uk.ac.wellcome.messaging.test.fixtures.SQS.{Queue, QueuePair}
import uk.ac.wellcome.monitoring.MetricsSender
import uk.ac.wellcome.monitoring.fixtures.MetricsSenderFixture
import uk.ac.wellcome.platform.goobi_reader.GoobiRecordMetadata
import uk.ac.wellcome.platform.goobi_reader.fixtures.GoobiReaderFixtures
import uk.ac.wellcome.storage.ObjectStore
import uk.ac.wellcome.storage.dynamo._
import uk.ac.wellcome.storage.fixtures.LocalDynamoDb.Table
import uk.ac.wellcome.storage.fixtures.S3.Bucket
import uk.ac.wellcome.storage.vhs.{HybridRecord, VersionedHybridStore}
import uk.ac.wellcome.test.fixtures.{Akka, TestWith}
import uk.ac.wellcome.test.utils.ExtendedPatience

import scala.concurrent.ExecutionContext.Implicits.global

class GoobiReaderWorkerServiceTest
    extends FunSpec
    with Akka
    with MetricsSenderFixture
    with ScalaFutures
    with ExtendedPatience
    with MockitoSugar
    with GoobiReaderFixtures
    with Messaging
    with SQS
    with Inside {

  private val id = "mets-0001"
  private val goobiS3Prefix = "goobi"
  private val sourceKey = s"$id.xml"
  private val contents = "muddling the machinations of morose METS"
  private val updatedContents = "Updated contents"
  private val eventTime = Instant.parse("2018-01-01T01:00:00.000Z")

  it("processes a notification") {
    withGoobiReaderWorkerService(s3Client) {
      case (bucket, QueuePair(queue, _), _, table, _) =>
        s3Client.putObject(bucket.name, sourceKey, contents)

        sendNotificationToSQS(
          queue = queue,
          body = anS3Notification(sourceKey, bucket.name, eventTime)
        )

        eventually {
          assertRecordStored(
            id = id,
            version = 1,
            expectedMetadata = GoobiRecordMetadata(eventTime),
            expectedContents = contents,
            table = table,
            bucket = bucket)
        }
    }
  }

  it("ingests an object with a space in the s3Key") {
    withGoobiReaderWorkerService(s3Client) {
      case (bucket, QueuePair(queue, _), _, table, _) =>
        val sourceKey = s"<KEY>"
        val urlEncodedSourceKey =
          java.net.URLEncoder.encode(sourceKey, "utf-8")
        s3Client.putObject(bucket.name, sourceKey, contents)

        sendNotificationToSQS(
          queue = queue,
          body = anS3Notification(urlEncodedSourceKey, bucket.name, eventTime)
        )

        eventually {

          assertRecordStored(
            id = s"$id work",
            version = 1,
            expectedMetadata = GoobiRecordMetadata(eventTime),
            expectedContents = contents,
            table = table,
            bucket = bucket)
        }
    }
  }

  it("doesn't overwrite a newer work with an older work") {
    withGoobiReaderWorkerService(s3Client) {
      case (bucket, QueuePair(queue, dlq), _, table, vhs) =>
        val contentStream = new ByteArrayInputStream(contents.getBytes)
        vhs.updateRecord(id = id)(
          ifNotExisting = (contentStream, GoobiRecordMetadata(eventTime)))(
          ifExisting = (t, m) => (t, m))
        eventually {
          val expectedMetadata = GoobiRecordMetadata(eventTime)
          assertRecordStored(
            id = id,
            version = 1,
            expectedMetadata = expectedMetadata,
            expectedContents = contents,
            table = table,
            bucket = bucket)
        }

        s3Client.putObject(bucket.name, sourceKey, contents)

        val olderEventTime = eventTime.minus(1, ChronoUnit.HOURS)

        sendNotificationToSQS(
          queue = queue,
          body = anS3Notification(sourceKey, bucket.name, olderEventTime)
        )

        eventually {

          val expectedMetadata = GoobiRecordMetadata(eventTime)
          assertRecordStored(
            id = id,
            version = 1,
            expectedMetadata,
            contents,
            table,
            bucket)
        }
    }
  }

  it("overwrites an older work with an newer work") {
    withGoobiReaderWorkerService(s3Client) {
      case (bucket, QueuePair(queue, dlq), _, table, vhs) =>
        val contentStream = new ByteArrayInputStream(contents.getBytes)
        vhs.updateRecord(id = id)(
          ifNotExisting = (contentStream, GoobiRecordMetadata(eventTime)))(
          ifExisting = (t, m) => (t, m))
        eventually {

          assertRecordStored(
            id = id,
            version = 1,
            GoobiRecordMetadata(eventTime),
            contents,
            table,
            bucket)
        }

        s3Client.putObject(bucket.name, sourceKey, updatedContents)
        val newerEventTime = eventTime.plus(1, ChronoUnit.HOURS)

        sendNotificationToSQS(
          queue = queue,
          body = anS3Notification(sourceKey, bucket.name, newerEventTime)
        )

        eventually {
          val expectedMetadata = GoobiRecordMetadata(newerEventTime)
          assertRecordStored(
            id = id,
            version = 2,
            expectedMetadata,
            updatedContents,
            table,
            bucket)
        }
    }
  }

  it("fails gracefully if Json cannot be parsed") {
    withGoobiReaderWorkerService(s3Client) {
      case (bucket, QueuePair(queue, dlq), metricsSender, table, _) =>
        sendNotificationToSQS(
          queue = queue,
          body = "NotJson"
        )

        eventually {
          assertMessageSentToDlq(queue, dlq)
          assertUpdateNotSaved(bucket, table)
          verify(metricsSender, never())
            .countRecognisedFailure(any[String])
        }
    }
  }

  it("does not fail gracefully if content cannot be fetched") {
    withGoobiReaderWorkerService(s3Client) {
      case (bucket, QueuePair(queue, dlq), metricsSender, table, _) =>
        val sourceKey = "NotThere.<PASSWORD>"

        sendNotificationToSQS(
          queue = queue,
          body = anS3Notification(sourceKey, bucket.name, eventTime)
        )

        eventually {
          assertMessageSentToDlq(queue, dlq)
          assertUpdateNotSaved(bucket, table)
          verify(metricsSender, times(3))
            .countRecognisedFailure(any[String])
        }
    }
  }

  it("does not fail gracefully when there is an unexpected failure") {
    val mockClient = mock[AmazonS3Client]
    val expectedException = new RuntimeException("Failed!")
    when(mockClient.getObject(any[String], any[String]))
      .thenThrow(expectedException)
    withGoobiReaderWorkerService(mockClient) {
      case (bucket, QueuePair(queue, dlq), metricsSender, table, _) =>
        val sourceKey = "any.xml"
        sendNotificationToSQS(
          queue = queue,
          body = anS3Notification(sourceKey, bucket.name, eventTime)
        )

        eventually {
          assertMessageSentToDlq(queue, dlq)
          assertUpdateNotSaved(bucket, table)
          verify(metricsSender, times(3))
            .countRecognisedFailure(any[String])
        }
    }
  }

  private def assertRecordStored(id: String,
                                 version: Int,
                                 expectedMetadata: GoobiRecordMetadata,
                                 expectedContents: String,
                                 table: Table,
                                 bucket: Bucket) = {

    inside(getHybridRecord(table, id)) {
{'fraction_non_alphanumeric': 0.0860864040660737, 'fraction_numerical': 0.006861499364675985, 'mean_word_length': 2.3815252416756176, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 6}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2568255', 'n_tokens_mistral': 4575, 'n_tokens_neox': 4442, 'n_words': 908}