-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
c5455ae
commit 11153c4
Showing
6 changed files
with
206 additions
and
15 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
65 changes: 65 additions & 0 deletions
65
...is-it/src/test/scala/com/snowplowanalytics/snowplow/sources/kinesis/KinesisSinkSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| /* | ||
| * Copyright (c) 2023-present Snowplow Analytics Ltd. All rights reserved. | ||
| * | ||
| * This program is licensed to you under the Snowplow Community License Version 1.0, | ||
| * and you may not use this file except in compliance with the Snowplow Community License Version 1.0. | ||
| * You may obtain a copy of the Snowplow Community License Version 1.0 at https://docs.snowplow.io/community-license-1.0 | ||
| */ | ||
| package com.snowplowanalytics.snowplow.sinks.kinesis | ||
|
|
||
| import cats.effect.{IO, Resource} | ||
| import cats.effect.testing.specs2.CatsResource | ||
|
|
||
| import scala.concurrent.duration.{DurationInt, FiniteDuration} | ||
|
|
||
| import org.specs2.mutable.SpecificationLike | ||
|
|
||
| import org.testcontainers.containers.localstack.LocalStackContainer | ||
|
|
||
| import software.amazon.awssdk.regions.providers.DefaultAwsRegionProviderChain | ||
| import software.amazon.awssdk.regions.Region | ||
|
|
||
| import com.snowplowanalytics.snowplow.it.kinesis._ | ||
| import com.snowplowanalytics.snowplow.sinks.{Sink, Sinkable} | ||
|
|
||
| import Utils._ | ||
|
|
||
| class KinesisSinkSpec extends CatsResource[IO, (Region, LocalStackContainer, Sink[IO])] with SpecificationLike { | ||
| import KinesisSinkSpec._ | ||
|
|
||
| override val Timeout: FiniteDuration = 3.minutes | ||
|
|
||
| /** Resources which are shared across tests */ | ||
| override val resource: Resource[IO, (Region, LocalStackContainer, Sink[IO])] = | ||
| for { | ||
| region <- Resource.eval(IO.blocking((new DefaultAwsRegionProviderChain).getRegion)) | ||
| localstack <- Localstack.resource(region, KINESIS_INITIALIZE_STREAMS, KinesisSinkSpec.getClass.getSimpleName) | ||
| testSink <- KinesisSink.resource[IO](getKinesisSinkConfig(localstack.getEndpoint)(testStream1Name)) | ||
| } yield (region, localstack, testSink) | ||
|
|
||
| override def is = s2""" | ||
| KinesisSinkSpec should | ||
| write to output stream $e1 | ||
| """ | ||
|
|
||
| def e1 = withResource { case (region, localstack, testSink) => | ||
| val testPayload = "test-payload" | ||
| val testInput = List(Sinkable(testPayload.getBytes(), Some("myPk"), Map(("", "")))) | ||
|
|
||
| for { | ||
| kinesisClient <- getKinesisClient(localstack.getEndpoint, region) | ||
| _ <- testSink.sink(testInput) | ||
| _ <- IO.sleep(3.seconds) | ||
| result = getDataFromKinesis(kinesisClient, testStream1Name) | ||
| } yield List( | ||
| result.events must haveSize(1), | ||
| result.events must beEqualTo(List(testPayload)) | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| object KinesisSinkSpec { | ||
| val testStream1Name = "test-sink-stream-1" | ||
| val KINESIS_INITIALIZE_STREAMS: String = | ||
| List(s"$testStream1Name:1").mkString(",") | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
69 changes: 69 additions & 0 deletions
69
...s/src/test/scala/com/snowplowanalytics/snowplow/sinks/kinesis/KinesisSinkConfigSpec.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| /* | ||
| * Copyright (c) 2023-present Snowplow Analytics Ltd. All rights reserved. | ||
| * | ||
| * This program is licensed to you under the Snowplow Community License Version 1.0, | ||
| * and you may not use this file except in compliance with the Snowplow Community License Version 1.0. | ||
| * You may obtain a copy of the Snowplow Community License Version 1.0 at https://docs.snowplow.io/community-license-1.0 | ||
| */ | ||
| package com.snowplowanalytics.snowplow.sinks.kinesis | ||
|
|
||
| import io.circe.literal._ | ||
| import org.specs2.Specification | ||
| import scala.concurrent.duration.FiniteDuration | ||
|
|
||
| import java.net.URI | ||
|
|
||
| class KinesisSinkConfigSpec extends Specification { | ||
|
|
||
| def is = s2""" | ||
| The KinesisSource decoder should: | ||
| Decode a valid JSON config $e1 | ||
| Fail to decode an invalid JSON config $e2 | ||
| """ | ||
|
|
||
| def e1 = { | ||
| val json = json""" | ||
| { | ||
| "streamName": "my-stream", | ||
| "throttledBackoffPolicy": { | ||
| "minBackoff": "100ms", | ||
| "maxBackoff": "500ms", | ||
| "maxRetries": 3 | ||
| }, | ||
| "recordLimit": 1000, | ||
| "byteLimit": 1000, | ||
| "customEndpoint": "http://localhost:4040" | ||
| } | ||
| """ | ||
|
|
||
| json.as[KinesisSinkConfig] must beRight.like { case c: KinesisSinkConfig => | ||
| List( | ||
| c.streamName must beEqualTo("my-stream"), | ||
| c.throttledBackoffPolicy.minBackoff must beEqualTo(FiniteDuration(100, "ms")), | ||
| c.throttledBackoffPolicy.maxBackoff must beEqualTo(FiniteDuration(500, "ms")), | ||
| c.throttledBackoffPolicy.maxRetries must beEqualTo(Some(3)), | ||
| c.recordLimit must beEqualTo(1000), | ||
| c.byteLimit must beEqualTo(1000), | ||
| c.customEndpoint must beEqualTo(Some(URI.create("http://localhost:4040"))) | ||
| ).reduce(_ and _) | ||
| } | ||
| } | ||
|
|
||
| def e2 = { | ||
| val json = json""" | ||
| { | ||
| "throttledBackoffPolicy": { | ||
| "minBackoff": "100ms", | ||
| "maxBackoff": "500ms", | ||
| "maxRetries": 3 | ||
| }, | ||
| "recordLimit": 1000, | ||
| "byteLimit": 1000, | ||
| "customEndpoint": "http://localhost:4040" | ||
| } | ||
| """ | ||
|
|
||
| json.as[KinesisSinkConfig] must beLeft | ||
|
|
||
| } | ||
| } |