Validate source/destination definition ids before create connection -…

… issue 144 (airbytehq#4115) * ConnectionsHandler checks that sourceDefinitionId and destinationDefinitionId are valid before creating a connector. * Whitespace fixes Co-authored-by: Jenny Brown <[email protected]>
tyschroed · Jun 18, 2021 · 39d4221 · 39d4221
1 parent 397aaa9
commit 39d4221
Show file tree

Hide file tree

Showing 3 changed files with 64 additions and 8 deletions.
diff --git a/...nfig/persistence/src/main/java/io/airbyte/config/persistence/ConfigNotFoundException.java b/...nfig/persistence/src/main/java/io/airbyte/config/persistence/ConfigNotFoundException.java
@@ -25,6 +25,7 @@
 package io.airbyte.config.persistence;
 
 import io.airbyte.config.ConfigSchema;
+import java.util.UUID;
 
 public class ConfigNotFoundException extends Exception {
 
@@ -37,6 +38,10 @@ public ConfigNotFoundException(ConfigSchema type, String configId) {
     this.configId = configId;
   }
 
+  public ConfigNotFoundException(ConfigSchema type, UUID uuid) {
+    this(type, uuid.toString());
+  }
+
   public ConfigSchema getType() {
     return type;
   }

diff --git a/airbyte-server/src/main/java/io/airbyte/server/handlers/ConnectionsHandler.java b/airbyte-server/src/main/java/io/airbyte/server/handlers/ConnectionsHandler.java
@@ -109,6 +109,10 @@ public ConnectionRead createConnection(ConnectionCreate connectionCreate) throws
       standardSync.withManual(true);
     }
 
+    // Validate source and destination
+    configRepository.getSourceConnection(connectionCreate.getSourceId());
+    configRepository.getDestinationConnection(connectionCreate.getDestinationId());
+
     configRepository.writeStandardSync(standardSync);
 
     trackNewConnection(standardSync);

diff --git a/airbyte-server/src/test/java/io/airbyte/server/handlers/ConnectionsHandlerTest.java b/airbyte-server/src/test/java/io/airbyte/server/handlers/ConnectionsHandlerTest.java
@@ -24,8 +24,7 @@
 
 package io.airbyte.server.handlers;
 
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.*;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
@@ -45,12 +44,7 @@
 import io.airbyte.api.model.SyncMode;
 import io.airbyte.api.model.WorkspaceIdRequestBody;
 import io.airbyte.commons.enums.Enums;
-import io.airbyte.config.DataType;
-import io.airbyte.config.Schedule;
-import io.airbyte.config.SourceConnection;
-import io.airbyte.config.StandardDestinationDefinition;
-import io.airbyte.config.StandardSourceDefinition;
-import io.airbyte.config.StandardSync;
+import io.airbyte.config.*;
 import io.airbyte.config.persistence.ConfigNotFoundException;
 import io.airbyte.config.persistence.ConfigRepository;
 import io.airbyte.protocol.models.ConfiguredAirbyteCatalog;
@@ -120,6 +114,59 @@ void testCreateConnection() throws JsonValidationException, ConfigNotFoundExcept
     verify(configRepository).writeStandardSync(standardSync);
   }
 
+  @Test
+  void testCreateConnectionWithBadDefinitionIds() throws JsonValidationException, ConfigNotFoundException, IOException {
+    when(uuidGenerator.get()).thenReturn(standardSync.getConnectionId());
+    UUID sourceDefinitionIdBad = UUID.randomUUID();
+    UUID destinationDefinitionIdBad = UUID.randomUUID();
+
+    final StandardSourceDefinition sourceDefinition = new StandardSourceDefinition()
+        .withName("source-test")
+        .withSourceDefinitionId(UUID.randomUUID());
+    final StandardDestinationDefinition destinationDefinition = new StandardDestinationDefinition()
+        .withName("destination-test")
+        .withDestinationDefinitionId(UUID.randomUUID());
+    when(configRepository.getStandardSync(standardSync.getConnectionId())).thenReturn(standardSync);
+    when(configRepository.getSourceDefinitionFromConnection(standardSync.getConnectionId())).thenReturn(sourceDefinition);
+    when(configRepository.getDestinationDefinitionFromConnection(standardSync.getConnectionId())).thenReturn(destinationDefinition);
+
+    when(configRepository.getSourceConnection(sourceDefinitionIdBad))
+        .thenThrow(new ConfigNotFoundException(ConfigSchema.SOURCE_CONNECTION, sourceDefinitionIdBad));
+    when(configRepository.getDestinationConnection(destinationDefinitionIdBad))
+        .thenThrow(new ConfigNotFoundException(ConfigSchema.DESTINATION_CONNECTION, destinationDefinitionIdBad));
+
+    final AirbyteCatalog catalog = ConnectionHelpers.generateBasicApiCatalog();
+
+    final ConnectionCreate connectionCreateBadSource = new ConnectionCreate()
+        .sourceId(sourceDefinitionIdBad)
+        .destinationId(standardSync.getDestinationId())
+        .operationIds(standardSync.getOperationIds())
+        .name("presto to hudi")
+        .namespaceDefinition(NamespaceDefinitionType.SOURCE)
+        .namespaceFormat(null)
+        .prefix("presto_to_hudi")
+        .status(ConnectionStatus.ACTIVE)
+        .schedule(ConnectionHelpers.generateBasicConnectionSchedule())
+        .syncCatalog(catalog);
+
+    assertThrows(ConfigNotFoundException.class, () -> connectionsHandler.createConnection(connectionCreateBadSource));
+
+    final ConnectionCreate connectionCreateBadDestination = new ConnectionCreate()
+        .sourceId(standardSync.getSourceId())
+        .destinationId(destinationDefinitionIdBad)
+        .operationIds(standardSync.getOperationIds())
+        .name("presto to hudi")
+        .namespaceDefinition(NamespaceDefinitionType.SOURCE)
+        .namespaceFormat(null)
+        .prefix("presto_to_hudi")
+        .status(ConnectionStatus.ACTIVE)
+        .schedule(ConnectionHelpers.generateBasicConnectionSchedule())
+        .syncCatalog(catalog);
+
+    assertThrows(ConfigNotFoundException.class, () -> connectionsHandler.createConnection(connectionCreateBadDestination));
+
+  }
+
   @Test
   void testUpdateConnection() throws JsonValidationException, ConfigNotFoundException, IOException {
     final AirbyteCatalog catalog = ConnectionHelpers.generateBasicApiCatalog();