Revert "EnhancedMonitoring and FailedRecordCount outputs should be nu…

…llable (#418)"

This reverts commit ba6ee35.

src/fun/scala/kinesis/mock/DescribeStreamSummaryTests.scala

@@ -1,5 +1,7 @@
 package kinesis.mock
 
+import java.util.Collections
+
 import cats.syntax.all._
 import software.amazon.awssdk.services.kinesis.model._
 
@@ -12,6 +14,12 @@ class DescribeStreamSummaryTests extends AwsFunctionalTests {
         .builder()
         .consumerCount(0)
         .encryptionType(EncryptionType.NONE)
+        .enhancedMonitoring(
+          EnhancedMetrics
+            .builder()
+            .shardLevelMetricsWithStrings(Collections.emptyList[String]())
+            .build()
+        )
         .openShardCount(genStreamShardCount)
         .retentionPeriodHours(24)
         .streamARN(

src/fun/scala/kinesis/mock/DescribeStreamTests.scala

@@ -1,5 +1,7 @@
 package kinesis.mock
 
+import java.util.Collections
+
 import software.amazon.awssdk.services.kinesis.model._
 
 import kinesis.mock.syntax.javaFuture._
@@ -19,6 +21,12 @@ class DescribeStreamTests extends AwsFunctionalTests {
       expected = StreamDescription
         .builder()
         .encryptionType(EncryptionType.NONE)
+        .enhancedMonitoring(
+          EnhancedMetrics
+            .builder()
+            .shardLevelMetricsWithStrings(Collections.emptyList[String]())
+            .build()
+        )
         .hasMoreShards(false)
         .shards(res.streamDescription().shards())
         .retentionPeriodHours(24)

src/main/scala/kinesis/mock/api/DeleteStreamRequest.scala

@@ -51,7 +51,7 @@ final case class DeleteStreamRequest(
                   shards = SortedMap.empty,
                   streamStatus = StreamStatus.DELETING,
                   tags = Tags.empty,
-                  enhancedMonitoring = None,
+                  enhancedMonitoring = Vector.empty,
                   consumers = SortedMap.empty
                 )
               )

src/main/scala/kinesis/mock/api/DisableEnhancedMonitoringRequest.scala

@@ -29,17 +29,17 @@ final case class DisableEnhancedMonitoringRequest(
             .flatMap(_ => CommonValidations.findStream(arn, streams))
             .map { stream =>
               val current =
-                stream.enhancedMonitoring.map(_.flatMap(_.shardLevelMetrics))
-              val desired: Option[Vector[ShardLevelMetric]] =
+                stream.enhancedMonitoring.flatMap(_.shardLevelMetrics)
+              val desired =
                 if (shardLevelMetrics.contains(ShardLevelMetric.ALL))
-                  None
-                else current.map(_.diff(shardLevelMetrics))
+                  Vector.empty
+                else current.diff(shardLevelMetrics)
 
               (
                 streams.updateStream(
                   stream
                     .copy(enhancedMonitoring =
-                      desired.map(x => Vector(ShardLevelMetrics(x)))
+                      Vector(ShardLevelMetrics(desired))
                     )
                 ),
                 DisableEnhancedMonitoringResponse(

src/main/scala/kinesis/mock/api/DisableEnhancedMonitoringResponse.scala

@@ -7,8 +7,8 @@ import io.circe
 import kinesis.mock.models._
 
 final case class DisableEnhancedMonitoringResponse(
-    currentShardLevelMetrics: Option[Vector[ShardLevelMetric]],
-    desiredShardLevelMetrics: Option[Vector[ShardLevelMetric]],
+    currentShardLevelMetrics: Vector[ShardLevelMetric],
+    desiredShardLevelMetrics: Vector[ShardLevelMetric],
     streamName: StreamName,
     streamArn: StreamArn
 )
@@ -35,10 +35,10 @@ object DisableEnhancedMonitoringResponse {
     for {
       currentShardLevelMetrics <- x
         .downField("CurrentShardLevelMetrics")
-        .as[Option[Vector[ShardLevelMetric]]]
+        .as[Vector[ShardLevelMetric]]
       desiredShardLevelMetrics <- x
         .downField("DesiredShardLevelMetrics")
-        .as[Option[Vector[ShardLevelMetric]]]
+        .as[Vector[ShardLevelMetric]]
       streamName <- x.downField("StreamName").as[StreamName]
       streamArn <- x.downField("StreamARN").as[StreamArn]
     } yield DisableEnhancedMonitoringResponse(

src/main/scala/kinesis/mock/api/EnableEnhancedMonitoringRequest.scala

@@ -29,23 +29,18 @@ final case class EnableEnhancedMonitoringRequest(
             .flatMap(_ => CommonValidations.findStream(arn, streams))
             .map { stream =>
               val current =
-                stream.enhancedMonitoring.map(_.flatMap(_.shardLevelMetrics))
+                stream.enhancedMonitoring.flatMap(_.shardLevelMetrics)
               val desired =
                 if (shardLevelMetrics.contains(ShardLevelMetric.ALL))
-                  Some(
-                    ShardLevelMetric.values
-                      .filterNot(_ == ShardLevelMetric.ALL)
-                      .toVector
-                  )
-                else
-                  (current
-                    .map(x => (x ++ shardLevelMetrics).distinct)
-                    .orElse(Some(shardLevelMetrics)))
+                  ShardLevelMetric.values
+                    .filterNot(_ == ShardLevelMetric.ALL)
+                    .toVector
+                else (current ++ shardLevelMetrics).distinct
               (
                 streams.updateStream(
                   stream
                     .copy(enhancedMonitoring =
-                      desired.map(x => Vector(ShardLevelMetrics(x)))
+                      Vector(ShardLevelMetrics(desired))
                     )
                 ),
                 EnableEnhancedMonitoringResponse(

src/main/scala/kinesis/mock/api/EnableEnhancedMonitoringResponse.scala

@@ -7,8 +7,8 @@ import io.circe
 import kinesis.mock.models._
 
 final case class EnableEnhancedMonitoringResponse(
-    currentShardLevelMetrics: Option[Vector[ShardLevelMetric]],
-    desiredShardLevelMetrics: Option[Vector[ShardLevelMetric]],
+    currentShardLevelMetrics: Vector[ShardLevelMetric],
+    desiredShardLevelMetrics: Vector[ShardLevelMetric],
     streamName: StreamName,
     streamArn: StreamArn
 )
@@ -35,10 +35,10 @@ object EnableEnhancedMonitoringResponse {
     for {
       currentShardLevelMetrics <- x
         .downField("CurrentShardLevelMetrics")
-        .as[Option[Vector[ShardLevelMetric]]]
+        .as[Vector[ShardLevelMetric]]
       desiredShardLevelMetrics <- x
         .downField("DesiredShardLevelMetrics")
-        .as[Option[Vector[ShardLevelMetric]]]
+        .as[Vector[ShardLevelMetric]]
       streamName <- x.downField("StreamName").as[StreamName]
       streamArn <- x.downField("StreamARN").as[StreamArn]
     } yield EnableEnhancedMonitoringResponse(

src/main/scala/kinesis/mock/api/PutRecordsRequest.scala

@@ -117,7 +117,7 @@ final case class PutRecordsRequest(
             ),
             PutRecordsResponse(
               stream.encryptionType,
-              None,
+              0,
               asRecords.map { case (_, _, _, entry) =>
                 entry
               }

src/main/scala/kinesis/mock/api/PutRecordsResponse.scala

@@ -8,7 +8,7 @@ import kinesis.mock.models.EncryptionType
 
 final case class PutRecordsResponse(
     encryptionType: EncryptionType,
-    failedRecordCount: Option[Int],
+    failedRecordCount: Int,
     records: Vector[PutRecordsResultEntry]
 )
 
@@ -26,7 +26,7 @@ object PutRecordsResponse {
     x =>
       for {
         encryptionType <- x.downField("EncryptionType").as[EncryptionType]
-        failedRecordCount <- x.downField("FailedRecordCount").as[Option[Int]]
+        failedRecordCount <- x.downField("FailedRecordCount").as[Int]
         records <- x.downField("Records").as[Vector[PutRecordsResultEntry]]
       } yield PutRecordsResponse(encryptionType, failedRecordCount, records)
 

src/main/scala/kinesis/mock/models/StreamData.scala

@@ -15,7 +15,7 @@ import kinesis.mock.instances.circe._
 final case class StreamData(
     consumers: SortedMap[ConsumerName, Consumer],
     encryptionType: EncryptionType,
-    enhancedMonitoring: Option[Vector[ShardLevelMetrics]],
+    enhancedMonitoring: Vector[ShardLevelMetrics],
     keyId: Option[String],
     retentionPeriod: FiniteDuration,
     shards: SortedMap[Shard, Vector[KinesisRecord]],
@@ -81,7 +81,7 @@ object StreamData {
       encryptionType <- x.downField("encryptionType").as[EncryptionType]
       enhancedMonitoring <- x
         .downField("enhancedMonitoring")
-        .as[Option[Vector[ShardLevelMetrics]]]
+        .as[Vector[ShardLevelMetrics]]
       keyId <- x.downField("keyId").as[Option[String]]
       retentionPeriod <- x.downField("retentionPeriod").as[FiniteDuration]
       shards <- x
@@ -144,7 +144,7 @@ object StreamData {
     StreamData(
       SortedMap.empty,
       EncryptionType.NONE,
-      None,
+      Vector(ShardLevelMetrics(Vector.empty)),
       None,
       minRetentionPeriod,
       shards,

src/main/scala/kinesis/mock/models/StreamDescription.scala

@@ -11,7 +11,7 @@ import kinesis.mock.instances.circe._
 
 final case class StreamDescription(
     encryptionType: Option[EncryptionType],
-    enhancedMonitoring: Option[Vector[ShardLevelMetrics]],
+    enhancedMonitoring: Vector[ShardLevelMetrics],
     hasMoreShards: Boolean,
     keyId: Option[String],
     retentionPeriodHours: Int,
@@ -100,7 +100,7 @@ object StreamDescription {
         .as[Option[EncryptionType]]
       enhancedMonitoring <- x
         .downField("EnhancedMonitoring")
-        .as[Option[Vector[ShardLevelMetrics]]]
+        .as[Vector[ShardLevelMetrics]]
       hasMoreShards <- x.downField("HasMoreShards").as[Boolean]
       keyId <- x.downField("KeyId").as[Option[String]]
       retentionPeriodHours <- x.downField("RetentionPeriodHours").as[Int]

src/main/scala/kinesis/mock/models/StreamDescriptionSummary.scala

@@ -11,7 +11,7 @@ import kinesis.mock.instances.circe._
 final case class StreamDescriptionSummary(
     consumerCount: Option[Int],
     encryptionType: Option[EncryptionType],
-    enhancedMonitoring: Option[Vector[ShardLevelMetrics]],
+    enhancedMonitoring: Vector[ShardLevelMetrics],
     keyId: Option[String],
     openShardCount: Int,
     retentionPeriodHours: Int,
@@ -76,7 +76,7 @@ object StreamDescriptionSummary {
       encryptionType <- x.downField("EncryptionType").as[Option[EncryptionType]]
       enhancedMonitoring <- x
         .downField("EnhancedMonitoring")
-        .as[Option[Vector[ShardLevelMetrics]]]
+        .as[Vector[ShardLevelMetrics]]
       keyId <- x.downField("KeyId").as[Option[String]]
       openShardCount <- x.downField("OpenShardCount").as[Int]
       retentionPeriodHours <- x.downField("RetentionPeriodHours").as[Int]

src/main/scala/kinesis/mock/models/Streams.scala

@@ -42,7 +42,7 @@ final case class Streams(streams: SortedMap[StreamArn, StreamData]) {
             shards = SortedMap.empty,
             streamStatus = StreamStatus.DELETING,
             tags = Tags.empty,
-            enhancedMonitoring = None,
+            enhancedMonitoring = Vector.empty,
             consumers = SortedMap.empty
           )
         )

src/test/scala/kinesis/mock/api/DisableEnhancedMonitoringTests.scala

@@ -21,18 +21,16 @@ class DisableEnhancedMonitoringTests
 
       val updated = streams.findAndUpdateStream(streamArn)(stream =>
         stream.copy(enhancedMonitoring =
-          Some(
-            Vector(
-              ShardLevelMetrics(
-                Vector(
-                  ShardLevelMetric.IncomingBytes,
-                  ShardLevelMetric.IncomingRecords,
-                  ShardLevelMetric.OutgoingBytes,
-                  ShardLevelMetric.OutgoingRecords,
-                  ShardLevelMetric.WriteProvisionedThroughputExceeded,
-                  ShardLevelMetric.ReadProvisionedThroughputExceeded,
-                  ShardLevelMetric.IteratorAgeMilliseconds
-                )
+          Vector(
+            ShardLevelMetrics(
+              Vector(
+                ShardLevelMetric.IncomingBytes,
+                ShardLevelMetric.IncomingRecords,
+                ShardLevelMetric.OutgoingBytes,
+                ShardLevelMetric.OutgoingRecords,
+                ShardLevelMetric.WriteProvisionedThroughputExceeded,
+                ShardLevelMetric.ReadProvisionedThroughputExceeded,
+                ShardLevelMetric.IteratorAgeMilliseconds
               )
             )
           )
@@ -54,7 +52,7 @@ class DisableEnhancedMonitoringTests
         s <- streamsRef.get
         updatedMetrics = s.streams
           .get(streamArn)
-          .map(_.enhancedMonitoring.map(_.flatMap(_.shardLevelMetrics)))
+          .map(_.enhancedMonitoring.flatMap(_.shardLevelMetrics))
 
       } yield assert(
         res.isRight && res.exists { case response =>

src/test/scala/kinesis/mock/api/EnableEnhancedMonitoringTests.scala

@@ -34,7 +34,7 @@ class EnableEnhancedMonitoringTests
         s <- streamsRef.get
         updatedMetrics = s.streams
           .get(streamArn)
-          .map(_.enhancedMonitoring.map(_.flatMap(_.shardLevelMetrics)))
+          .map(_.enhancedMonitoring.flatMap(_.shardLevelMetrics))
 
       } yield assert(
         res.isRight && res.exists { case response =>

src/test/scala/kinesis/mock/cache/DescribeStreamSummaryTests.scala

@@ -44,7 +44,7 @@ class DescribeStreamSummaryTests
         expected = StreamDescriptionSummary(
           Some(0),
           Some(EncryptionType.NONE),
-          None,
+          Vector(ShardLevelMetrics(Vector.empty)),
           None,
           1,
           24,

src/test/scala/kinesis/mock/cache/DescribeStreamTests.scala

@@ -60,7 +60,7 @@ class DescribeStreamTests
           .map(x => x.shards)
         expected = StreamDescription(
           Some(EncryptionType.NONE),
-          None,
+          Vector(ShardLevelMetrics(Vector.empty)),
           false,
           None,
           24,

src/test/scala/kinesis/mock/cache/DisableEnhancedMonitoringTests.scala

@@ -67,11 +67,11 @@ class DisableEnhancedMonitoringTests
           .rethrow
           .map(
             _.streamDescriptionSummary.enhancedMonitoring
-              .map(_.flatMap(_.shardLevelMetrics))
+              .flatMap(_.shardLevelMetrics)
           )
       } yield assert(
         res.desiredShardLevelMetrics == streamMonitoring && !res.desiredShardLevelMetrics
-          .exists(_.contains(ShardLevelMetric.IncomingBytes))
+          .contains(ShardLevelMetric.IncomingBytes)
       )
   })
 }

src/test/scala/kinesis/mock/cache/EnableEnhancedMonitoringTests.scala

@@ -55,11 +55,11 @@ class EnableEnhancedMonitoringTests
           .rethrow
           .map(
             _.streamDescriptionSummary.enhancedMonitoring
-              .map(_.flatMap(_.shardLevelMetrics))
+              .flatMap(_.shardLevelMetrics)
           )
       } yield assert(
         res.desiredShardLevelMetrics == streamMonitoring && res.desiredShardLevelMetrics
-          .exists(_.contains(ShardLevelMetric.IncomingBytes))
+          .contains(ShardLevelMetric.IncomingBytes)
       )
   })
 }