<commit_before>package uk.ac.wellcome.platform.sierra_reader.sink

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Source
import io.circe.parser._
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.{BeforeAndAfterAll, FunSpec, Matchers}
import uk.ac.wellcome.test.utils.S3Local
import scala.collection.JavaConversions._

class SequentialS3SinkTest
  extends FunSpec
  with Matchers
  with S3Local
  with BeforeAndAfterAll
  with ScalaFutures {

  implicit val system = ActorSystem()
  implicit val materialiser = ActorMaterializer()
  implicit val executionContext = system.dispatcher

  override def afterAll(): Unit = {
    system.terminate()
    materialiser.shutdown()
    super.afterAll()
  }

  val bucketName = createBucketAndReturnName("sequential-s3-sink-test")

  it("puts a single JSON in S3") {
    val sink = SequentialS3Sink(s3Client, bucketName = bucketName, keyPrefix = "testA_")

    val json = parse(s"""{"hello": "world"}""").right.get

    val futureDone = Source.single(json)
      .zipWithIndex
      .runWith(sink)

    whenReady(futureDone) { _ =>
      val s3objects = s3Client.listObjects(bucketName).getObjectSummaries
      s3objects should have size 1
      s3objects.head.getKey() shouldBe "testA_0000.json"
    }
  }

  it("puts multiple JSON bodies into S3") {
    val sink = SequentialS3Sink(s3Client, bucketName = bucketName, keyPrefix = "testB_")

    val json1 = parse(s"""{"red": "orange"}""").right.get
    val json2 = parse(s"""{"orange": "yellow"}""").right.get
    val json3 = parse(s"""{"yellow": "green"}""").right.get

    val futureDone = Source(List(json1, json2, json3))
      .zipWithIndex
      .runWith(sink)

    whenReady(futureDone) { _ =>
      val s3objects = s3Client.listObjects(bucketName).getObjectSummaries
      s3objects should have size 3
      s3objects.map { _.getKey() } shouldBe List("testB_0000.json", "testB_0001.json", "testB_0002.json")
    }
  }
}<commit_msg>Make stronger assertions in our tests<commit_after>package uk.ac.wellcome.platform.sierra_reader.sink

import akka.actor.ActorSystem
import akka.stream.ActorMaterializer
import akka.stream.scaladsl.Source
import io.circe.Json
import io.circe.parser._
import org.apache.commons.io.IOUtils
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.{BeforeAndAfterAll, FunSpec, Matchers}
import uk.ac.wellcome.test.utils.S3Local

import scala.collection.JavaConversions._

class SequentialS3SinkTest
  extends FunSpec
  with Matchers
  with S3Local
  with BeforeAndAfterAll
  with ScalaFutures {

  implicit val system = ActorSystem()
  implicit val materialiser = ActorMaterializer()
  implicit val executionContext = system.dispatcher

  override def afterAll(): Unit = {
    system.terminate()
    materialiser.shutdown()
    super.afterAll()
  }

  val bucketName = createBucketAndReturnName("sequential-s3-sink-test")

  it("puts a single JSON in S3") {
    val sink = SequentialS3Sink(s3Client, bucketName = bucketName, keyPrefix = "testA_")

    val json = parse(s"""{"hello": "world"}""").right.get

    val futureDone = Source.single(json)
      .zipWithIndex
      .runWith(sink)

    whenReady(futureDone) { _ =>
      val s3objects = s3Client.listObjects(bucketName).getObjectSummaries
      s3objects should have size 1
      s3objects.head.getKey() shouldBe "testA_0000.json"

      getJsonFromS3("testA_0000.json") shouldBe json
    }
  }

  it("puts multiple JSON bodies into S3") {
    val sink = SequentialS3Sink(s3Client, bucketName = bucketName, keyPrefix = "testB_")

    val json0 = parse(s"""{"red": "orange"}""").right.get
    val json1 = parse(s"""{"orange": "yellow"}""").right.get
    val json2 = parse(s"""{"yellow": "green"}""").right.get

    val futureDone = Source(List(json0, json1, json2))
      .zipWithIndex
      .runWith(sink)

    whenReady(futureDone) { _ =>
      val s3objects = s3Client.listObjects(bucketName).getObjectSummaries
      s3objects should have size 3
      s3objects.map { _.getKey() } shouldBe List("testB_0000.json", "testB_0001.json", "testB_0002.json")

      getJsonFromS3("testB_0000.json") shouldBe json0
      getJsonFromS3("testB_0001.json") shouldBe json1
      getJsonFromS3("testB_0002.json") shouldBe json2
    }
  }

  private def getJsonFromS3(key: String): Json = {
    val content = IOUtils.toString(s3Client.getObject(bucketName, key).getObjectContent)
    parse(content).right.get
  }
}
{'fraction_non_alphanumeric': 0.11659192825112108, 'fraction_numerical': 0.02443946188340807, 'mean_word_length': 4.075085324232082, 'pattern_counts': {'":': 8, '<': 3, '<?xml version=': 0, '>': 7, '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': '2659996', 'n_tokens_mistral': 1525, 'n_tokens_neox': 1467, 'n_words': 301}