Fix oracle encrypt tests

airbytehq · Apr 30, 2022 · dba3264 · dba3264
1 parent 31b03aa
commit dba3264
Show file tree

Hide file tree

Showing 7 changed files with 42 additions and 15 deletions.
diff --git a/airbyte-db/lib/src/main/java/io/airbyte/db/jdbc/JdbcUtils.java b/airbyte-db/lib/src/main/java/io/airbyte/db/jdbc/JdbcUtils.java
@@ -29,17 +29,25 @@ public static String getFullyQualifiedTableName(final String schemaName, final S
   }
 
   public static Map<String, String> parseJdbcParameters(final JsonNode config, final String jdbcUrlParamsKey) {
+    return parseJdbcParameters(config, jdbcUrlParamsKey, "&");
+  }
+
+  public static Map<String, String> parseJdbcParameters(final JsonNode config, final String jdbcUrlParamsKey, final String delimiter) {
     if (config.has(jdbcUrlParamsKey)) {
-      return parseJdbcParameters(config.get(jdbcUrlParamsKey).asText());
+      return parseJdbcParameters(config.get(jdbcUrlParamsKey).asText(), delimiter);
     } else {
       return Maps.newHashMap();
     }
   }
 
   public static Map<String, String> parseJdbcParameters(final String jdbcPropertiesString) {
+    return parseJdbcParameters(jdbcPropertiesString, "&");
+  }
+
+  public static Map<String, String> parseJdbcParameters(final String jdbcPropertiesString, final String delimiter) {
     final Map<String, String> parameters = new HashMap<>();
     if (!jdbcPropertiesString.isBlank()) {
-      final String[] keyValuePairs = jdbcPropertiesString.split("&");
+      final String[] keyValuePairs = jdbcPropertiesString.split(delimiter);
       for (final String kv : keyValuePairs) {
         final String[] split = kv.split("=");
         if (split.length == 2) {

diff --git a/...tions/destination/oracle_strict_encrypt/OracleStrictEncryptDestinationAcceptanceTest.java b/...tions/destination/oracle_strict_encrypt/OracleStrictEncryptDestinationAcceptanceTest.java
@@ -178,8 +178,7 @@ public void testEncryption() throws SQLException {
             config.get("sid").asText()),
         "oracle.jdbc.driver.OracleDriver",
         JdbcUtils.parseJdbcParameters("oracle.net.encryption_client=REQUIRED;" +
-            "oracle.net.encryption_types_client=( "
-            + algorithm + " )"));
+            "oracle.net.encryption_types_client=( " + algorithm + " )", ";"));
 
     final String network_service_banner =
         "select network_service_banner from v$session_connect_info where sid in (select distinct sid from v$mystat)";
@@ -204,8 +203,7 @@ public void testCheckProtocol() throws SQLException {
             clone.get("sid").asText()),
         "oracle.jdbc.driver.OracleDriver",
         JdbcUtils.parseJdbcParameters("oracle.net.encryption_client=REQUIRED;" +
-            "oracle.net.encryption_types_client=( "
-            + algorithm + " )"));
+            "oracle.net.encryption_types_client=( " + algorithm + " )", ";"));
 
     final String network_service_banner = "SELECT sys_context('USERENV', 'NETWORK_PROTOCOL') as network_protocol FROM dual";
     final List<JsonNode> collect = database.unsafeQuery(network_service_banner).collect(Collectors.toList());

diff --git a/...ors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java b/...ors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java
@@ -295,12 +295,16 @@ public JdbcDatabase createDatabase(final JsonNode config) throws SQLException {
         jdbcConfig.get("jdbc_url").asText(),
         driverClass,
         streamingQueryConfigProvider,
-        JdbcUtils.parseJdbcParameters(jdbcConfig, "connection_properties"),
+        JdbcUtils.parseJdbcParameters(jdbcConfig, "connection_properties", getJdbcParameterDelimiter()),
         sourceOperations);
 
     quoteString = (quoteString == null ? database.getMetaData().getIdentifierQuoteString() : quoteString);
 
     return database;
   }
 
+  protected String getJdbcParameterDelimiter() {
+    return "&";
+  }
+
 }
diff --git a/.../testFixtures/java/io/airbyte/integrations/source/jdbc/test/JdbcSourceAcceptanceTest.java b/.../testFixtures/java/io/airbyte/integrations/source/jdbc/test/JdbcSourceAcceptanceTest.java
@@ -25,6 +25,7 @@
 import io.airbyte.db.jdbc.JdbcDatabase;
 import io.airbyte.db.jdbc.JdbcSourceOperations;
 import io.airbyte.db.jdbc.JdbcUtils;
+import io.airbyte.db.jdbc.streaming.AdaptiveStreamingQueryConfig;
 import io.airbyte.integrations.base.Source;
 import io.airbyte.integrations.source.jdbc.AbstractJdbcSource;
 import io.airbyte.integrations.source.relationaldb.models.DbState;
@@ -185,19 +186,25 @@ protected String primaryKeyClause(final List<String> columns) {
     return clause.toString();
   }
 
+  protected String getJdbcParameterDelimiter() {
+    return "&";
+  }
+
   public void setup() throws Exception {
     source = getSource();
     config = getConfig();
     final JsonNode jdbcConfig = getToDatabaseConfigFunction().apply(config);
 
     streamName = TABLE_NAME;
 
-    database = Databases.createJdbcDatabase(
+    database = Databases.createStreamingJdbcDatabase(
         jdbcConfig.get("username").asText(),
         jdbcConfig.has("password") ? jdbcConfig.get("password").asText() : null,
         jdbcConfig.get("jdbc_url").asText(),
         getDriverClass(),
-        JdbcUtils.parseJdbcParameters(jdbcConfig, "connection_properties"));
+        AdaptiveStreamingQueryConfig::new,
+        JdbcUtils.parseJdbcParameters(jdbcConfig, "connection_properties", getJdbcParameterDelimiter()),
+        JdbcUtils.getDefaultSourceOperations());
 
     if (supportsSchemas()) {
       createSchemas();

diff --git a/...te/integrations/source/oracle_strict_encrypt/OracleStrictEncryptSourceAcceptanceTest.java b/...te/integrations/source/oracle_strict_encrypt/OracleStrictEncryptSourceAcceptanceTest.java
@@ -61,8 +61,8 @@ protected void setupEnvironment(final TestDestinationEnv environment) throws Exc
             config.get("port").asText(),
             config.get("sid").asText()),
         "oracle.jdbc.driver.OracleDriver",
-        JdbcUtils.parseJdbcParameters("oracle.net.encryption_client=REQUIRED&" +
-            "oracle.net.encryption_types_client=( 3DES168 )"));
+        JdbcUtils.parseJdbcParameters("oracle.net.encryption_client=REQUIRED;" +
+            "oracle.net.encryption_types_client=( 3DES168 )", ";"));
 
     database.execute(connection -> {
       connection.createStatement().execute("CREATE USER JDBC_SPACE IDENTIFIED BY JDBC_SPACE DEFAULT TABLESPACE USERS QUOTA UNLIMITED ON USERS");

diff --git a/...ntegrations/source/oracle_strict_encrypt/OracleStrictEncryptJdbcSourceAcceptanceTest.java b/...ntegrations/source/oracle_strict_encrypt/OracleStrictEncryptJdbcSourceAcceptanceTest.java
@@ -178,7 +178,12 @@ public void createSchemas() throws SQLException {
     }
   }
 
-  public void executeOracleStatement(final String query) throws SQLException {
+  @Override
+  protected String getJdbcParameterDelimiter() {
+    return ";";
+  }
+
+  public void executeOracleStatement(final String query) {
     try (final Connection conn = DriverManager.getConnection(
         ORACLE_DB.getJdbcUrl(),
         ORACLE_DB.getUsername(),
@@ -192,8 +197,8 @@ public void executeOracleStatement(final String query) throws SQLException {
 
   public static void logSQLException(final SQLException ex) {
     for (final Throwable e : ex) {
-      if (e instanceof SQLException) {
-        if (ignoreSQLException(((SQLException) e).getSQLState()) == false) {
+      if (e instanceof final SQLException sqlException) {
+        if (!ignoreSQLException(sqlException.getSQLState())) {
           LOGGER.info("SQLState: " + ((SQLException) e).getSQLState());
           LOGGER.info("Error Code: " + ((SQLException) e).getErrorCode());
           LOGGER.info("Message: " + e.getMessage());

diff --git a/...ctors/source-oracle/src/main/java/io/airbyte/integrations/source/oracle/OracleSource.java b/...ctors/source-oracle/src/main/java/io/airbyte/integrations/source/oracle/OracleSource.java
@@ -93,7 +93,7 @@ public JsonNode toDatabaseConfig(final JsonNode config) {
       }
     }
     if (!additionalParameters.isEmpty()) {
-      final String connectionParams = String.join(";", additionalParameters);
+      final String connectionParams = String.join(getJdbcParameterDelimiter(), additionalParameters);
       configBuilder.put("connection_properties", connectionParams);
     }
 
@@ -171,6 +171,11 @@ public Set<String> getExcludedInternalNameSpaces() {
     return Set.of();
   }
 
+  @Override
+  protected String getJdbcParameterDelimiter() {
+    return ";";
+  }
+
   public static void main(final String[] args) throws Exception {
     final Source source = OracleSource.sshWrappedSource();
     LOGGER.info("starting source: {}", OracleSource.class);