Eliziario/hubspot oauth

airbyte-oauth/src/main/java/io/airbyte/oauth/BaseOAuthFlow.java

@@ -14,7 +14,6 @@
 import java.io.IOException;
 import java.lang.reflect.Type;
 import java.net.URI;
-import java.net.URISyntaxException;
 import java.net.URLEncoder;
 import java.net.http.HttpClient;
 import java.net.http.HttpClient.Version;
@@ -100,31 +99,6 @@ public String getDestinationConsentUrl(final UUID workspaceId, final UUID destin
     return formatConsentUrl(destinationDefinitionId, getClientIdUnsafe(oAuthParamConfig), redirectUrl);
   }
 
-  protected String formatConsentUrl(String clientId,
-                                    String redirectUrl,
-                                    String host,
-                                    String path,
-                                    String scope,
-                                    String responseType)
-      throws IOException {
-    final URIBuilder builder = new URIBuilder()
-        .setScheme("https")
-        .setHost(host)
-        .setPath(path)
-        // required
-        .addParameter("client_id", clientId)
-        .addParameter("redirect_uri", redirectUrl)
-        .addParameter("state", getState())
-        // optional
-        .addParameter("response_type", responseType)
-        .addParameter("scope", scope);
-    try {
-      return builder.build().toString();
-    } catch (URISyntaxException e) {
-      throw new IOException("Failed to format Consent URL for OAuth flow", e);
-    }
-  }
-
   /**
    * Depending on the OAuth flow implementation, the URL to grant user's consent may differ,
    * especially in the query parameters to be provided. This function should generate such consent URL
@@ -235,6 +209,7 @@ protected Map<String, Object> extractRefreshToken(final JsonNode data, String ac
     } else {
       LOGGER.info("Oauth flow failed. Data received from server: {}", data);
       throw new IOException(String.format("Missing 'refresh_token' in query params from %s. Response: %s", accessTokenUrl));
+
     }
     return Map.of("credentials", result);
 

airbyte-oauth/src/main/java/io/airbyte/oauth/OAuthImplementationFactory.java

@@ -6,11 +6,7 @@
 
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.config.persistence.ConfigRepository;
-import io.airbyte.oauth.flows.AsanaOAuthFlow;
-import io.airbyte.oauth.flows.GithubOAuthFlow;
-import io.airbyte.oauth.flows.SalesforceOAuthFlow;
-import io.airbyte.oauth.flows.SurveymonkeyOAuthFlow;
-import io.airbyte.oauth.flows.TrelloOAuthFlow;
+import io.airbyte.oauth.flows.*;
 import io.airbyte.oauth.flows.facebook.FacebookMarketingOAuthFlow;
 import io.airbyte.oauth.flows.facebook.FacebookPagesOAuthFlow;
 import io.airbyte.oauth.flows.facebook.InstagramOAuthFlow;
@@ -39,6 +35,7 @@ public OAuthImplementationFactory(final ConfigRepository configRepository) {
         .put("airbyte/source-salesforce", new SalesforceOAuthFlow(configRepository))
         .put("airbyte/source-surveymonkey", new SurveymonkeyOAuthFlow(configRepository))
         .put("airbyte/source-trello", new TrelloOAuthFlow(configRepository))
+        .put("airbyte/source-hubspot", new HubspotOAuthFlow(configRepository))
         .build();
   }
 

airbyte-oauth/src/main/java/io/airbyte/oauth/flows/HubspotOAuthFlow.java

@@ -0,0 +1,93 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.oauth.flows;
+
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.config.persistence.ConfigRepository;
+import io.airbyte.oauth.BaseOAuthFlow;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.net.http.HttpClient;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Supplier;
+import org.apache.http.client.utils.URIBuilder;
+
+public class HubspotOAuthFlow extends BaseOAuthFlow {
+
+  private final String AUTHORIZE_URL = "https://app.hubspot.com/oauth/authorize";
+
+  public HubspotOAuthFlow(ConfigRepository configRepository) {
+    super(configRepository);
+  }
+
+  public HubspotOAuthFlow(ConfigRepository configRepository, HttpClient httpClient, Supplier<String> stateSupplier) {
+    super(configRepository, httpClient, stateSupplier, TOKEN_REQUEST_CONTENT_TYPE.JSON);
+  }
+
+  /**
+   * Depending on the OAuth flow implementation, the URL to grant user's consent may differ,
+   * especially in the query parameters to be provided. This function should generate such consent URL
+   * accordingly.
+   *
+   * @param definitionId The configured definition ID of this client
+   * @param clientId The configured client ID
+   * @param redirectUrl the redirect URL
+   */
+  @Override
+  protected String formatConsentUrl(UUID definitionId, String clientId, String redirectUrl) throws IOException {
+    try {
+      return new URIBuilder(AUTHORIZE_URL)
+          .addParameter("client_id", clientId)
+          .addParameter("redirect_uri", redirectUrl)
+          .addParameter("state", getState())
+          .addParameter("scopes", getScopes())
+          .build().toString();
+    } catch (URISyntaxException e) {
+      throw new IOException("Failed to format Consent URL for OAuth flow", e);
+    }
+  }
+
+  @Override
+  protected Map<String, String> getAccessTokenQueryParameters(String clientId, String clientSecret, String authCode, String redirectUrl) {
+    return ImmutableMap.<String, String>builder()
+        // required
+        .put("client_id", clientId)
+        .put("redirect_uri", redirectUrl)
+        .put("client_secret", clientSecret)
+        .put("code", authCode)
+        .put("grant_type", "authorization_code")
+        .build();
+  }
+
+  private String getScopes() {
+    return String.join(" ", "content",
+        "crm.schemas.deals.read",
+        "crm.objects.owners.read",
+        "forms",
+        "tickets",
+        "e-commerce",
+        "crm.objects.companies.read",
+        "crm.lists.read",
+        "crm.objects.deals.read",
+        "crm.schemas.contacts.read",
+        "crm.objects.contacts.read",
+        "crm.schemas.companies.read",
+        "files",
+        "forms-uploaded-files",
+        "files.ui_hidden.read");
+  }
+
+  /**
+   * Returns the URL where to retrieve the access token from.
+   *
+   * @param oAuthParamConfig the configuration map
+   */
+  @Override
+  protected String getAccessTokenUrl() {
+    return "https://api.hubapi.com/oauth/v1/token";
+  }
+
+}

airbyte-oauth/src/test-integration/java/io.airbyte.oauth.flows/FacebookOAuthFlowIntegrationTest.java

@@ -31,7 +31,7 @@ public class FacebookOAuthFlowIntegrationTest extends OAuthFlowIntegrationTest {
   protected static final String REDIRECT_URL = "http://localhost:9000/auth_flow";
 
   @Override
-  protected Path get_credentials_path() {
+  protected Path getCredentialsPath() {
     return CREDENTIALS_PATH;
   }
 

airbyte-oauth/src/test-integration/java/io.airbyte.oauth.flows/GithubOAuthFlowIntegrationTest.java

@@ -31,7 +31,7 @@ public class GithubOAuthFlowIntegrationTest extends OAuthFlowIntegrationTest {
   protected static final int SERVER_LISTENING_PORT = 8000;
 
   @Override
-  protected Path get_credentials_path() {
+  protected Path getCredentialsPath() {
     return CREDENTIALS_PATH;
   }
 

airbyte-oauth/src/test-integration/java/io.airbyte.oauth.flows/SurveymonkeyOAuthFlowIntegrationTest.java

@@ -30,7 +30,7 @@ public class SurveymonkeyOAuthFlowIntegrationTest extends OAuthFlowIntegrationTe
   protected static final String REDIRECT_URL = "http://localhost:3000/auth_flow";
 
   @Override
-  protected Path get_credentials_path() {
+  protected Path getCredentialsPath() {
     return CREDENTIALS_PATH;
   }
 

airbyte-oauth/src/test-integration/java/io/airbyte/oauth/flows/HubspotOAuthFlowIntegrationTest.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.oauth.flows;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.when;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.config.SourceOAuthParameter;
+import io.airbyte.config.persistence.ConfigNotFoundException;
+import io.airbyte.config.persistence.ConfigRepository;
+import io.airbyte.oauth.OAuthFlowImplementation;
+import io.airbyte.validation.json.JsonValidationException;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import org.junit.jupiter.api.Test;
+
+public class HubspotOAuthFlowIntegrationTest extends OAuthFlowIntegrationTest {
+
+  @Override
+  protected Path getCredentialsPath() {
+    return Path.of("secrets/hubspot.json");
+  }
+
+  @Override
+  protected OAuthFlowImplementation getFlowObject(ConfigRepository configRepository) {
+    return new HubspotOAuthFlow(configRepository);
+  }
+
+  @Test
+  public void testFullOAuthFlow() throws InterruptedException, ConfigNotFoundException, IOException, JsonValidationException {
+    int limit = 100;
+    final UUID workspaceId = UUID.randomUUID();
+    final UUID definitionId = UUID.randomUUID();
+    final String fullConfigAsString = new String(Files.readAllBytes(getCredentialsPath()));
+    final JsonNode credentialsJson = Jsons.deserialize(fullConfigAsString);
+    when(configRepository.listSourceOAuthParam()).thenReturn(List.of(new SourceOAuthParameter()
+        .withOauthParameterId(UUID.randomUUID())
+        .withSourceDefinitionId(definitionId)
+        .withWorkspaceId(workspaceId)
+        .withConfiguration(Jsons.jsonNode(ImmutableMap.builder()
+            .put("client_id", credentialsJson.get("credentials").get("client_id").asText())
+            .put("client_secret", credentialsJson.get("credentials").get("client_secret").asText())
+            .build()))));
+    var flowObject = getFlowObject(configRepository);
+    final String url = flowObject.getSourceConsentUrl(workspaceId, definitionId, REDIRECT_URL);
+    LOGGER.info("Waiting for user consent at: {}", url);
+    // TODO: To automate, start a selenium job to navigate to the Consent URL and click on allowing
+    // access...
+    while (!serverHandler.isSucceeded() && limit > 0) {
+      Thread.sleep(1000);
+      limit -= 1;
+    }
+    assertTrue(serverHandler.isSucceeded(), "Failed to get User consent on time");
+    final Map<String, Object> params = flowObject.completeSourceOAuth(workspaceId, definitionId,
+        Map.of("code", serverHandler.getParamValue()), REDIRECT_URL);
+    LOGGER.info("Response from completing OAuth Flow is: {}", params.toString());
+    assertTrue(params.containsKey("credentials"));
+    final Map<String, Object> credentials = (Map<String, Object>) params.get("credentials");
+    assertTrue(credentials.containsKey("refresh_token"));
+    assertTrue(credentials.get("refresh_token").toString().length() > 0);
+    assertTrue(credentials.containsKey("access_token"));
+    assertTrue(credentials.get("access_token").toString().length() > 0);
+  }
+
+}

airbyte-oauth/src/test-integration/java/io/airbyte/oauth/flows/OAuthFlowIntegrationTest.java

@@ -38,13 +38,15 @@ public abstract class OAuthFlowIntegrationTest {
   protected HttpServer server;
   protected ServerHandler serverHandler;
 
-  protected abstract Path get_credentials_path();
+  protected Path getCredentialsPath() {
+    return Path.of("secrets/config.json");
+  };
 
   protected abstract OAuthFlowImplementation getFlowObject(ConfigRepository configRepository);
 
   @BeforeEach
   public void setup() throws IOException {
-    if (!Files.exists(get_credentials_path())) {
+    if (!Files.exists(getCredentialsPath())) {
       throw new IllegalStateException(
           "Must provide path to a oauth credentials file.");
     }

airbyte-oauth/src/test/java/io/airbyte/oauth/flows/HubspotOAuthFlowTest.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.oauth.flows;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.commons.json.Jsons;
+import io.airbyte.config.SourceOAuthParameter;
+import io.airbyte.config.persistence.ConfigNotFoundException;
+import io.airbyte.config.persistence.ConfigRepository;
+import io.airbyte.validation.json.JsonValidationException;
+import java.io.IOException;
+import java.net.http.HttpClient;
+import java.net.http.HttpResponse;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+public class HubspotOAuthFlowTest {
+
+  private UUID workspaceId;
+  private UUID definitionId;
+  private ConfigRepository configRepository;
+  private HubspotOAuthFlow flow;
+  private HttpClient httpClient;
+
+  private static final String REDIRECT_URL = "https://airbyte.io";
+
+  private static String getConstantState() {
+    return "state";
+  }
+
+  @BeforeEach
+  public void setup() throws IOException, JsonValidationException {
+    workspaceId = UUID.randomUUID();
+    definitionId = UUID.randomUUID();
+    configRepository = mock(ConfigRepository.class);
+    httpClient = mock(HttpClient.class);
+    when(configRepository.listSourceOAuthParam()).thenReturn(List.of(new SourceOAuthParameter()
+        .withOauthParameterId(UUID.randomUUID())
+        .withSourceDefinitionId(definitionId)
+        .withWorkspaceId(workspaceId)
+        .withConfiguration(Jsons.jsonNode(ImmutableMap.builder()
+            .put("client_id", "test_client_id")
+            .put("client_secret", "test_client_secret")
+            .build()))));
+    flow = new HubspotOAuthFlow(configRepository, httpClient, HubspotOAuthFlowTest::getConstantState);
+
+  }
+
+  @Test
+  public void testGetSourceConcentUrl() throws IOException, ConfigNotFoundException {
+    final String concentUrl =
+        flow.getSourceConsentUrl(workspaceId, definitionId, REDIRECT_URL);
+    assertEquals(concentUrl,
+        "https://app.hubspot.com/oauth/authorize?client_id=test_client_id&redirect_uri=https%3A%2F%2Fairbyte.io&state=state&scopes=content+crm.schemas.deals.read+crm.objects.owners.read+forms+tickets+e-commerce+crm.objects.companies.read+crm.lists.read+crm.objects.deals.read+crm.schemas.contacts.read+crm.objects.contacts.read+crm.schemas.companies.read+files+forms-uploaded-files+files.ui_hidden.read");
+  }
+
+  @Test
+  public void testCompleteSourceOAuth() throws IOException, InterruptedException, ConfigNotFoundException {
+    final var response = mock(HttpResponse.class);
+    var returnedCredentials = "{\"refresh_token\":\"refresh_token_response\"}";
+    when(response.body()).thenReturn(returnedCredentials);
+    when(httpClient.send(any(), any())).thenReturn(response);
+    final Map<String, Object> queryParams = Map.of("code", "test_code");
+    final Map<String, Object> actualQueryParams =
+        flow.completeSourceOAuth(workspaceId, definitionId, queryParams, REDIRECT_URL);
+    assertEquals(Jsons.serialize(Map.of("credentials", Jsons.deserialize(returnedCredentials))), Jsons.serialize(actualQueryParams));
+  }
+
+}