import org.apache.log4j.{Level, Logger}
import org.specs2.mutable._
import Predef.{conforms => _, _}

class ExampleTest extends Specification {
  Logger.getRootLogger.setLevel(Level.ERROR)

  sequential

  "the transformStream method" should {
    implicit val fun = Example.transformStream _

    "with 10 identical records" should {
      val records = Seq.fill(10)("""{"item_id":"abc123","amount":1.23,"time":1431504603105}""")
      "return a single record with the correct total" in new spark(records) {
        collector.length mustEqual 1
        val output = collector.head
        output.total mustEqual BigDecimal(12.3)
      }
    }

    "with invalid records" should {
      val records = Seq("this is not json", """{"this":"isn't in the right format"}""")
      "output no records" in new spark(records) {
        collector.length mustEqual 0
      }
    }
  }
}