-
Notifications
You must be signed in to change notification settings - Fork 32
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement TestContainers Unit Tests with CaptureProxy with the KafkaC…
…onfigurationProxyTest Signed-off-by: Andre Kurait <[email protected]>
- Loading branch information
1 parent
ffbb46e
commit a9d5d4e
Showing
6 changed files
with
431 additions
and
5 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
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
133 changes: 133 additions & 0 deletions
133
...test/java/org/opensearch/migrations/trafficcapture/proxyserver/CaptureProxyContainer.java
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,133 @@ | ||
| package org.opensearch.migrations.trafficcapture.proxyserver; | ||
|
|
||
| import com.github.dockerjava.api.command.InspectContainerResponse; | ||
| import java.time.Duration; | ||
| import java.util.List; | ||
| import java.util.Set; | ||
| import java.util.function.Supplier; | ||
| import java.util.stream.Collectors; | ||
| import lombok.AllArgsConstructor; | ||
| import lombok.extern.slf4j.Slf4j; | ||
| import org.opensearch.migrations.testutils.PortFinder; | ||
| import org.testcontainers.containers.Container; | ||
| import org.testcontainers.containers.GenericContainer; | ||
| import org.testcontainers.containers.KafkaContainer; | ||
| import org.testcontainers.containers.wait.strategy.HttpWaitStrategy; | ||
| import org.testcontainers.containers.wait.strategy.WaitStrategyTarget; | ||
| import org.testcontainers.lifecycle.Startable; | ||
|
|
||
| @Slf4j | ||
| public class CaptureProxyContainer | ||
| extends GenericContainer implements AutoCloseable, WaitStrategyTarget, Startable { | ||
|
|
||
| private static final Duration TIMEOUT_DURATION = Duration.ofSeconds(30); | ||
| private final Supplier<String> destinationUriSupplier; | ||
| private final Supplier<String> kafkaUriSupplier; | ||
| private Integer listeningPort; | ||
| private Thread serverThread; | ||
|
|
||
| public CaptureProxyContainer(final Supplier<String> destinationUriSupplier, | ||
| final Supplier<String> kafkaUriSupplier) { | ||
| this.destinationUriSupplier = destinationUriSupplier; | ||
| this.kafkaUriSupplier = kafkaUriSupplier; | ||
| } | ||
|
|
||
| public CaptureProxyContainer(final String destinationUri, final String kafkaUri) { | ||
| this.destinationUriSupplier = () -> destinationUri; | ||
| this.kafkaUriSupplier = () -> kafkaUri; | ||
| } | ||
|
|
||
| public CaptureProxyContainer(final Container<?> destination, final KafkaContainer kafka) { | ||
| this(() -> getUriFromContainer(destination), () -> getUriFromContainer(kafka)); | ||
| } | ||
|
|
||
| public static String getUriFromContainer(final Container<?> container) { | ||
| return "http://" + container.getHost() + ":" + container.getFirstMappedPort(); | ||
| } | ||
|
|
||
| @Override | ||
| public void start() { | ||
| this.listeningPort = PortFinder.findOpenPort(); | ||
| serverThread = new Thread(() -> { | ||
| try { | ||
| String[] args = { | ||
| "--kafkaConnection", kafkaUriSupplier.get(), | ||
| "--destinationUri", destinationUriSupplier.get(), | ||
| "--listenPort", String.valueOf(listeningPort), | ||
| "--insecureDestination" | ||
| }; | ||
|
|
||
| CaptureProxy.main(args); | ||
| } catch (Exception e) { | ||
| throw new AssertionError("Should not have exception", e); | ||
| } | ||
| }); | ||
|
|
||
| serverThread.start(); | ||
| new HttpWaitStrategy().forPort(listeningPort) | ||
| .withStartupTimeout(TIMEOUT_DURATION) | ||
| .waitUntilReady(this); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isRunning() { | ||
| return serverThread != null; | ||
| } | ||
|
|
||
| @Override | ||
| public void stop() { | ||
| if (serverThread != null) { | ||
| serverThread.interrupt(); | ||
| this.serverThread = null; | ||
| } | ||
| this.listeningPort = null; | ||
| close(); | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| } | ||
|
|
||
| @Override | ||
| public Set<Integer> getLivenessCheckPortNumbers() { | ||
| return getExposedPorts() | ||
| .stream() | ||
| .map(this::getMappedPort) | ||
| .collect(Collectors.toSet()); | ||
| } | ||
|
|
||
| @Override | ||
| public String getHost() { | ||
| return "localhost"; | ||
| } | ||
|
|
||
| @Override | ||
| public Integer getMappedPort(int originalPort) { | ||
| if (getExposedPorts().contains(originalPort)) { | ||
| return listeningPort; | ||
| } | ||
| return null; | ||
| } | ||
|
|
||
| @Override | ||
| public List<Integer> getExposedPorts() { | ||
| // Internal and External ports are the same | ||
| return List.of(listeningPort); | ||
| } | ||
|
|
||
| @Override | ||
| public InspectContainerResponse getContainerInfo() { | ||
| return new InspectNonContainerResponse("captureProxy"); | ||
| } | ||
|
|
||
| @AllArgsConstructor | ||
| static class InspectNonContainerResponse extends InspectContainerResponse { | ||
|
|
||
| private String name; | ||
|
|
||
| @Override | ||
| public String getName() { | ||
| return name; | ||
| } | ||
| } | ||
| } |
84 changes: 84 additions & 0 deletions
84
...src/test/java/org/opensearch/migrations/trafficcapture/proxyserver/ContainerTestBase.java
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,84 @@ | ||
| package org.opensearch.migrations.trafficcapture.proxyserver; | ||
|
|
||
| import eu.rekawek.toxiproxy.Proxy; | ||
| import eu.rekawek.toxiproxy.ToxiproxyClient; | ||
| import java.io.IOException; | ||
| import org.junit.jupiter.api.AfterEach; | ||
| import org.junit.jupiter.api.BeforeEach; | ||
| import org.testcontainers.containers.GenericContainer; | ||
| import org.testcontainers.containers.KafkaContainer; | ||
| import org.testcontainers.containers.ToxiproxyContainer; | ||
| import org.testcontainers.junit.jupiter.Testcontainers; | ||
| import org.testcontainers.lifecycle.Startables; | ||
| import org.testcontainers.utility.DockerImageName; | ||
|
|
||
| @Testcontainers(disabledWithoutDocker = true) | ||
| public class ContainerTestBase { | ||
|
|
||
| private final ToxiproxyContainer toxiproxy = new ToxiproxyContainer( | ||
| "ghcr.io/shopify/toxiproxy:latest").withAccessToHost(true); | ||
| private final int DESTINATION_PROXY_PORT = 8666; | ||
| private final int KAFKA_PROXY_PORT = 8667; | ||
| public Proxy kafkaProxy = null; | ||
| public Proxy destinationProxy = null; | ||
| public String kafkaProxyUrl; | ||
| public String destinationProxyUrl; | ||
|
|
||
| @BeforeEach | ||
| public void setUp() throws IOException { | ||
|
|
||
| var kafkaHostPort = ReusableContainerTestSetup.kafka.getFirstMappedPort(); | ||
| var destinationHostPort = ReusableContainerTestSetup.destination.getFirstMappedPort(); | ||
|
|
||
| toxiproxy.start(); | ||
|
|
||
| org.testcontainers.Testcontainers.exposeHostPorts(kafkaHostPort); | ||
| org.testcontainers.Testcontainers.exposeHostPorts(destinationHostPort); | ||
|
|
||
|
|
||
| final ToxiproxyClient toxiproxyClient = new ToxiproxyClient(toxiproxy.getHost(), | ||
| toxiproxy.getControlPort()); | ||
|
|
||
| kafkaProxy = toxiproxyClient.createProxy("kafka", "0.0.0.0:" + KAFKA_PROXY_PORT, | ||
| "host.testcontainers.internal" + ":" + kafkaHostPort); | ||
| destinationProxy = toxiproxyClient.createProxy("destination", | ||
| "0.0.0.0:" + DESTINATION_PROXY_PORT, | ||
| "host.testcontainers.internal" + ":" + destinationHostPort); | ||
|
|
||
| kafkaProxyUrl = | ||
| "http://" + toxiproxy.getHost() + ":" + toxiproxy.getMappedPort(KAFKA_PROXY_PORT); | ||
| destinationProxyUrl = | ||
| "http://" + toxiproxy.getHost() + ":" + toxiproxy.getMappedPort(DESTINATION_PROXY_PORT); | ||
|
|
||
| kafkaProxy.enable(); | ||
| destinationProxy.enable(); | ||
| } | ||
|
|
||
| @AfterEach | ||
| public void tearDown() { | ||
| if (kafkaProxy != null) { | ||
| try { | ||
| kafkaProxy.delete(); | ||
| } catch (IOException ignored) {} | ||
| kafkaProxy = null; | ||
| } | ||
| if (destinationProxy != null) { | ||
| try { | ||
| destinationProxy.delete(); | ||
| } catch (IOException ignored) {} | ||
| destinationProxy = null; | ||
| } | ||
| } | ||
|
|
||
| public static class ReusableContainerTestSetup { | ||
|
|
||
| static private final KafkaContainer kafka = new KafkaContainer( | ||
| DockerImageName.parse("confluentinc/cp-kafka:latest")); | ||
| static private final GenericContainer<?> destination = new GenericContainer( | ||
| "httpd:alpine").withExposedPorts(80); // Container Port | ||
|
|
||
| static { | ||
| Startables.deepStart(kafka, destination).join(); | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.