16250 Destination Redis: Add SSH support

airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/base/ssh/SshBastionContainer.java

@@ -78,6 +78,10 @@ public void stopAndCloseContainers(final JdbcDatabaseContainer<?> db) {
     db.close();
   }
 
+  public void stopAndClose() {
+    bastion.close();
+  }
+
   public GenericContainer getContainer() {
     return bastion;
   }

airbyte-integrations/connectors/destination-redis/src/main/java/io/airbyte/integrations/destination/redis/RedisDestination.java

@@ -9,9 +9,11 @@
 import io.airbyte.integrations.base.AirbyteMessageConsumer;
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.base.IntegrationRunner;
+import io.airbyte.integrations.base.ssh.SshWrappedDestination;
 import io.airbyte.protocol.models.AirbyteConnectionStatus;
 import io.airbyte.protocol.models.AirbyteMessage;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
+import java.util.List;
 import java.util.function.Consumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -21,7 +23,14 @@ class RedisDestination extends BaseConnector implements Destination {
   private static final Logger LOGGER = LoggerFactory.getLogger(RedisDestination.class);
 
   public static void main(String[] args) throws Exception {
-    new IntegrationRunner(new RedisDestination()).run(args);
+    LOGGER.info("starting destination: {}", RedisDestination.class);
+    final Destination destination = RedisDestination.sshWrappedDestination();
+    new IntegrationRunner(destination).run(args);
+    LOGGER.info("completed destination: {}", RedisDestination.class);
+  }
+
+  public static Destination sshWrappedDestination() {
+    return new SshWrappedDestination(new RedisDestination(), List.of("host"), List.of("port"));
   }
 
   @Override

airbyte-integrations/connectors/destination-redis/src/test-integration/java/io/airbyte/integrations/destination/redis/SshKeyRedisDestinationAcceptanceTest.java

@@ -0,0 +1,16 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.redis;
+
+import io.airbyte.integrations.base.ssh.SshTunnel;
+
+public class SshKeyRedisDestinationAcceptanceTest extends SshRedisDestinationAcceptanceTest {
+
+  @Override
+  public SshTunnel.TunnelMethod getTunnelMethod() {
+    return SshTunnel.TunnelMethod.SSH_KEY_AUTH;
+  }
+
+}

airbyte-integrations/connectors/destination-redis/src/test-integration/java/io/airbyte/integrations/destination/redis/SshPasswordRedisDestinationAcceptanceTest.java

@@ -0,0 +1,16 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.redis;
+
+import io.airbyte.integrations.base.ssh.SshTunnel;
+
+public class SshPasswordRedisDestinationAcceptanceTest extends SshRedisDestinationAcceptanceTest {
+
+  @Override
+  public SshTunnel.TunnelMethod getTunnelMethod() {
+    return SshTunnel.TunnelMethod.SSH_PASSWORD_AUTH;
+  }
+
+}

airbyte-integrations/connectors/destination-redis/src/test-integration/java/io/airbyte/integrations/destination/redis/SshRedisDestinationAcceptanceTest.java

@@ -0,0 +1,126 @@
+/*
+ * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.redis;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.integrations.base.ssh.SshBastionContainer;
+import io.airbyte.integrations.base.ssh.SshTunnel;
+import io.airbyte.integrations.destination.redis.RedisContainerInitializr.RedisContainer;
+import io.airbyte.integrations.standardtest.destination.DestinationAcceptanceTest;
+import io.airbyte.integrations.standardtest.destination.comparator.AdvancedTestDataComparator;
+import io.airbyte.integrations.standardtest.destination.comparator.TestDataComparator;
+import io.airbyte.integrations.util.HostPortResolver;
+import java.util.Comparator;
+import java.util.List;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.testcontainers.containers.Network;
+
+public abstract class SshRedisDestinationAcceptanceTest extends DestinationAcceptanceTest {
+
+  private static final SshBastionContainer bastion = new SshBastionContainer();
+  private static final Network network = Network.newNetwork();
+  private static RedisContainerInitializr.RedisContainer redisContainer;
+  private JsonNode configJson;
+  private RedisCache redisCache;
+  private RedisNameTransformer redisNameTransformer;
+
+  @BeforeAll
+  static void initContainers() {
+    redisContainer = new RedisContainer()
+        .withExposedPorts(6379)
+        .withNetwork(network);
+    redisContainer.start();
+    bastion.initAndStartBastion(network);
+  }
+
+  @AfterAll
+  static void stop() {
+    redisContainer.close();
+    bastion.stopAndClose();
+  }
+
+  public abstract SshTunnel.TunnelMethod getTunnelMethod();
+
+  @Override
+  protected void setup(TestDestinationEnv testEnv) {
+    configJson = RedisDataFactory.jsonConfig(
+        redisContainer.getHost(),
+        redisContainer.getFirstMappedPort());
+    var redisConfig = new RedisConfig(configJson);
+    redisCache = new RedisHCache(redisConfig);
+    redisNameTransformer = new RedisNameTransformer();
+  }
+
+  @Override
+  protected void tearDown(TestDestinationEnv testEnv) {
+    redisCache.flushAll();
+  }
+
+  @Override
+  protected String getImageName() {
+    return "airbyte/destination-redis:dev";
+  }
+
+  @Override
+  protected JsonNode getConfig() throws Exception {
+    return bastion.getTunnelConfig(getTunnelMethod(), ImmutableMap.builder()
+        .put("host", HostPortResolver.resolveIpAddress(redisContainer))
+        .put("port", redisContainer.getExposedPorts().get(0))
+        .put("username", configJson.get("username"))
+        .put("password", configJson.get("password"))
+        .put("cache_type", configJson.get("cache_type")));
+  }
+
+  @Override
+  protected JsonNode getFailCheckConfig() {
+    return RedisDataFactory.jsonConfig(
+        "127.0.0.9",
+        8080);
+  }
+
+  @Override
+  protected List<JsonNode> retrieveRecords(TestDestinationEnv testEnv,
+      String streamName,
+      String namespace,
+      JsonNode streamSchema) {
+    var key = redisNameTransformer.keyName(namespace, streamName);
+    return redisCache.getAll(key).stream()
+        .sorted(Comparator.comparing(RedisRecord::getTimestamp))
+        .map(RedisRecord::getData)
+        .map(Jsons::deserialize)
+        .collect(Collectors.toList());
+  }
+
+
+  @Override
+  protected boolean implementsNamespaces() {
+    return true;
+  }
+
+  @Override
+  protected TestDataComparator getTestDataComparator() {
+    return new AdvancedTestDataComparator();
+  }
+
+  @Override
+  protected boolean supportBasicDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportArrayDataTypeTest() {
+    return true;
+  }
+
+  @Override
+  protected boolean supportObjectDataTypeTest() {
+    return true;
+  }
+
+}