Handler Unit Tests and remaining SourceImplementation API endpoints (#48)

dataline-api/src/main/openapi/config.yaml

@@ -221,8 +221,8 @@ paths:
     post:
       tags:
       - source_implementation
-      summary: Get source implementations for workspace
-      operationId: getSourceImplementationsForWorkspace
+      summary: List source implementations for workspace
+      operationId: listSourceImplementationsForWorkspace
       requestBody:
         content:
           application/json:

dataline-config-persistence/build.gradle

@@ -1,4 +1,6 @@
 dependencies {
+    implementation group: 'commons-io', name: 'commons-io', version: '2.7'
+
     implementation group: "com.fasterxml.jackson.core", name: "jackson-databind", version: "2.9.8"
     implementation group: "com.networknt", name: "json-schema-validator", version: "1.0.42"
 

dataline-config-persistence/src/main/java/io/dataline/config/persistence/DefaultConfigPersistence.java

@@ -12,18 +12,20 @@ import java.util.HashSet;
 import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
+import org.apache.commons.io.FileUtils;
 
 // we force all interaction with disk storage to be effectively single threaded.
-public class ConfigPersistenceImpl implements ConfigPersistence {
+public class DefaultConfigPersistence implements ConfigPersistence {
   private static final Object lock = new Object();
-  private static final String CONFIG_STORAGE_ROOT = "data/config/";
-  private static final String CONFIG_SCHEMA_ROOT = "dataline-config/src/main/resources/json/";
+  private static final String CONFIG_SCHEMA_ROOT = "../dataline-config/src/main/resources/json/";
 
   private final ObjectMapper objectMapper;
-  final JsonSchemaValidation jsonSchemaValidation;
+  private final JsonSchemaValidation jsonSchemaValidation;
+  private final String storageRoot;
 
-  public ConfigPersistenceImpl() {
-    jsonSchemaValidation = JsonSchemaValidation.getInstance();
+  public DefaultConfigPersistence(String storageRoot) {
+    this.storageRoot = storageRoot;
+    jsonSchemaValidation = new JsonSchemaValidation();
     objectMapper = new ObjectMapper();
   }
 
@@ -70,8 +72,10 @@ public class ConfigPersistenceImpl implements ConfigPersistence {
   public <T> void writeConfig(
       PersistenceConfigType persistenceConfigType, String configId, T config) {
     synchronized (lock) {
+      final String configPath = getConfigPath(persistenceConfigType, configId);
+      ensureDirectory(getConfigDirectory(persistenceConfigType));
       try {
-        objectMapper.writeValue(new File(getConfigPath(persistenceConfigType, configId)), config);
+        objectMapper.writeValue(new File(configPath), config);
       } catch (IOException e) {
         throw new RuntimeException(e);
       }
@@ -99,7 +103,7 @@ public class ConfigPersistenceImpl implements ConfigPersistence {
   }
 
   private String getConfigDirectory(PersistenceConfigType persistenceConfigType) {
-    return String.format("%s/%s", CONFIG_STORAGE_ROOT, persistenceConfigType.toString());
+    return String.format("%s/%s", storageRoot, persistenceConfigType.toString());
   }
 
   private String getConfigPath(PersistenceConfigType persistenceConfigType, String configId) {
@@ -113,8 +117,8 @@ public class ConfigPersistenceImpl implements ConfigPersistence {
   }
 
   private Optional<Path> getFile(PersistenceConfigType persistenceConfigType, String id) {
-    String configPath = getConfigPath(persistenceConfigType, id);
-    final Path path = Paths.get(configPath);
+    ensureDirectory(getConfigDirectory(persistenceConfigType));
+    final Path path = Paths.get(getConfigPath(persistenceConfigType, id));
     if (Files.exists(path)) {
       return Optional.of(path);
     } else {
@@ -184,7 +188,10 @@ public class ConfigPersistenceImpl implements ConfigPersistence {
             () ->
                 new ConfigNotFoundException(
                     String.format(
-                        "config type: %s id: %s not found", persistenceConfigType, configId)));
+                        "config type: %s id: %s not found in path %s",
+                        persistenceConfigType,
+                        configId,
+                        getConfigPath(persistenceConfigType, configId))));
   }
 
   private <T> T fileToPojo(File file, Class<T> clazz) {
@@ -194,4 +201,12 @@ public class ConfigPersistenceImpl implements ConfigPersistence {
       throw new RuntimeException(e);
     }
   }
+
+  private void ensureDirectory(String path) {
+    try {
+      FileUtils.forceMkdir(new File(path));
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
 }

dataline-config-persistence/src/main/java/io/dataline/config/persistence/JsonSchemaValidation.java

@@ -9,13 +9,7 @@ public class JsonSchemaValidation {
   private final SchemaValidatorsConfig schemaValidatorsConfig;
   private final JsonSchemaFactory jsonSchemaFactory;
 
-  private static final JsonSchemaValidation INSTANCE = new JsonSchemaValidation();
-
-  public static JsonSchemaValidation getInstance() {
-    return INSTANCE;
-  }
-
-  private JsonSchemaValidation() {
+  public JsonSchemaValidation() {
     this.schemaValidatorsConfig = new SchemaValidatorsConfig();
     this.jsonSchemaFactory = JsonSchemaFactory.getInstance(SpecVersion.VersionFlag.V7);
   }

dataline-config-persistence/src/main/java/io/dataline/config/persistence/PersistenceConstants.java

@@ -2,7 +2,9 @@ package io.dataline.config.persistence;
 
 import java.util.UUID;
 
-public class WorkspaceConstants {
+public class PersistenceConstants {
+  public static String DEFAULT_TEST_ROOT = "/tmp/data/config/";
+
   // for MVP we only support one workspace per deployment and we hard code its id.
   public static UUID DEFAULT_WORKSPACE_ID = UUID.fromString("5ae6b09b-fdec-41af-aaf7-7d94cfc33ef6");
 }

dataline-config/src/main/resources/json/SourceConnectionImplementation.json

@@ -3,13 +3,17 @@
   "$id": "https://github.com/datalineio/dataline/blob/master/dataline-config/src/main/resources/json/SourceConnectionConfiguration.json",
   "title": "SourceConnectionConfiguration",
   "description": "information required for connection to a destination.",  "type": "object",
-  "required": ["sourceSpecificationId", "sourceImplementationId", "configuration"],
+  "required": ["sourceSpecificationId", "sourceImplementationId", "workspaceId","configuration"],
   "additionalProperties": false,
   "properties": {
     "sourceSpecificationId": {
       "type": "string",
       "format": "uuid"
     },
+    "workspaceId": {
+      "type": "string",
+      "format": "uuid"
+    },
     "sourceImplementationId": {
       "type": "string",
       "format": "uuid"

dataline-server/build.gradle

@@ -14,6 +14,7 @@ dependencies {
     implementation group: "com.fasterxml.jackson.core", name: "jackson-databind", version: "2.9.8"
     implementation group: "com.networknt", name: "json-schema-validator", version: "1.0.42"
 
+
     implementation project(':dataline-api')
     implementation project(':dataline-commons')
     implementation project(':dataline-config')

dataline-server/src/main/java/io/dataline/server/apis/ConfigurationApi.java

@@ -2,11 +2,12 @@ package io.dataline.server.apis;
 
 import io.dataline.api.model.*;
 import io.dataline.config.persistence.ConfigPersistence;
-import io.dataline.config.persistence.ConfigPersistenceImpl;
+import io.dataline.config.persistence.DefaultConfigPersistence;
 import io.dataline.server.handlers.SourceImplementationsHandler;
 import io.dataline.server.handlers.SourceSpecificationsHandler;
 import io.dataline.server.handlers.SourcesHandler;
 import io.dataline.server.handlers.WorkspacesHandler;
+import io.dataline.server.validation.IntegrationSchemaValidation;
 import javax.validation.Valid;
 import javax.ws.rs.Path;
 
@@ -18,11 +19,14 @@ public class ConfigurationApi implements io.dataline.api.V1Api {
   private final SourceImplementationsHandler sourceImplementationsHandler;
 
   public ConfigurationApi() {
-    ConfigPersistence configPersistence = new ConfigPersistenceImpl();
+    // todo: configure with env variable.
+    ConfigPersistence configPersistence = new DefaultConfigPersistence("../data/config/");
     workspacesHandler = new WorkspacesHandler(configPersistence);
     sourcesHandler = new SourcesHandler(configPersistence);
     sourceSpecificationsHandler = new SourceSpecificationsHandler(configPersistence);
-    sourceImplementationsHandler = new SourceImplementationsHandler(configPersistence);
+    sourceImplementationsHandler =
+        new SourceImplementationsHandler(
+            configPersistence, new IntegrationSchemaValidation(configPersistence));
   }
 
   // WORKSPACE
@@ -69,22 +73,17 @@ public class ConfigurationApi implements io.dataline.api.V1Api {
     return sourceImplementationsHandler.createSourceImplementation(sourceImplementationCreate);
   }
 
-  @Override
-  public SourceImplementationDiscoverSchemaRead discoverSchemaForSourceImplementation(
-      @Valid SourceImplementationIdRequestBody sourceImplementationIdRequestBody) {
-    return null;
-  }
-
   @Override
   public SourceImplementationRead getSourceImplementation(
       @Valid SourceImplementationIdRequestBody sourceImplementationIdRequestBody) {
-    return null;
+    return sourceImplementationsHandler.getSourceImplementation(sourceImplementationIdRequestBody);
   }
 
   @Override
-  public SourceImplementationReadList getSourceImplementationsForWorkspace(
+  public SourceImplementationReadList listSourceImplementationsForWorkspace(
       @Valid WorkspaceIdRequestBody workspaceIdRequestBody) {
-    return null;
+    return sourceImplementationsHandler.listSourceImplementationsForWorkspace(
+        workspaceIdRequestBody);
   }
 
   @Override
@@ -96,6 +95,12 @@ public class ConfigurationApi implements io.dataline.api.V1Api {
   @Override
   public SourceImplementationRead updateSourceImplementation(
       @Valid SourceImplementationUpdate sourceImplementationUpdate) {
+    return sourceImplementationsHandler.updateSourceImplementation(sourceImplementationUpdate);
+  }
+
+  @Override
+  public SourceImplementationDiscoverSchemaRead discoverSchemaForSourceImplementation(
+      @Valid SourceImplementationIdRequestBody sourceImplementationIdRequestBody) {
     return null;
   }
 

dataline-server/src/main/java/io/dataline/server/errors/CatchAllExceptionMapper.java

@@ -4,12 +4,16 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.ext.ExceptionMapper;
 import javax.ws.rs.ext.Provider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @Provider
 public class CatchAllExceptionMapper implements ExceptionMapper<Throwable> {
+  private static final Logger LOGGER = LoggerFactory.getLogger(CatchAllExceptionMapper.class);
 
   @Override
   public Response toResponse(Throwable e) {
+    LOGGER.debug("catch all exception", e);
     return Response.status(500)
         .entity(
             new ObjectMapper()

dataline-server/src/main/java/io/dataline/server/errors/KnownExceptionMapper.java

@@ -4,14 +4,18 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.ext.ExceptionMapper;
 import javax.ws.rs.ext.Provider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 @Provider
 public class KnownExceptionMapper implements ExceptionMapper<KnownException> {
+  private static final Logger LOGGER = LoggerFactory.getLogger(KnownExceptionMapper.class);
 
   @Override
   public Response toResponse(KnownException e) {
+    LOGGER.debug("Known exception", e);
     return Response.status(e.getHttpCode())
-        .entity(new ObjectMapper().createObjectNode().put("message: ", e.getMessage()))
+        .entity(new ObjectMapper().createObjectNode().put("message", e.getMessage()))
         .type("application/json")
         .build();
   }

dataline-server/src/main/java/io/dataline/server/handlers/SourceImplementationsHandler.java

@@ -1,7 +1,6 @@
 package io.dataline.server.handlers;
 
-import io.dataline.api.model.SourceImplementationCreate;
-import io.dataline.api.model.SourceImplementationRead;
+import io.dataline.api.model.*;
 import io.dataline.config.SourceConnectionImplementation;
 import io.dataline.config.persistence.ConfigNotFoundException;
 import io.dataline.config.persistence.ConfigPersistence;
@@ -9,48 +8,139 @@ import io.dataline.config.persistence.JsonValidationException;
 import io.dataline.config.persistence.PersistenceConfigType;
 import io.dataline.server.errors.KnownException;
 import io.dataline.server.validation.IntegrationSchemaValidation;
+import java.util.List;
 import java.util.UUID;
+import java.util.function.Supplier;
+import java.util.stream.Collectors;
 
 public class SourceImplementationsHandler {
-  private final ConfigPersistence configPersistence;
 
-  public SourceImplementationsHandler(ConfigPersistence configPersistence) {
+  private final Supplier<UUID> uuidGenerator;
+  private final ConfigPersistence configPersistence;
+  private final IntegrationSchemaValidation validator;
+
+  public SourceImplementationsHandler(
+      ConfigPersistence configPersistence,
+      IntegrationSchemaValidation integrationSchemaValidation,
+      Supplier<UUID> uuidGenerator) {
     this.configPersistence = configPersistence;
+    this.validator = integrationSchemaValidation;
+    this.uuidGenerator = uuidGenerator;
+  }
+
+  public SourceImplementationsHandler(
+      ConfigPersistence configPersistence,
+      IntegrationSchemaValidation integrationSchemaValidation) {
+    this(configPersistence, integrationSchemaValidation, UUID::randomUUID);
   }
 
   public SourceImplementationRead createSourceImplementation(
       SourceImplementationCreate sourceImplementationCreate) {
+    // validate configuration
+    validateSourceImplementation(
+        sourceImplementationCreate.getSourceSpecificationId(),
+        sourceImplementationCreate.getConnectionConfiguration());
+
+    // persist
+    final UUID sourceImplementationId = uuidGenerator.get();
+    persistSourceConnectionImplementation(
+        sourceImplementationCreate.getSourceSpecificationId(),
+        sourceImplementationCreate.getWorkspaceId(),
+        sourceImplementationId,
+        sourceImplementationCreate.getConnectionConfiguration());
+
+    // read configuration from db
+    return getSourceImplementationInternal(sourceImplementationId);
+  }
+
+  public SourceImplementationRead updateSourceImplementation(
+      SourceImplementationUpdate sourceImplementationUpdate) {
+    // get existing implementation
+    final SourceImplementationRead persistedSourceImplementation =
+        getSourceImplementationInternal(sourceImplementationUpdate.getSourceImplementationId());
+
+    // validate configuration
+    validateSourceImplementation(
+        persistedSourceImplementation.getSourceSpecificationId(),
+        sourceImplementationUpdate.getConnectionConfiguration());
+
+    // persist
+    persistSourceConnectionImplementation(
+        persistedSourceImplementation.getSourceSpecificationId(),
+        persistedSourceImplementation.getWorkspaceId(),
+        sourceImplementationUpdate.getSourceImplementationId(),
+        sourceImplementationUpdate.getConnectionConfiguration());
+
+    // read configuration from db
+    return getSourceImplementationInternal(sourceImplementationUpdate.getSourceImplementationId());
+  }
+
+  public SourceImplementationRead getSourceImplementation(
+      SourceImplementationIdRequestBody sourceImplementationIdRequestBody) {
+
+    return getSourceImplementationInternal(
+        sourceImplementationIdRequestBody.getSourceImplementationId());
+  }
+
+  public SourceImplementationReadList listSourceImplementationsForWorkspace(
+      WorkspaceIdRequestBody workspaceIdRequestBody) {
     try {
-      // validate configuration
-      final IntegrationSchemaValidation validator =
-          new IntegrationSchemaValidation(configPersistence);
-      validator.validateSourceConnectionConfiguration(
-          sourceImplementationCreate.getSourceSpecificationId(),
-          sourceImplementationCreate.getConnectionConfiguration());
 
-      // persist
-      final UUID sourceImplementationId = UUID.randomUUID();
-      final SourceConnectionImplementation newSourceConnectionImplementation =
-          new SourceConnectionImplementation();
-      newSourceConnectionImplementation.setSourceSpecificationId(
-          sourceImplementationCreate.getSourceSpecificationId());
-      newSourceConnectionImplementation.setSourceImplementationId(sourceImplementationId);
-      newSourceConnectionImplementation.setConfiguration(
-          sourceImplementationCreate.getConnectionConfiguration());
+      final List<SourceImplementationRead> reads =
+          configPersistence
+              .getConfigs(
+                  PersistenceConfigType.SOURCE_CONNECTION_IMPLEMENTATION,
+                  SourceConnectionImplementation.class)
+              .stream()
+              .filter(
+                  sourceConnectionImplementation ->
+                      sourceConnectionImplementation
+                          .getWorkspaceId()
+                          .equals(workspaceIdRequestBody.getWorkspaceId()))
+              .map(this::toSourceImplementationRead)
+              .collect(Collectors.toList());
 
-      configPersistence.writeConfig(
-          PersistenceConfigType.SOURCE_CONNECTION_IMPLEMENTATION,
-          sourceImplementationId.toString(),
-          newSourceConnectionImplementation);
+      final SourceImplementationReadList sourceImplementationReadList =
+          new SourceImplementationReadList();
+      sourceImplementationReadList.setSources(reads);
+      return sourceImplementationReadList;
+    } catch (JsonValidationException e) {
+      throw new KnownException(
+          422,
+          String.format(
+              "Attempted to retrieve a configuration does not fulfill the specification. Errors: %s",
+              e.getMessage()));
+    }
+  }
 
-      // read configuration from db
-      final SourceConnectionImplementation retrievedSourceConnectionImplementation =
+  private SourceImplementationRead getSourceImplementationInternal(UUID sourceImplementationId) {
+    // read configuration from db
+    final SourceConnectionImplementation retrievedSourceConnectionImplementation;
+    try {
+      retrievedSourceConnectionImplementation =
           configPersistence.getConfig(
               PersistenceConfigType.SOURCE_CONNECTION_IMPLEMENTATION,
               sourceImplementationId.toString(),
               SourceConnectionImplementation.class);
+    } catch (JsonValidationException e) {
+      throw new KnownException(
+          422,
+          String.format(
+              "The provided configuration does not fulfill the specification. Errors: %s",
+              e.getMessage()));
+    } catch (ConfigNotFoundException e) {
+      throw new KnownException(
+          422, String.format("Could not find source specification: %s.", sourceImplementationId));
+    }
 
-      return toSourceImplementationRead(retrievedSourceConnectionImplementation);
+    return toSourceImplementationRead(retrievedSourceConnectionImplementation);
+  }
+
+  private void validateSourceImplementation(
+      UUID sourceConnectionSpecificationId, Object implementation) {
+    try {
+      validator.validateSourceConnectionConfiguration(
+          sourceConnectionSpecificationId, implementation);
     } catch (JsonValidationException e) {
       throw new KnownException(
           422,
@@ -61,19 +151,37 @@ public class SourceImplementationsHandler {
       throw new KnownException(
           422,
           String.format(
-              "Could not find source specification: %s.",
-              sourceImplementationCreate.getSourceSpecificationId()));
+              "Could not find source specification: %s.", sourceConnectionSpecificationId));
     }
   }
 
+  private void persistSourceConnectionImplementation(
+      UUID sourceSpecificationId,
+      UUID workspaceId,
+      UUID sourceImplementationId,
+      Object configuration) {
+    final SourceConnectionImplementation sourceConnectionImplementation =
+        new SourceConnectionImplementation();
+    sourceConnectionImplementation.setSourceSpecificationId(sourceSpecificationId);
+    sourceConnectionImplementation.setWorkspaceId(workspaceId);
+    sourceConnectionImplementation.setSourceImplementationId(sourceImplementationId);
+    sourceConnectionImplementation.setConfiguration(configuration);
+
+    configPersistence.writeConfig(
+        PersistenceConfigType.SOURCE_CONNECTION_IMPLEMENTATION,
+        sourceImplementationId.toString(),
+        sourceConnectionImplementation);
+  }
+
   private SourceImplementationRead toSourceImplementationRead(
       SourceConnectionImplementation sourceConnectionImplementation) {
     final SourceImplementationRead sourceImplementationRead = new SourceImplementationRead();
-    sourceConnectionImplementation.setSourceImplementationId(
+    sourceImplementationRead.setSourceImplementationId(
         sourceConnectionImplementation.getSourceImplementationId());
-    sourceConnectionImplementation.setSourceSpecificationId(
+    sourceImplementationRead.setWorkspaceId(sourceConnectionImplementation.getWorkspaceId());
+    sourceImplementationRead.setSourceSpecificationId(
         sourceConnectionImplementation.getSourceSpecificationId());
-    sourceConnectionImplementation.setConfiguration(
+    sourceImplementationRead.setConnectionConfiguration(
         sourceConnectionImplementation.getConfiguration());
 
     return sourceImplementationRead;

dataline-server/src/main/java/io/dataline/server/handlers/WorkspacesHandler.java

@@ -23,7 +23,7 @@ public class WorkspacesHandler {
   @SuppressWarnings("unused")
   public WorkspaceRead getWorkspaceBySlug(SlugRequestBody slugRequestBody) {
     // for now we assume there is one workspace and it has a default uuid.
-    return getWorkspaceFromId(WorkspaceConstants.DEFAULT_WORKSPACE_ID);
+    return getWorkspaceFromId(PersistenceConstants.DEFAULT_WORKSPACE_ID);
   }
 
   private WorkspaceRead getWorkspaceFromId(UUID workspaceIdUuid) {
@@ -69,6 +69,8 @@ public class WorkspacesHandler {
     persistedWorkspace.setAnonymousDataCollection(workspaceUpdate.getAnonymousDataCollection());
     persistedWorkspace.setNews(workspaceUpdate.getNews());
     persistedWorkspace.setSecurityUpdates(workspaceUpdate.getSecurityUpdates());
+    configPersistence.writeConfig(
+        PersistenceConfigType.STANDARD_WORKSPACE, workspaceId, persistedWorkspace);
 
     return getWorkspaceFromId(workspaceUpdate.getWorkspaceId());
   }

dataline-server/src/main/java/io/dataline/server/validation/IntegrationSchemaValidation.java

@@ -16,7 +16,7 @@ public class IntegrationSchemaValidation {
     this.configPersistence = configPersistence;
 
     this.objectMapper = new ObjectMapper();
-    jsonSchemaValidation = JsonSchemaValidation.getInstance();
+    this.jsonSchemaValidation = new JsonSchemaValidation();
   }
 
   public void validateSourceConnectionConfiguration(

dataline-server/src/test/java/io/dataline/server/handlers/SourceImplementationsHandlerTest.java

@@ -0,0 +1,231 @@
+package io.dataline.server.handlers;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.Sets;
+import io.dataline.api.model.*;
+import io.dataline.config.SourceConnectionImplementation;
+import io.dataline.config.SourceConnectionSpecification;
+import io.dataline.config.persistence.ConfigNotFoundException;
+import io.dataline.config.persistence.ConfigPersistence;
+import io.dataline.config.persistence.JsonValidationException;
+import io.dataline.config.persistence.PersistenceConfigType;
+import io.dataline.server.helpers.SourceSpecificationHelpers;
+import io.dataline.server.validation.IntegrationSchemaValidation;
+import java.io.File;
+import java.io.IOException;
+import java.util.UUID;
+import java.util.function.Supplier;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class SourceImplementationsHandlerTest {
+  private ConfigPersistence configPersistence;
+  private SourceConnectionSpecification sourceConnectionSpecification;
+  private SourceConnectionImplementation sourceConnectionImplementation;
+  private SourceImplementationsHandler sourceImplementationsHandler;
+  private IntegrationSchemaValidation validator;
+  private Supplier<UUID> uuidGenerator;
+
+  @SuppressWarnings("unchecked")
+  @BeforeEach
+  void setUp() {
+    configPersistence = mock(ConfigPersistence.class);
+    validator = mock(IntegrationSchemaValidation.class);
+    uuidGenerator = mock(Supplier.class);
+
+    sourceConnectionSpecification = SourceSpecificationHelpers.generateSourceSpecification();
+    sourceConnectionImplementation =
+        generateSourceImplementation(sourceConnectionSpecification.getSourceSpecificationId());
+
+    sourceImplementationsHandler =
+        new SourceImplementationsHandler(configPersistence, validator, uuidGenerator);
+  }
+
+  private JsonNode getTestImplementationJson() {
+    final File implementationFile =
+        new File("../dataline-server/src/test/resources/json/TestImplementation.json");
+
+    try {
+      return new ObjectMapper().readTree(implementationFile);
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  private SourceConnectionImplementation generateSourceImplementation(UUID sourceSpecificationId) {
+    final UUID workspaceId = UUID.randomUUID();
+    final UUID sourceImplementationId = UUID.randomUUID();
+
+    JsonNode implementationJson = getTestImplementationJson();
+
+    final SourceConnectionImplementation sourceConnectionImplementation =
+        new SourceConnectionImplementation();
+    sourceConnectionImplementation.setWorkspaceId(workspaceId);
+    sourceConnectionImplementation.setSourceSpecificationId(sourceSpecificationId);
+    sourceConnectionImplementation.setSourceImplementationId(sourceImplementationId);
+    sourceConnectionImplementation.setConfiguration(implementationJson.toString());
+
+    return sourceConnectionImplementation;
+  }
+
+  @Test
+  void testCreateSourceImplementation() throws JsonValidationException, ConfigNotFoundException {
+    when(uuidGenerator.get())
+        .thenReturn(sourceConnectionImplementation.getSourceImplementationId());
+
+    when(configPersistence.getConfig(
+            PersistenceConfigType.SOURCE_CONNECTION_IMPLEMENTATION,
+            sourceConnectionImplementation.getSourceImplementationId().toString(),
+            SourceConnectionImplementation.class))
+        .thenReturn(sourceConnectionImplementation);
+
+    final SourceImplementationCreate sourceImplementationCreate = new SourceImplementationCreate();
+    sourceImplementationCreate.setWorkspaceId(sourceConnectionImplementation.getWorkspaceId());
+    sourceImplementationCreate.setSourceSpecificationId(
+        sourceConnectionSpecification.getSourceSpecificationId());
+    sourceImplementationCreate.setConnectionConfiguration(getTestImplementationJson().toString());
+
+    final SourceImplementationRead actualSourceImplementationRead =
+        sourceImplementationsHandler.createSourceImplementation(sourceImplementationCreate);
+
+    SourceImplementationRead expectedSourceImplementationRead = new SourceImplementationRead();
+    expectedSourceImplementationRead.setSourceSpecificationId(
+        sourceConnectionSpecification.getSourceSpecificationId());
+    expectedSourceImplementationRead.setWorkspaceId(
+        sourceConnectionImplementation.getWorkspaceId());
+    expectedSourceImplementationRead.setSourceImplementationId(
+        sourceConnectionImplementation.getSourceImplementationId());
+    expectedSourceImplementationRead.setConnectionConfiguration(
+        getTestImplementationJson().toString());
+
+    assertEquals(expectedSourceImplementationRead, actualSourceImplementationRead);
+
+    verify(validator)
+        .validateSourceConnectionConfiguration(
+            sourceConnectionSpecification.getSourceSpecificationId(),
+            sourceConnectionImplementation.getConfiguration());
+
+    verify(configPersistence)
+        .writeConfig(
+            PersistenceConfigType.SOURCE_CONNECTION_IMPLEMENTATION,
+            sourceConnectionImplementation.getSourceImplementationId().toString(),
+            sourceConnectionImplementation);
+  }
+
+  @Test
+  void testUpdateSourceImplementation() throws JsonValidationException, ConfigNotFoundException {
+    final Object configuration = sourceConnectionImplementation.getConfiguration();
+    final JsonNode newConfiguration;
+    try {
+      newConfiguration = new ObjectMapper().readTree(configuration.toString());
+    } catch (JsonProcessingException e) {
+      throw new RuntimeException(e);
+    }
+    ((ObjectNode) newConfiguration).put("apiKey", "987-xyz");
+
+    final SourceConnectionImplementation expectedSourceConnectionImplementation =
+        new SourceConnectionImplementation();
+    expectedSourceConnectionImplementation.setWorkspaceId(
+        sourceConnectionImplementation.getWorkspaceId());
+    expectedSourceConnectionImplementation.setSourceSpecificationId(
+        sourceConnectionImplementation.getSourceSpecificationId());
+    expectedSourceConnectionImplementation.setSourceImplementationId(
+        sourceConnectionImplementation.getSourceImplementationId());
+    expectedSourceConnectionImplementation.setConfiguration(newConfiguration.toString());
+
+    when(configPersistence.getConfig(
+            PersistenceConfigType.SOURCE_CONNECTION_IMPLEMENTATION,
+            sourceConnectionImplementation.getSourceImplementationId().toString(),
+            SourceConnectionImplementation.class))
+        .thenReturn(sourceConnectionImplementation)
+        .thenReturn(expectedSourceConnectionImplementation);
+
+    final SourceImplementationUpdate sourceImplementationUpdate = new SourceImplementationUpdate();
+    sourceImplementationUpdate.setSourceImplementationId(
+        sourceConnectionImplementation.getSourceImplementationId());
+    sourceImplementationUpdate.setConnectionConfiguration(newConfiguration.toString());
+    final SourceImplementationRead actualSourceImplementationRead =
+        sourceImplementationsHandler.updateSourceImplementation(sourceImplementationUpdate);
+
+    SourceImplementationRead expectedSourceImplementationRead = new SourceImplementationRead();
+    expectedSourceImplementationRead.setSourceSpecificationId(
+        sourceConnectionSpecification.getSourceSpecificationId());
+    expectedSourceImplementationRead.setWorkspaceId(
+        sourceConnectionImplementation.getWorkspaceId());
+    expectedSourceImplementationRead.setSourceImplementationId(
+        sourceConnectionImplementation.getSourceImplementationId());
+    expectedSourceImplementationRead.setConnectionConfiguration(newConfiguration.toString());
+
+    assertEquals(expectedSourceImplementationRead, actualSourceImplementationRead);
+
+    verify(configPersistence)
+        .writeConfig(
+            PersistenceConfigType.SOURCE_CONNECTION_IMPLEMENTATION,
+            sourceConnectionImplementation.getSourceImplementationId().toString(),
+            expectedSourceConnectionImplementation);
+  }
+
+  @Test
+  void testGetSourceImplementation() throws JsonValidationException, ConfigNotFoundException {
+    when(configPersistence.getConfig(
+            PersistenceConfigType.SOURCE_CONNECTION_IMPLEMENTATION,
+            sourceConnectionImplementation.getSourceImplementationId().toString(),
+            SourceConnectionImplementation.class))
+        .thenReturn(sourceConnectionImplementation);
+
+    SourceImplementationRead expectedSourceImplementationRead = new SourceImplementationRead();
+    expectedSourceImplementationRead.setSourceSpecificationId(
+        sourceConnectionImplementation.getSourceSpecificationId());
+    expectedSourceImplementationRead.setWorkspaceId(
+        sourceConnectionImplementation.getWorkspaceId());
+    expectedSourceImplementationRead.setSourceImplementationId(
+        sourceConnectionImplementation.getSourceImplementationId());
+    expectedSourceImplementationRead.setConnectionConfiguration(
+        sourceConnectionImplementation.getConfiguration());
+
+    final SourceImplementationIdRequestBody sourceImplementationIdRequestBody =
+        new SourceImplementationIdRequestBody();
+    sourceImplementationIdRequestBody.setSourceImplementationId(
+        expectedSourceImplementationRead.getSourceImplementationId());
+
+    final SourceImplementationRead actualSourceImplementationRead =
+        sourceImplementationsHandler.getSourceImplementation(sourceImplementationIdRequestBody);
+
+    assertEquals(expectedSourceImplementationRead, actualSourceImplementationRead);
+  }
+
+  @Test
+  void testListSourceImplementationsForWorkspace() throws JsonValidationException {
+    when(configPersistence.getConfigs(
+            PersistenceConfigType.SOURCE_CONNECTION_IMPLEMENTATION,
+            SourceConnectionImplementation.class))
+        .thenReturn(Sets.newHashSet(sourceConnectionImplementation));
+
+    SourceImplementationRead expectedSourceImplementationRead = new SourceImplementationRead();
+    expectedSourceImplementationRead.setSourceSpecificationId(
+        sourceConnectionImplementation.getSourceSpecificationId());
+    expectedSourceImplementationRead.setWorkspaceId(
+        sourceConnectionImplementation.getWorkspaceId());
+    expectedSourceImplementationRead.setSourceImplementationId(
+        sourceConnectionImplementation.getSourceImplementationId());
+    expectedSourceImplementationRead.setConnectionConfiguration(
+        sourceConnectionImplementation.getConfiguration());
+
+    final WorkspaceIdRequestBody workspaceIdRequestBody = new WorkspaceIdRequestBody();
+    workspaceIdRequestBody.setWorkspaceId(sourceConnectionImplementation.getWorkspaceId());
+
+    final SourceImplementationReadList actualSourceImplementationRead =
+        sourceImplementationsHandler.listSourceImplementationsForWorkspace(workspaceIdRequestBody);
+
+    assertEquals(
+        expectedSourceImplementationRead, actualSourceImplementationRead.getSources().get(0));
+  }
+}

dataline-server/src/test/java/io/dataline/server/handlers/SourceSpecificationsHandlerTest.java

@@ -0,0 +1,52 @@
+package io.dataline.server.handlers;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.Sets;
+import io.dataline.api.model.SourceIdRequestBody;
+import io.dataline.api.model.SourceSpecificationRead;
+import io.dataline.config.SourceConnectionSpecification;
+import io.dataline.config.persistence.ConfigPersistence;
+import io.dataline.config.persistence.JsonValidationException;
+import io.dataline.config.persistence.PersistenceConfigType;
+import io.dataline.server.helpers.SourceSpecificationHelpers;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class SourceSpecificationsHandlerTest {
+  private ConfigPersistence configPersistence;
+  private SourceConnectionSpecification sourceConnectionSpecification;
+  private SourceSpecificationsHandler sourceSpecificationHandler;
+
+  @BeforeEach
+  void setUp() {
+    configPersistence = mock(ConfigPersistence.class);
+    sourceConnectionSpecification = SourceSpecificationHelpers.generateSourceSpecification();
+    sourceSpecificationHandler = new SourceSpecificationsHandler(configPersistence);
+  }
+
+  @Test
+  void testGetSourceSpecification() throws JsonValidationException {
+    when(configPersistence.getConfigs(
+            PersistenceConfigType.SOURCE_CONNECTION_SPECIFICATION,
+            SourceConnectionSpecification.class))
+        .thenReturn(Sets.newHashSet(sourceConnectionSpecification));
+
+    SourceSpecificationRead expectedSourceSpecificationRead = new SourceSpecificationRead();
+    expectedSourceSpecificationRead.setSourceId(sourceConnectionSpecification.getSourceId());
+    expectedSourceSpecificationRead.setSourceSpecificationId(
+        sourceConnectionSpecification.getSourceSpecificationId());
+    expectedSourceSpecificationRead.setConnectionSpecification(
+        sourceConnectionSpecification.getSpecification());
+
+    final SourceIdRequestBody sourceIdRequestBody = new SourceIdRequestBody();
+    sourceIdRequestBody.setSourceId(expectedSourceSpecificationRead.getSourceId());
+
+    final SourceSpecificationRead actualSourceSpecificationRead =
+        sourceSpecificationHandler.getSourceSpecification(sourceIdRequestBody);
+
+    assertEquals(expectedSourceSpecificationRead, actualSourceSpecificationRead);
+  }
+}

dataline-server/src/test/java/io/dataline/server/handlers/SourcesHandlerTest.java

@@ -0,0 +1,97 @@
+package io.dataline.server.handlers;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.Sets;
+import io.dataline.api.model.SourceIdRequestBody;
+import io.dataline.api.model.SourceRead;
+import io.dataline.api.model.SourceReadList;
+import io.dataline.config.StandardSource;
+import io.dataline.config.persistence.ConfigNotFoundException;
+import io.dataline.config.persistence.ConfigPersistence;
+import io.dataline.config.persistence.JsonValidationException;
+import io.dataline.config.persistence.PersistenceConfigType;
+import java.util.Optional;
+import java.util.UUID;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class SourcesHandlerTest {
+  private ConfigPersistence configPersistence;
+  private StandardSource source;
+  private SourcesHandler sourceHandler;
+
+  @BeforeEach
+  void setUp() {
+    configPersistence = mock(ConfigPersistence.class);
+    source = generateSource();
+    sourceHandler = new SourcesHandler(configPersistence);
+  }
+
+  private StandardSource generateSource() {
+    final UUID sourceId = UUID.randomUUID();
+
+    final StandardSource standardSource = new StandardSource();
+    standardSource.setSourceId(sourceId);
+    standardSource.setName("presto");
+
+    return standardSource;
+  }
+
+  @Test
+  void testListSources() throws JsonValidationException {
+    final StandardSource source2 = generateSource();
+    configPersistence.writeConfig(
+        PersistenceConfigType.STANDARD_SOURCE, source2.getSourceId().toString(), source2);
+
+    when(configPersistence.getConfigs(PersistenceConfigType.STANDARD_SOURCE, StandardSource.class))
+        .thenReturn(Sets.newHashSet(source, source2));
+
+    SourceRead expectedSourceRead1 = new SourceRead();
+    expectedSourceRead1.setSourceId(source.getSourceId());
+    expectedSourceRead1.setName(source.getName());
+
+    SourceRead expectedSourceRead2 = new SourceRead();
+    expectedSourceRead2.setSourceId(source2.getSourceId());
+    expectedSourceRead2.setName(source2.getName());
+
+    final SourceReadList actualSourceReadList = sourceHandler.listSources();
+
+    final Optional<SourceRead> actualSourceRead1 =
+        actualSourceReadList.getSources().stream()
+            .filter(sourceRead -> sourceRead.getSourceId().equals(source.getSourceId()))
+            .findFirst();
+    final Optional<SourceRead> actualSourceRead2 =
+        actualSourceReadList.getSources().stream()
+            .filter(sourceRead -> sourceRead.getSourceId().equals(source2.getSourceId()))
+            .findFirst();
+
+    assertTrue(actualSourceRead1.isPresent());
+    assertEquals(expectedSourceRead1, actualSourceRead1.get());
+    assertTrue(actualSourceRead2.isPresent());
+    assertEquals(expectedSourceRead2, actualSourceRead2.get());
+  }
+
+  @Test
+  void testGetSource() throws JsonValidationException, ConfigNotFoundException {
+    when(configPersistence.getConfig(
+            PersistenceConfigType.STANDARD_SOURCE,
+            source.getSourceId().toString(),
+            StandardSource.class))
+        .thenReturn(source);
+
+    SourceRead expectedSourceRead = new SourceRead();
+    expectedSourceRead.setSourceId(source.getSourceId());
+    expectedSourceRead.setName(source.getName());
+
+    final SourceIdRequestBody sourceIdRequestBody = new SourceIdRequestBody();
+    sourceIdRequestBody.setSourceId(source.getSourceId());
+
+    final SourceRead actualSourceRead = sourceHandler.getSource(sourceIdRequestBody);
+
+    assertEquals(expectedSourceRead, actualSourceRead);
+  }
+}

dataline-server/src/test/java/io/dataline/server/handlers/WorkspacesHandlerTest.java

@@ -0,0 +1,126 @@
+package io.dataline.server.handlers;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import io.dataline.api.model.SlugRequestBody;
+import io.dataline.api.model.WorkspaceIdRequestBody;
+import io.dataline.api.model.WorkspaceRead;
+import io.dataline.api.model.WorkspaceUpdate;
+import io.dataline.config.StandardWorkspace;
+import io.dataline.config.persistence.*;
+import java.util.UUID;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class WorkspacesHandlerTest {
+  private ConfigPersistence configPersistence;
+  private StandardWorkspace workspace;
+  private WorkspacesHandler workspacesHandler;
+
+  @BeforeEach
+  void setUp() {
+    configPersistence = mock(ConfigPersistence.class);
+    workspace = generateWorkspace();
+    workspacesHandler = new WorkspacesHandler(configPersistence);
+  }
+
+  private StandardWorkspace generateWorkspace() {
+    final UUID workspaceId = PersistenceConstants.DEFAULT_WORKSPACE_ID;
+
+    final StandardWorkspace standardWorkspace = new StandardWorkspace();
+    standardWorkspace.setWorkspaceId(workspaceId);
+    standardWorkspace.setEmail("test@dataline.io");
+    standardWorkspace.setName("test workspace");
+    standardWorkspace.setSlug("default");
+    standardWorkspace.setInitialSetupComplete(false);
+
+    return standardWorkspace;
+  }
+
+  @Test
+  void testGetWorkspace() throws JsonValidationException, ConfigNotFoundException {
+    when(configPersistence.getConfig(
+            PersistenceConfigType.STANDARD_WORKSPACE,
+            workspace.getWorkspaceId().toString(),
+            StandardWorkspace.class))
+        .thenReturn(workspace);
+
+    final WorkspaceIdRequestBody workspaceIdRequestBody = new WorkspaceIdRequestBody();
+    workspaceIdRequestBody.setWorkspaceId(workspace.getWorkspaceId());
+
+    final WorkspaceRead workspaceRead = new WorkspaceRead();
+    workspaceRead.setWorkspaceId(workspace.getWorkspaceId());
+    workspaceRead.setName("test workspace");
+    workspaceRead.setSlug("default");
+    workspaceRead.setInitialSetupComplete(false);
+
+    assertEquals(workspaceRead, workspacesHandler.getWorkspace(workspaceIdRequestBody));
+  }
+
+  @Test
+  void testGetWorkspaceBySlug() throws JsonValidationException, ConfigNotFoundException {
+    when(configPersistence.getConfig(
+            PersistenceConfigType.STANDARD_WORKSPACE,
+            workspace.getWorkspaceId().toString(),
+            StandardWorkspace.class))
+        .thenReturn(workspace);
+
+    final SlugRequestBody slugRequestBody = new SlugRequestBody();
+    slugRequestBody.setSlug("default");
+
+    final WorkspaceRead workspaceRead = new WorkspaceRead();
+    workspaceRead.setWorkspaceId(workspace.getWorkspaceId());
+    workspaceRead.setName("test workspace");
+    workspaceRead.setSlug("default");
+    workspaceRead.setInitialSetupComplete(false);
+
+    assertEquals(workspaceRead, workspacesHandler.getWorkspaceBySlug(slugRequestBody));
+  }
+
+  @Test
+  void testUpdateWorkspace() throws JsonValidationException, ConfigNotFoundException {
+
+    final WorkspaceUpdate workspaceUpdate = new WorkspaceUpdate();
+    workspaceUpdate.setWorkspaceId(workspace.getWorkspaceId());
+    workspaceUpdate.setAnonymousDataCollection(true);
+    workspaceUpdate.setSecurityUpdates(false);
+    workspaceUpdate.setNews(false);
+    workspaceUpdate.setInitialSetupComplete(true);
+
+    final StandardWorkspace expectedWorkspace = new StandardWorkspace();
+    expectedWorkspace.setWorkspaceId(workspace.getWorkspaceId());
+    expectedWorkspace.setEmail("test@dataline.io");
+    expectedWorkspace.setName("test workspace");
+    expectedWorkspace.setSlug("default");
+    expectedWorkspace.setAnonymousDataCollection(true);
+    expectedWorkspace.setSecurityUpdates(false);
+    expectedWorkspace.setNews(false);
+    expectedWorkspace.setInitialSetupComplete(true);
+
+    when(configPersistence.getConfig(
+            PersistenceConfigType.STANDARD_WORKSPACE,
+            workspace.getWorkspaceId().toString(),
+            StandardWorkspace.class))
+        .thenReturn(workspace)
+        .thenReturn(expectedWorkspace);
+
+    final WorkspaceRead actualWorkspaceRead = workspacesHandler.updateWorkspace(workspaceUpdate);
+
+    final WorkspaceRead expectedWorkspaceRead = new WorkspaceRead();
+    expectedWorkspaceRead.setWorkspaceId(workspace.getWorkspaceId());
+    expectedWorkspaceRead.setName("test workspace");
+    expectedWorkspaceRead.setSlug("default");
+    expectedWorkspaceRead.setInitialSetupComplete(true);
+
+    verify(configPersistence)
+        .writeConfig(
+            PersistenceConfigType.STANDARD_WORKSPACE,
+            expectedWorkspace.getWorkspaceId().toString(),
+            expectedWorkspace);
+
+    assertEquals(expectedWorkspaceRead, actualWorkspaceRead);
+  }
+}

dataline-server/src/test/java/io/dataline/server/helpers/SourceSpecificationHelpers.java

@@ -0,0 +1,33 @@
+package io.dataline.server.helpers;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.dataline.config.SourceConnectionSpecification;
+import java.io.File;
+import java.io.IOException;
+import java.util.UUID;
+
+public class SourceSpecificationHelpers {
+  public static SourceConnectionSpecification generateSourceSpecification() {
+    final UUID sourceId = UUID.randomUUID();
+    final UUID sourceSpecificationId = UUID.randomUUID();
+
+    final File specificationFile =
+        new File("../dataline-server/src/test/resources/json/TestSpecification.json");
+
+    JsonNode specificationJson;
+    try {
+      specificationJson = new ObjectMapper().readTree(specificationFile);
+    } catch (IOException e) {
+      throw new RuntimeException(e);
+    }
+
+    final SourceConnectionSpecification sourceConnectionSpecification =
+        new SourceConnectionSpecification();
+    sourceConnectionSpecification.setSourceId(sourceId);
+    sourceConnectionSpecification.setSourceSpecificationId(sourceSpecificationId);
+    sourceConnectionSpecification.setSpecification(specificationJson.toString());
+
+    return sourceConnectionSpecification;
+  }
+}

dataline-server/src/test/resources/json/TestImplementation.json

@@ -0,0 +1,4 @@
+{
+  "apiKey": "123-abc",
+  "hostname": "dataline.io"
+}

dataline-server/src/test/resources/json/TestSpecification.json

@@ -0,0 +1,22 @@
+{
+  "specification": {
+    "$schema": "http://json-schema.org/draft-07/schema#",
+    "$id": "https://github.com/datalineio/dataline/blob/master/dataline-server/src/test/resources/json/TestSpecification.json",
+    "title": "TestSpecification",
+    "description": "information output by the connection.",
+    "type": "object",
+    "required": [
+      "apiKey",
+      "hostname"
+    ],
+    "additionalProperties": false,
+    "properties": {
+      "apiKey": {
+        "type": "string"
+      },
+      "hostname": {
+        "type": "string"
+      }
+    }
+  }
+}