Asana source: Add oauth init flow parameters support (#6832)

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/d0243522-dccf-4978-8ba0-37ed47a0bdbf.json

@@ -2,6 +2,6 @@
   "sourceDefinitionId": "d0243522-dccf-4978-8ba0-37ed47a0bdbf",
   "name": "Asana",
   "dockerRepository": "airbyte/source-asana",
-  "dockerImageTag": "0.1.2",
+  "dockerImageTag": "0.1.3",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/asana"
 }

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -7,7 +7,7 @@
 - sourceDefinitionId: d0243522-dccf-4978-8ba0-37ed47a0bdbf
   name: Asana
   dockerRepository: airbyte/source-asana
-  dockerImageTag: 0.1.2
+  dockerImageTag: 0.1.3
   documentationUrl: https://docs.airbyte.io/integrations/sources/asana
   sourceType: api
 - sourceDefinitionId: 686473f1-76d9-4994-9cc7-9b13da46147c

airbyte-integrations/bases/source-acceptance-test/CHANGELOG.md

@@ -1,5 +1,8 @@
 # Changelog
 
+## 0.1.23
+Fix incorrect auth init flow check defect.
+
 ## 0.1.22
 Fix checking schemas with root $ref keyword
 

airbyte-integrations/bases/source-acceptance-test/Dockerfile

@@ -9,7 +9,7 @@ COPY setup.py ./
 COPY pytest.ini ./
 RUN pip install .
 
-LABEL io.airbyte.version=0.1.22
+LABEL io.airbyte.version=0.1.23
 LABEL io.airbyte.name=airbyte/source-acceptance-test
 
 ENTRYPOINT ["python", "-m", "pytest", "-p", "source_acceptance_test.plugin"]

airbyte-integrations/bases/source-acceptance-test/source_acceptance_test/utils/json_schema_helper.py

@@ -187,7 +187,7 @@ def _scan_schema(subschema, path=""):
         if "oneOf" in subschema or "anyOf" in subschema:
             if annotate_one_of:
                 return [
-                    _scan_schema({"type": "object", **s}, path + "(0)")
+                    _scan_schema({"type": "object", **s}, path + f"({num})")
                     for num, s in enumerate(subschema.get("oneOf") or subschema.get("anyOf"))
                 ]
             return [_scan_schema({"type": "object", **s}, path) for s in subschema.get("oneOf") or subschema.get("anyOf")]

airbyte-integrations/bases/source-acceptance-test/unit_tests/test_core.py

@@ -254,6 +254,43 @@ def test_read(schema, record, should_fail):
             ),
             "Specified oauth fields are missed from spec schema:",
         ),
+        # SUCCESS: root object index equal to 1
+        (
+            ConnectorSpecification(
+                connectionSpecification={
+                    "type": "object",
+                    "properties": {
+                        "credentials": {
+                            "type": "object",
+                            "oneOf": [
+                                {
+                                    "properties": {
+                                        "api_key": {"type": "string"},
+                                    }
+                                },
+                                {
+                                    "properties": {
+                                        "client_id": {"type": "string"},
+                                        "client_secret": {"type": "string"},
+                                        "access_token": {"type": "string"},
+                                        "refresh_token": {"type": "string"},
+                                    }
+                                },
+                            ],
+                        }
+                    },
+                },
+                authSpecification={
+                    "auth_type": "oauth2.0",
+                    "oauth2Specification": {
+                        "rootObject": ["credentials", 1],
+                        "oauthFlowInitParameters": [["client_id"], ["client_secret"]],
+                        "oauthFlowOutputParameters": [["access_token"], ["refresh_token"]],
+                    },
+                },
+            ),
+            "",
+        ),
     ],
 )
 def test_validate_oauth_flow(connector_spec, expected_error):

airbyte-integrations/connectors/source-asana/Dockerfile

@@ -12,5 +12,5 @@ RUN pip install .
 ENV AIRBYTE_ENTRYPOINT "python /airbyte/integration_code/main.py"
 ENTRYPOINT ["python", "/airbyte/integration_code/main.py"]
 
-LABEL io.airbyte.version=0.1.2
+LABEL io.airbyte.version=0.1.3
 LABEL io.airbyte.name=airbyte/source-asana

airbyte-integrations/connectors/source-asana/integration_tests/invalid_config.json

@@ -1,3 +1,3 @@
 {
-  "access_token": "<wrong_access_token>"
+  "credentials": { "personal_access_token": "<wrong_access_token>" }
 }

airbyte-integrations/connectors/source-asana/setup.py

@@ -9,10 +9,7 @@
     "airbyte-cdk~=0.1",
 ]
 
-TEST_REQUIREMENTS = [
-    "pytest~=6.1",
-    "source-acceptance-test",
-]
+TEST_REQUIREMENTS = ["pytest~=6.1", "requests-mock~=1.9.3"]
 
 setup(
     name="source_asana",

airbyte-integrations/connectors/source-asana/source_asana/oauth.py

@@ -0,0 +1,36 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+from typing import Tuple
+
+import requests
+from airbyte_cdk.sources.streams.http.auth import Oauth2Authenticator
+
+
+class AsanaOauth2Authenticator(Oauth2Authenticator):
+    """
+    Unlike most Oauth services that accept oauth parameters in form of json
+    encoded body, Asana's oauth token endpoint expects oauth parameters to be
+    in form-encoded post body.
+    https://developers.asana.com/docs/oauth
+    """
+
+    def refresh_access_token(self) -> Tuple[str, int]:
+        """
+        Override base refresh_access_token method to send form-encoded oauth
+        parameters over POST request body.
+        Returns:
+            Tuple of access token and expiration time in seconds
+        """
+        data = {
+            "client_id": (None, self.client_id),
+            "client_secret": (None, self.client_secret),
+            "grant_type": (None, "refresh_token"),
+            "refresh_token": (None, self.refresh_token),
+        }
+
+        response = requests.post(self.token_refresh_endpoint, files=data)
+        response.raise_for_status()
+        response_body = response.json()
+        return response_body["access_token"], response_body["expires_in"]

airbyte-integrations/connectors/source-asana/source_asana/source.py

@@ -3,29 +3,46 @@
 #
 
 
-from typing import Any, List, Mapping, Tuple
+from typing import Any, List, Mapping, Tuple, Union
 
 from airbyte_cdk import AirbyteLogger
 from airbyte_cdk.models import SyncMode
 from airbyte_cdk.sources import AbstractSource
 from airbyte_cdk.sources.streams import Stream
 from airbyte_cdk.sources.streams.http.auth import TokenAuthenticator
+from source_asana.oauth import AsanaOauth2Authenticator
 
 from .streams import CustomFields, Projects, Sections, Stories, Tags, Tasks, TeamMemberships, Teams, Users, Workspaces
 
 
 class SourceAsana(AbstractSource):
     def check_connection(self, logger: AirbyteLogger, config: Mapping[str, Any]) -> Tuple[bool, Any]:
         try:
-            workspaces_stream = Workspaces(authenticator=TokenAuthenticator(token=config["access_token"]))
+            workspaces_stream = Workspaces(authenticator=self._get_authenticator(config))
             next(workspaces_stream.read_records(sync_mode=SyncMode.full_refresh))
             return True, None
         except Exception as e:
             return False, e
 
+    @staticmethod
+    def _get_authenticator(config: dict) -> Union[TokenAuthenticator, AsanaOauth2Authenticator]:
+        if "access_token" in config:
+            # Before Oauth we had Person Access Token stored under "access_token"
+            # config field, this code here is for backward compatibility
+            return TokenAuthenticator(token=config["access_token"])
+        creds = config.get("credentials")
+        if "personal_access_token" in creds:
+            return TokenAuthenticator(token=creds["personal_access_token"])
+        else:
+            return AsanaOauth2Authenticator(
+                token_refresh_endpoint="https://app.asana.com/-/oauth_token",
+                client_secret=creds["client_secret"],
+                client_id=creds["client_id"],
+                refresh_token=creds["refresh_token"],
+            )
+
     def streams(self, config: Mapping[str, Any]) -> List[Stream]:
-        authenticator = TokenAuthenticator(token=config["access_token"])
-        args = {"authenticator": authenticator}
+        args = {"authenticator": self._get_authenticator(config)}
         return [
             CustomFields(**args),
             Projects(**args),

airbyte-integrations/connectors/source-asana/source_asana/spec.json

@@ -4,15 +4,73 @@
     "$schema": "http://json-schema.org/draft-07/schema#",
     "title": "Asana Spec",
     "type": "object",
-    "required": ["access_token"],
-    "additionalProperties": false,
+    "additionalProperties": true,
     "properties": {
-      "access_token": {
-        "type": "string",
-        "title": "Personal Access Token",
-        "description": "Asana Personal Access Token (generate yours <a href=\"https://app.asana.com/0/developer-console\">here</a>).",
-        "airbyte_secret": true
+      "credentials": {
+        "title": "Authentication mechanism",
+        "description": "Choose how to authenticate to Github",
+        "type": "object",
+        "oneOf": [
+          {
+            "type": "object",
+            "title": "Authenticate with Personal Access Token",
+            "required": ["personal_access_token"],
+            "properties": {
+              "option_title": {
+                "type": "string",
+                "title": "Credentials title",
+                "description": "PAT Credentials",
+                "const": "PAT Credentials"
+              },
+              "personal_access_token": {
+                "type": "string",
+                "title": "Personal Access Token",
+                "description": "Asana Personal Access Token (generate yours <a href=\"https://app.asana.com/0/developer-console\">here</a>).",
+                "airbyte_secret": true
+              }
+            }
+          },
+          {
+            "type": "object",
+            "title": "Authenticate via Asana (Oauth)",
+            "required": ["client_id", "client_secret", "refresh_token"],
+            "properties": {
+              "option_title": {
+                "type": "string",
+                "title": "Credentials title",
+                "description": "OAuth Credentials",
+                "const": "OAuth Credentials"
+              },
+              "client_id": {
+                "type": "string",
+                "title": "",
+                "description": "",
+                "airbyte_secret": true
+              },
+              "client_secret": {
+                "type": "string",
+                "title": "",
+                "description": "",
+                "airbyte_secret": true
+              },
+              "refresh_token": {
+                "type": "string",
+                "title": "",
+                "description": "",
+                "airbyte_secret": true
+              }
+            }
+          }
+        ]
       }
     }
+  },
+  "authSpecification": {
+    "auth_type": "oauth2.0",
+    "oauth2Specification": {
+      "rootObject": ["credentials", 1],
+      "oauthFlowInitParameters": [["client_id"], ["client_secret"]],
+      "oauthFlowOutputParameters": [["refresh_token"]]
+    }
   }
 }

airbyte-integrations/connectors/source-asana/unit_tests/test_oauth.py

@@ -0,0 +1,31 @@
+#
+# Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+#
+
+import pytest
+import requests_mock
+from source_asana.oauth import AsanaOauth2Authenticator
+
+
+@pytest.fixture
+def req_mock():
+    with requests_mock.Mocker() as mock:
+        yield mock
+
+
+def test_oauth(req_mock):
+    URL = "https://example.com"
+    TOKEN = "test_token"
+    req_mock.post(URL, json={"access_token": TOKEN, "expires_in": 3600})
+    a = AsanaOauth2Authenticator(
+        token_refresh_endpoint=URL,
+        client_secret="client_secret",
+        client_id="client_id",
+        refresh_token="refresh_token",
+    )
+    token = a.get_access_token()
+    assert token == TOKEN
+    assert "multipart/form-data;" in req_mock.last_request.headers["Content-Type"]
+    assert "client_secret" in req_mock.last_request.body.decode()
+    assert "client_id" in req_mock.last_request.body.decode()
+    assert "refresh_token" in req_mock.last_request.body.decode()

docs/integrations/sources/asana.md

@@ -61,6 +61,7 @@ to obtain Personal Access Token for your account.
 
 | Version | Date       | Pull Request | Subject |
 | :------ | :--------  | :-----       | :------ |
+| 0.1.3   | 2021-10-06 | [](https://github.com/airbytehq/airbyte/pull/) | Add oauth init flow parameters support |
 | 0.1.2   | 2021-09-24 | [6402](https://github.com/airbytehq/airbyte/pull/6402) | Fix SAT tests: update schemas and invalid_config.json file |
 | 0.1.1   | 2021-06-09 | [3973](https://github.com/airbytehq/airbyte/pull/3973) | Add entrypoint and bump version for connector |
 | 0.1.0   | 2021-05-25 | [3510](https://github.com/airbytehq/airbyte/pull/3510) | New Source: Asana |

tools/bin/ci_integration_test.sh

@@ -49,7 +49,7 @@ run_status=${PIPESTATUS[0]}
 
 test $run_status == "0" || {
    # Build failed
-   link=$(cat build.out | grep -A1 "Publishing build scan..." | tail -n1 | tr -d "\n")
+   link=$(cat build.out | grep -a -A1 "Publishing build scan..." | tail -n1 | tr -d "\n")
    # Save gradle scan link to github GRADLE_SCAN_LINK variable for next job.
    # https://docs.github.com/en/actions/reference/workflow-commands-for-github-actions#setting-an-environment-variable
    echo "GRADLE_SCAN_LINK=$link" >> $GITHUB_ENV