Fix test assertion to be called. Fix array equality test for headers. (…

…#98)

Fix assertion to handle empty Option.

Co-authored-by: misha <DL-OTTCloudPlatform-Nova@bskyb.internal>

scheduler/src/test/scala/com/sky/kms/e2e/SchedulerIntSpec.scala

@@ -16,8 +16,9 @@ class SchedulerIntSpec extends SchedulerIntSpecBase {
         withSchedulerApp {
           val schedules = createSchedules(2, forTopics = List(scheduleTopic, extraScheduleTopic))
 
-          publish(schedules) andThen
-            (assertMessagesWrittenFrom(_, schedules))
+          publish(schedules)
+            .foreach(assertMessagesWrittenFrom(_, schedules))
+
           assertTombstoned(schedules)
         }
       }
@@ -46,9 +47,17 @@ class SchedulerIntSpec extends SchedulerIntSpecBase {
           val cr = consumeFirstFrom[Array[Byte]](schedule.outputTopic)
 
           cr.key should contain theSameElementsInOrderAs schedule.key
-          cr.value should contain theSameElementsInOrderAs schedule.value.get
+
+          schedule.value match {
+            case Some(value) => cr.value should contain theSameElementsAs value
+            case None        => cr.value shouldBe null
+          }
+
           cr.timestamp shouldBe time.toInstant.toEpochMilli +- Tolerance.toMillis
-          cr.headers().toArray.map(h => h.key() -> h.value()).toMap should contain theSameElementsAs schedule.headers
+          cr.headers().toArray.map(h => h.key() -> h.value().toList) should contain theSameElementsAs
+            schedule.headers.map {
+              case (k, v) => (k, v.toList)
+            }
       }
 
     def assertTombstoned(schedules: List[(ScheduleId, ScheduleEvent)]): Unit =

scheduler/src/test/scala/com/sky/kms/e2e/package.scala

@@ -8,7 +8,7 @@ import scala.concurrent.duration._
 
 package object e2e {
 
-  val Tolerance = 200 millis
+  val Tolerance = 400 millis
 
   def withSchedulerApp[T](
       scenario: => T)(implicit conf: SchedulerConfig, system: ActorSystem, mat: ActorMaterializer): T =