🏗️ SAT: add oauthFlowInitParameters verification for spec (airbytehq#…

…6534)

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

@@ -1,5 +1,8 @@
 # Changelog
 
+## 0.1.20
+Add oauth init flow parameter verification for spec.
+
 ## 0.1.19
 Assert a non-empty overlap between the fields present in the record and the declared json schema.
 

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.19
+LABEL io.airbyte.version=0.1.20
 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/tests/test_core.py

@@ -21,23 +21,33 @@
 
 @pytest.mark.default_timeout(10)
 class TestSpec(BaseTest):
-    def test_match_expected(self, connector_spec: ConnectorSpecification, connector_config: SecretDict, docker_runner: ConnectorRunner):
-        output = docker_runner.call_spec()
-        spec_messages = filter_output(output, Type.SPEC)
 
-        assert len(spec_messages) == 1, "Spec message should be emitted exactly once"
-        if connector_spec:
-            assert spec_messages[0].spec == connector_spec, "Spec should be equal to the one in spec.json file"
-
-        assert docker_runner.env_variables.get("AIRBYTE_ENTRYPOINT"), "AIRBYTE_ENTRYPOINT must be set in dockerfile"
-        assert docker_runner.env_variables.get("AIRBYTE_ENTRYPOINT") == " ".join(
-            docker_runner.entry_point
-        ), "env should be equal to space-joined entrypoint"
+    spec_cache: ConnectorSpecification = None
+
+    @pytest.fixture(name="actual_connector_spec")
+    def actual_connector_spec_fixture(request: BaseTest, docker_runner):
+        if not request.spec_cache:
+            output = docker_runner.call_spec()
+            spec_messages = filter_output(output, Type.SPEC)
+            assert len(spec_messages) == 1, "Spec message should be emitted exactly once"
+            assert docker_runner.env_variables.get("AIRBYTE_ENTRYPOINT"), "AIRBYTE_ENTRYPOINT must be set in dockerfile"
+            assert docker_runner.env_variables.get("AIRBYTE_ENTRYPOINT") == " ".join(
+                docker_runner.entry_point
+            ), "env should be equal to space-joined entrypoint"
+            spec = spec_messages[0].spec
+            request.spec_cache = spec
+        return request.spec_cache
+
+    def test_match_expected(
+        self, connector_spec: ConnectorSpecification, actual_connector_spec: ConnectorSpecification, connector_config: SecretDict
+    ):
 
+        if connector_spec:
+            assert actual_connector_spec == connector_spec, "Spec should be equal to the one in spec.json file"
         # Getting rid of technical variables that start with an underscore
         config = {key: value for key, value in connector_config.data.items() if not key.startswith("_")}
 
-        spec_message_schema = spec_messages[0].spec.connectionSpecification
+        spec_message_schema = actual_connector_spec.connectionSpecification
         validate(instance=config, schema=spec_message_schema)
 
         js_helper = JsonSchemaHelper(spec_message_schema)
@@ -56,6 +66,31 @@ def test_has_secret(self):
     def test_secret_never_in_the_output(self):
         """This test should be injected into any docker command it needs to know current config and spec"""
 
+    def test_oauth_flow_parameters(self, actual_connector_spec: ConnectorSpecification):
+        """
+        Check if connector has correct oauth flow parameters according to https://docs.airbyte.io/connector-development/connector-specification-reference
+        """
+        self._validate_authflow_parameters(actual_connector_spec)
+
+    @staticmethod
+    def _validate_authflow_parameters(connector_spec: ConnectorSpecification):
+        if not connector_spec.authSpecification:
+            return
+        spec_schema = connector_spec.connectionSpecification
+        oauth_spec = connector_spec.authSpecification.oauth2Specification
+        parameters: List[List[str]] = oauth_spec.oauthFlowInitParameters + oauth_spec.oauthFlowOutputParameters
+        root_object = oauth_spec.rootObject
+        assert len(root_object) == 2
+        assert root_object[0] in spec_schema.get("properties", {}), f"oauth root object {root_object[0]} does not exists"
+        if "oneOf" in spec_schema.get("properties")[root_object[0]]:
+            params = {"/" + "/".join([f"{root_object[0]}({root_object[1]})", *p]) for p in parameters}
+            schema_path = set(get_expected_schema_structure(spec_schema, annotate_one_of=True))
+        else:
+            params = {"/" + "/".join([root_object[0], *p]) for p in parameters}
+            schema_path = set(get_expected_schema_structure(spec_schema))
+        diff = params - schema_path
+        assert diff == set(), f"Specified ouath fields are missed from spec schema: {diff}"
+
 
 @pytest.mark.default_timeout(30)
 class TestConnection(BaseTest):

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

@@ -156,9 +156,10 @@ def _traverse_obj_and_get_path(obj, path=""):
     return paths
 
 
-def get_expected_schema_structure(schema: dict) -> List[str]:
+def get_expected_schema_structure(schema: dict, annotate_one_of: bool = False) -> List[str]:
     """
     Travers through json schema and compose list of property keys that object expected to have.
+    :param annotate_one_of: Generate one_of index in path
     :param schema: jsonschema to get expected paths
     :returns list of object property keys paths
     """
@@ -168,6 +169,11 @@ def get_expected_schema_structure(schema: dict) -> List[str]:
 
     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)")
+                    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")]
         schema_type = subschema.get("type", ["null"])
         if not isinstance(schema_type, list):

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

@@ -5,10 +5,19 @@
 from unittest.mock import MagicMock
 
 import pytest
-from airbyte_cdk.models import AirbyteMessage, AirbyteRecordMessage, AirbyteStream, ConfiguredAirbyteCatalog, ConfiguredAirbyteStream, Type
+from airbyte_cdk.models import (
+    AirbyteMessage,
+    AirbyteRecordMessage,
+    AirbyteStream,
+    ConfiguredAirbyteCatalog,
+    ConfiguredAirbyteStream,
+    ConnectorSpecification,
+    Type,
+)
 from source_acceptance_test.config import BasicReadTestConfig
 from source_acceptance_test.tests.test_core import TestBasicRead as _TestBasicRead
 from source_acceptance_test.tests.test_core import TestDiscovery as _TestDiscovery
+from source_acceptance_test.tests.test_core import TestSpec as _TestSpec
 
 
 @pytest.mark.parametrize(
@@ -71,3 +80,163 @@ def test_read(schema, record, should_fail):
             t.test_read(None, catalog, input_config, [], docker_runner_mock, MagicMock())
     else:
         t.test_read(None, catalog, input_config, [], docker_runner_mock, MagicMock())
+
+
+@pytest.mark.parametrize(
+    "connector_spec, expected_error",
+    [
+        # SUCCESS: no authSpecification specified
+        (ConnectorSpecification(connectionSpecification={}), ""),
+        # FAIL: Field specified in root object does not exist
+        (
+            ConnectorSpecification(
+                connectionSpecification={"type": "object"},
+                authSpecification={
+                    "auth_type": "oauth2.0",
+                    "oauth2Specification": {
+                        "rootObject": ["credentials", 0],
+                        "oauthFlowInitParameters": [["client_id"], ["client_secret"]],
+                        "oauthFlowOutputParameters": [["access_token"], ["refresh_token"]],
+                    },
+                },
+            ),
+            "oauth root object credentials does not exists",
+        ),
+        # FAIL: Some oauth fields missed
+        (
+            ConnectorSpecification(
+                connectionSpecification={
+                    "type": "object",
+                    "properties": {
+                        "credentials": {
+                            "type": "object",
+                            "properties": {
+                                "client_id": {"type": "string"},
+                                "client_secret": {"type": "string"},
+                                "access_token": {"type": "string"},
+                            },
+                        }
+                    },
+                },
+                authSpecification={
+                    "auth_type": "oauth2.0",
+                    "oauth2Specification": {
+                        "rootObject": ["credentials", 0],
+                        "oauthFlowInitParameters": [["client_id"], ["client_secret"]],
+                        "oauthFlowOutputParameters": [["access_token"], ["refresh_token"]],
+                    },
+                },
+            ),
+            "Specified ouath fields are missed from spec schema: {'/credentials/refresh_token'}",
+        ),
+        # SUCCESS: case w/o oneOf property
+        (
+            ConnectorSpecification(
+                connectionSpecification={
+                    "type": "object",
+                    "properties": {
+                        "credentials": {
+                            "type": "object",
+                            "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", 0],
+                        "oauthFlowInitParameters": [["client_id"], ["client_secret"]],
+                        "oauthFlowOutputParameters": [["access_token"], ["refresh_token"]],
+                    },
+                },
+            ),
+            "",
+        ),
+        # SUCCESS: case w/ oneOf property
+        (
+            ConnectorSpecification(
+                connectionSpecification={
+                    "type": "object",
+                    "properties": {
+                        "credentials": {
+                            "type": "object",
+                            "oneOf": [
+                                {
+                                    "properties": {
+                                        "client_id": {"type": "string"},
+                                        "client_secret": {"type": "string"},
+                                        "access_token": {"type": "string"},
+                                        "refresh_token": {"type": "string"},
+                                    }
+                                },
+                                {
+                                    "properties": {
+                                        "api_key": {"type": "string"},
+                                    }
+                                },
+                            ],
+                        }
+                    },
+                },
+                authSpecification={
+                    "auth_type": "oauth2.0",
+                    "oauth2Specification": {
+                        "rootObject": ["credentials", 0],
+                        "oauthFlowInitParameters": [["client_id"], ["client_secret"]],
+                        "oauthFlowOutputParameters": [["access_token"], ["refresh_token"]],
+                    },
+                },
+            ),
+            "",
+        ),
+        # FAIL: Wrong root object index
+        (
+            ConnectorSpecification(
+                connectionSpecification={
+                    "type": "object",
+                    "properties": {
+                        "credentials": {
+                            "type": "object",
+                            "oneOf": [
+                                {
+                                    "properties": {
+                                        "client_id": {"type": "string"},
+                                        "client_secret": {"type": "string"},
+                                        "access_token": {"type": "string"},
+                                        "refresh_token": {"type": "string"},
+                                    }
+                                },
+                                {
+                                    "properties": {
+                                        "api_key": {"type": "string"},
+                                    }
+                                },
+                            ],
+                        }
+                    },
+                },
+                authSpecification={
+                    "auth_type": "oauth2.0",
+                    "oauth2Specification": {
+                        "rootObject": ["credentials", 1],
+                        "oauthFlowInitParameters": [["client_id"], ["client_secret"]],
+                        "oauthFlowOutputParameters": [["access_token"], ["refresh_token"]],
+                    },
+                },
+            ),
+            "Specified ouath fields are missed from spec schema:",
+        ),
+    ],
+)
+def test_validate_oauth_flow(connector_spec, expected_error):
+    t = _TestSpec()
+    if expected_error:
+        with pytest.raises(AssertionError, match=expected_error):
+            t.test_oauth_flow_parameters(connector_spec)
+    else:
+        t.test_oauth_flow_parameters(connector_spec)