Add Snapchat Marketing oAuth backend

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

@@ -14,6 +14,7 @@
 import io.airbyte.oauth.flows.IntercomOAuthFlow;
 import io.airbyte.oauth.flows.SalesforceOAuthFlow;
 import io.airbyte.oauth.flows.SlackOAuthFlow;
+import io.airbyte.oauth.flows.SnapchatMarketingOAuthFlow;
 import io.airbyte.oauth.flows.SurveymonkeyOAuthFlow;
 import io.airbyte.oauth.flows.TrelloOAuthFlow;
 import io.airbyte.oauth.flows.facebook.FacebookMarketingOAuthFlow;
@@ -46,6 +47,7 @@ public OAuthImplementationFactory(final ConfigRepository configRepository, final
         .put("airbyte/source-instagram", new InstagramOAuthFlow(configRepository, httpClient))
         .put("airbyte/source-salesforce", new SalesforceOAuthFlow(configRepository, httpClient))
         .put("airbyte/source-slack", new SlackOAuthFlow(configRepository, httpClient))
+        .put("airbyte/source-snapchat-marketing", new SnapchatMarketingOAuthFlow(configRepository, httpClient))
         .put("airbyte/source-surveymonkey", new SurveymonkeyOAuthFlow(configRepository, httpClient))
         .put("airbyte/source-trello", new TrelloOAuthFlow(configRepository, httpClient))
         .build();

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

@@ -0,0 +1,76 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.oauth.flows;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.config.persistence.ConfigRepository;
+import io.airbyte.oauth.BaseOAuth2Flow;
+import java.io.IOException;
+import java.net.URISyntaxException;
+import java.net.http.HttpClient;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.function.Supplier;
+import org.apache.http.client.utils.URIBuilder;
+
+/**
+ * Following docs from https://marketingapi.snapchat.com/docs/#authentication
+ */
+public class SnapchatMarketingOAuthFlow extends BaseOAuth2Flow {
+  // Clickable link for IDE
+  // https://help.salesforce.com/s/articleView?language=en_US&id=sf.remoteaccess_oauth_web_server_flow.htm
+
+  private static final String AUTHORIZE_URL = "https://accounts.snapchat.com/login/oauth2/authorize";
+  private static final String ACCESS_TOKEN_URL = "https://accounts.snapchat.com/login/oauth2/access_token";
+  private static final String SCOPES = "snapchat-marketing-api";
+
+  public SnapchatMarketingOAuthFlow(final ConfigRepository configRepository, final HttpClient httpClient) {
+    super(configRepository, httpClient);
+  }
+
+  @VisibleForTesting
+  SnapchatMarketingOAuthFlow(final ConfigRepository configRepository, final HttpClient httpClient, final Supplier<String> stateSupplier) {
+    super(configRepository, httpClient, stateSupplier);
+  }
+
+  @Override
+  protected String formatConsentUrl(final UUID definitionId, final String clientId, final String redirectUrl) throws IOException {
+    try {
+      return new URIBuilder(AUTHORIZE_URL)
+          .addParameter("client_id", clientId)
+          .addParameter("redirect_uri", redirectUrl)
+          .addParameter("response_type", "code")
+          .addParameter("scope", SCOPES)
+          .addParameter("state", getState())
+          .build().toString();
+    } catch (final URISyntaxException e) {
+      throw new IOException("Failed to format Consent URL for OAuth flow", e);
+    }
+  }
+
+  @Override
+  protected String getAccessTokenUrl() {
+    return ACCESS_TOKEN_URL;
+  }
+
+  @Override
+  protected Map<String, String> getAccessTokenQueryParameters(final String clientId,
+                                                              final String clientSecret,
+                                                              final String authCode,
+                                                              final String redirectUrl) {
+    return ImmutableMap.<String, String>builder()
+        .putAll(super.getAccessTokenQueryParameters(clientId, clientSecret, authCode, redirectUrl))
+        .put("grant_type", "authorization_code")
+        .build();
+  }
+
+  @Override
+  protected List<String> getDefaultOAuthOutputPath() {
+    return List.of();
+  }
+
+}

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

@@ -0,0 +1,73 @@
+/*
+ * 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.net.http.HttpClient;
+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 SnapchatMarketingOAuthFlowIntegrationTest extends OAuthFlowIntegrationTest {
+
+  @Override
+  protected Path getCredentialsPath() {
+    return Path.of("secrets/snapchat.json");
+  }
+
+  @Override
+  protected String getRedirectUrl() {
+    return "https://f215-195-114-147-152.ngrok.io/auth_flow";
+  }
+
+  protected int getServerListeningPort() {
+    return 3000;
+  }
+
+  @Override
+  protected OAuthFlowImplementation getFlowImplementation(ConfigRepository configRepository, HttpClient httpClient) {
+    return new SnapchatMarketingOAuthFlow(configRepository, httpClient);
+  }
+
+  @Test
+  public void testFullSnapchatMarketingOAuthFlow() throws InterruptedException, ConfigNotFoundException, IOException, JsonValidationException {
+    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("client_id").asText())
+            .put("client_secret", credentialsJson.get("client_secret").asText())
+            .build()))));
+    final String url = getFlowImplementation(configRepository, httpClient).getSourceConsentUrl(workspaceId, definitionId, getRedirectUrl());
+    LOGGER.info("Waiting for user consent at: {}", url);
+    waitForResponse(20);
+    assertTrue(serverHandler.isSucceeded(), "Failed to get User consent on time");
+    final Map<String, Object> params = flow.completeSourceOAuth(workspaceId, definitionId,
+        Map.of("code", serverHandler.getParamValue()), getRedirectUrl());
+    LOGGER.info("Response from completing OAuth Flow is: {}", params.toString());
+    assertTrue(params.containsKey("refresh_token"));
+    assertTrue(params.get("refresh_token").toString().length() > 0);
+  }
+
+}

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

@@ -0,0 +1,80 @@
+/*
+ * 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 SnapchatMarketingOAuthFlowTest {
+
+  private UUID workspaceId;
+  private UUID definitionId;
+  private ConfigRepository configRepository;
+  private SnapchatMarketingOAuthFlow snapchatOAuthFlow;
+  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()))));
+    snapchatOAuthFlow = new SnapchatMarketingOAuthFlow(configRepository, httpClient, SnapchatMarketingOAuthFlowTest::getConstantState);
+
+  }
+
+  @Test
+  public void testGetSourceConsentUrl() throws IOException, InterruptedException, ConfigNotFoundException {
+    final String consentUrl = snapchatOAuthFlow.getSourceConsentUrl(workspaceId, definitionId, REDIRECT_URL);
+    assertEquals(
+        "https://accounts.snapchat.com/login/oauth2/authorize?client_id=test_client_id&redirect_uri=https%3A%2F%2Fairbyte.io&response_type=code&scope=snapchat-marketing-api&state=state",
+        consentUrl);
+  }
+
+  @Test
+  public void testCompleteSourceOAuth() throws IOException, JsonValidationException, InterruptedException, ConfigNotFoundException {
+
+    final Map<String, String> returnedCredentials = Map.of("refresh_token", "refresh_token_response");
+    final HttpResponse response = mock(HttpResponse.class);
+    when(response.body()).thenReturn(Jsons.serialize(returnedCredentials));
+    when(httpClient.send(any(), any())).thenReturn(response);
+    final Map<String, Object> queryParams = Map.of("code", "test_code");
+    final Map<String, Object> actualQueryParams =
+        snapchatOAuthFlow.completeSourceOAuth(workspaceId, definitionId, queryParams, REDIRECT_URL);
+    assertEquals(Jsons.serialize(returnedCredentials), Jsons.serialize(actualQueryParams));
+  }
+
+}