Skip to content

[FLINK-27941][connectors/aws] Migrate remaining tests to JUnit 5 #244

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
jlalwani-amazon wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[FLINK-27941][connectors/aws] Migrate remaining tests to JUnit 5 #244

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 16 additions & 16 deletions .../test/java/org/apache/flink/connector/firehose/table/test/KinesisFirehoseTableITTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@
import org.apache.flink.test.resources.ResourceTestUtils;
import org.apache.flink.test.util.SQLJobSubmission;
import org.apache.flink.util.DockerImageVersions;
import org.apache.flink.util.TestLogger;

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
Expand All @@ -35,16 +34,17 @@
import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import org.assertj.core.api.Assertions;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.rules.Timeout;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.Network;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;
import software.amazon.awssdk.core.SdkSystemSetting;
import software.amazon.awssdk.http.SdkHttpClient;
Expand Down Expand Up @@ -74,7 +74,9 @@
import static org.apache.flink.connector.firehose.sink.testutils.KinesisFirehoseTestUtils.createFirehoseClient;

/** End to End test for Kinesis Firehose Table sink API. */
public class KinesisFirehoseTableITTest extends TestLogger {
@Testcontainers
@Timeout(value = 10, unit = TimeUnit.MINUTES)
public class KinesisFirehoseTableITTest {

private static final Logger LOG = LoggerFactory.getLogger(KinesisFirehoseTableITTest.class);

Expand All @@ -95,9 +97,7 @@ public class KinesisFirehoseTableITTest extends TestLogger {
private static final int NUM_ELEMENTS = 5;
private static final Network network = Network.newNetwork();

@ClassRule public static final Timeout TIMEOUT = new Timeout(10, TimeUnit.MINUTES);

@ClassRule
@Container
public static LocalstackContainer mockFirehoseContainer =
new LocalstackContainer(DockerImageName.parse(DockerImageVersions.LOCALSTACK))
.withNetwork(network)
Expand All @@ -117,7 +117,7 @@ public class KinesisFirehoseTableITTest extends TestLogger {
public static final FlinkContainers FLINK =
FlinkContainers.builder().withTestcontainersSettings(TESTCONTAINERS_SETTINGS).build();

@Before
@BeforeEach
public void setup() throws Exception {
System.setProperty(SdkSystemSetting.CBOR_ENABLED.property(), "false");

Expand All @@ -139,17 +139,17 @@ public void setup() throws Exception {
LOG.info("Done setting up the localstack.");
}

@BeforeClass
@BeforeAll
public static void setupFlink() throws Exception {
FLINK.start();
}

@AfterClass
@AfterAll
public static void stopFlink() {
FLINK.stop();
}

@After
@AfterEach
public void teardown() {
System.clearProperty(SdkSystemSetting.CBOR_ENABLED.property());

Expand Down
24 changes: 11 additions & 13 deletions ...src/test/java/org/apache/flink/connector/kinesis/table/test/KinesisStreamsTableApiIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,14 +36,13 @@
import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import com.google.common.collect.ImmutableList;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.rules.Timeout;
import org.rnorth.ducttape.ratelimits.RateLimiter;
import org.rnorth.ducttape.ratelimits.RateLimiterBuilder;
import org.slf4j.Logger;
Expand Down Expand Up @@ -79,6 +78,7 @@
/** End-to-end test for Kinesis Streams Table API Sink using Kinesalite. */
@Testcontainers
@ExtendWith(MiniClusterExtension.class)
@Timeout(value = 10, unit = TimeUnit.MINUTES)
public class KinesisStreamsTableApiIT {

private static final Logger LOGGER = LoggerFactory.getLogger(KinesisStreamsTableApiIT.class);
Expand All @@ -96,8 +96,6 @@ public class KinesisStreamsTableApiIT {
ResourceTestUtils.getResource(".*kinesis-streams.jar");
private static final Network network = Network.newNetwork();

@ClassRule public static final Timeout TIMEOUT = new Timeout(10, TimeUnit.MINUTES);

@Container
public static final LocalstackContainer LOCALSTACK_CONTAINER =
new LocalstackContainer(DockerImageName.parse(LOCALSTACK_DOCKER_IMAGE_VERSION))
Expand All @@ -123,17 +121,17 @@ public class KinesisStreamsTableApiIT {
public static final FlinkContainers FLINK =
FlinkContainers.builder().withTestcontainersSettings(TESTCONTAINERS_SETTINGS).build();

@BeforeClass
@BeforeAll
public static void setupFlink() throws Exception {
FLINK.start();
}

@AfterClass
@AfterAll
public static void stopFlink() {
FLINK.stop();
}

@Before
@BeforeEach
public void setUp() throws Exception {
System.setProperty(SdkSystemSetting.CBOR_ENABLED.property(), "false");
httpClient = AWSServicesTestUtils.createHttpClient();
Expand All @@ -144,7 +142,7 @@ public void setUp() throws Exception {
prepareStream(LARGE_ORDERS_STREAM);
}

@After
@AfterEach
public void teardown() {
System.clearProperty(SdkSystemSetting.CBOR_ENABLED.property());
AWSGeneralUtil.closeResources(httpClient, kinesisClient);
Expand Down
27 changes: 14 additions & 13 deletions ...s-sqs-e2e-tests/src/test/java/org/apache/flink/connector/sqs/sink/test/SqsSinkITTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,18 +28,18 @@
import org.apache.flink.connector.testframe.container.TestcontainersSettings;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.DockerImageVersions;
import org.apache.flink.util.TestLogger;

import org.assertj.core.api.Assertions;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.containers.Network;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;
import software.amazon.awssdk.core.SdkSystemSetting;
import software.amazon.awssdk.http.SdkHttpClient;
Expand All @@ -55,7 +55,8 @@
import static org.junit.jupiter.api.Assertions.assertEquals;

/** End to End test for SQS sink API. */
public class SqsSinkITTest extends TestLogger {
@Testcontainers
public class SqsSinkITTest {

private static final Logger LOG = LoggerFactory.getLogger(SqsSinkITTest.class);

Expand All @@ -65,7 +66,7 @@ public class SqsSinkITTest extends TestLogger {
private SqsClient sqsClient;
private static final Network network = Network.newNetwork();

@ClassRule
@Container
public static LocalstackContainer mockSqsContainer =
new LocalstackContainer(DockerImageName.parse(DockerImageVersions.LOCALSTACK))
.withNetwork(network)
Expand All @@ -85,7 +86,7 @@ public class SqsSinkITTest extends TestLogger {
public static final FlinkContainers FLINK =
FlinkContainers.builder().withTestcontainersSettings(TESTCONTAINERS_SETTINGS).build();

@Before
@BeforeEach
public void setup() throws Exception {
httpClient = AWSServicesTestUtils.createHttpClient();
sqsClient = createSqsClient(mockSqsContainer.getEndpoint(), httpClient);
Expand All @@ -94,17 +95,17 @@ public void setup() throws Exception {
LOG.info("Done setting up the localstack.");
}

@BeforeClass
@BeforeAll
public static void setupFlink() throws Exception {
FLINK.start();
}

@AfterClass
@AfterAll
public static void stopFlink() {
FLINK.stop();
}

@After
@AfterEach
public void teardown() {
System.clearProperty(SdkSystemSetting.CBOR_ENABLED.property());
httpClient.close();
Expand Down
39 changes: 20 additions & 19 deletions .../java/org/apache/flink/glue/schema/registry/test/GlueSchemaRegistryAvroKinesisITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,20 +33,20 @@
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.StringUtils;
import org.apache.flink.util.TestLogger;

import com.amazonaws.services.schemaregistry.utils.AWSSchemaRegistryConstants;
import com.amazonaws.services.schemaregistry.utils.AvroRecordType;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.junit.After;
import org.junit.Assume;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assumptions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
Expand All @@ -73,7 +73,8 @@
import static org.assertj.core.api.Assertions.assertThat;

/** End-to-end test for Glue Schema Registry AVRO format using Localstack. */
public class GlueSchemaRegistryAvroKinesisITCase extends TestLogger {
@Testcontainers
public class GlueSchemaRegistryAvroKinesisITCase {
private static final Logger LOGGER =
LoggerFactory.getLogger(GlueSchemaRegistryAvroKinesisITCase.class);

Expand All @@ -95,21 +96,21 @@ public class GlueSchemaRegistryAvroKinesisITCase extends TestLogger {
private KinesisClient kinesisClient;
private GSRKinesisPubsubClient gsrKinesisPubsubClient;

@ClassRule
@Container
public static LocalstackContainer mockKinesisContainer =
new LocalstackContainer(DockerImageName.parse(LOCALSTACK_DOCKER_IMAGE_VERSION))
.withNetworkAliases("localstack");

@Before
@BeforeEach
public void setup() throws Exception {
System.setProperty(SdkSystemSetting.CBOR_ENABLED.property(), "false");

Assume.assumeTrue(
"Access key not configured, skipping test...",
!StringUtils.isNullOrWhitespaceOnly(ACCESS_KEY));
Assume.assumeTrue(
"Secret key not configured, skipping test...",
!StringUtils.isNullOrWhitespaceOnly(SECRET_KEY));
Assumptions.assumeTrue(
!StringUtils.isNullOrWhitespaceOnly(ACCESS_KEY),
"Access key not configured, skipping test...");
Assumptions.assumeTrue(
!StringUtils.isNullOrWhitespaceOnly(SECRET_KEY),
"Secret key not configured, skipping test...");

StaticCredentialsProvider gsrCredentialsProvider =
StaticCredentialsProvider.create(
Expand All @@ -128,7 +129,7 @@ public void setup() throws Exception {
LOGGER.info("Done setting up the localstack.");
}

@After
@AfterEach
public void teardown() {
System.clearProperty(SdkSystemSetting.CBOR_ENABLED.property());
AWSGeneralUtil.closeResources(httpClient, kinesisClient);
Expand All @@ -142,7 +143,7 @@ public void testGSRJsonGenericFormatWithFlink() throws Exception {
for (GenericRecord msg : messages) {
gsrKinesisPubsubClient.sendMessage(getSchema().toString(), INPUT_STREAM, msg);
}
log.info("generated records");
LOGGER.info("generated records");

DataStream<GenericRecord> input =
env.fromSource(createSource(), WatermarkStrategy.noWatermarks(), "source")
Expand All @@ -155,11 +156,11 @@ public void testGSRJsonGenericFormatWithFlink() throws Exception {
List<Object> results =
gsrKinesisPubsubClient.readAllMessages(OUTPUT_STREAM, OUTPUT_STREAM_ARN);
while (deadline.hasTimeLeft() && results.size() < messages.size()) {
log.info("waiting for results..");
LOGGER.info("waiting for results..");
Thread.sleep(1000);
results = gsrKinesisPubsubClient.readAllMessages(OUTPUT_STREAM, OUTPUT_STREAM_ARN);
}
log.info("results: {}", results);
LOGGER.info("results: {}", results);

assertThat(results).containsExactlyInAnyOrderElementsOf(messages);
}
Expand Down
Loading
Loading