Parse list of dicts in json_schema_helper.find_nodes()

airbyte-integrations/bases/source-acceptance-test/setup.py

@@ -13,7 +13,7 @@
     "inflection~=0.5",
     "pdbpp~=0.10",
     "pydantic~=1.6",
-    "pytest~=6.1",
+    "pytest~=6.2,>=6.2.4",
     "pytest-sugar~=0.9",
     "pytest-timeout~=1.4",
     "pprintpp~=0.4",

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

@@ -4,7 +4,7 @@
 
 
 from functools import reduce
-from typing import Any, List, Mapping, Optional, Set
+from typing import Any, Dict, List, Mapping, Optional, Set, Text, Union
 
 import pendulum
 from jsonref import JsonRef
@@ -122,21 +122,23 @@ def get_node(self, path: List[str]) -> Any:
         return node
 
     def find_nodes(self, keys: List[str]) -> List[List[str]]:
-        """Get all nodes of schema that has specifies properties
+        """Find all paths that lead to nodes with the specified keys.
 
-        :param keys:
+        :param keys: list of keys
         :return: list of json object paths
         """
         variant_paths = []
 
-        def traverse_schema(_schema, path=None):
+        def traverse_schema(_schema: Union[Dict[Text, Any], List], path=None):
             path = path or []
             if path and path[-1] in keys:
                 variant_paths.append(path)
-            for item in _schema:
-                next_obj = _schema[item] if isinstance(_schema, dict) else item
-                if isinstance(next_obj, (list, dict)):
-                    traverse_schema(next_obj, [*path, item])
+            if isinstance(_schema, dict):
+                for item in _schema:
+                    traverse_schema(_schema[item], [*path, item])
+            elif isinstance(_schema, list):
+                for i, item in enumerate(_schema):
+                    traverse_schema(_schema[i], [*path, i])
 
         traverse_schema(self._schema)
         return variant_paths

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

@@ -3,7 +3,7 @@
 #
 
 from enum import Enum
-from typing import Union
+from typing import Any, List, Text, Union
 
 import pendulum
 import pytest
@@ -208,3 +208,52 @@ def test_get_object_strucutre(object, pathes):
 )
 def test_get_expected_schema_structure(schema, pathes):
     assert get_expected_schema_structure(schema) == pathes
+
+
+@pytest.mark.parametrize(
+    "keys, num_paths, last_value",
+    [
+        (["description"], 1, "Tests that keys can be found inside lists of dicts"),
+        (["option1"], 2, {"a_key": "a_value"}),
+        (["option2"], 1, ["value1", "value2"]),
+        (["nonexistent_key"], 0, None),
+        (["option1", "option2"], 3, ["value1", "value2"])
+    ],
+)
+def test_find_and_get_nodes(keys: List[Text], num_paths: int, last_value: Any):
+    schema = {
+        "title": "Key_inside_oneOf",
+        "description": "Tests that keys can be found inside lists of dicts",
+        "type": "object",
+        "properties": {
+            "credentials": {
+                "type": "object",
+                "oneOf": [
+                    {
+                        "type": "object",
+                        "properties": {
+                            "common": {"type": "string", "const": "option1", "default": "option1"},
+                            "option1": {"type": "string"},
+                        },
+                    },
+                    {
+                        "type": "object",
+                        "properties": {
+                            "common": {"type": "string", "const": "option2", "default": "option2"},
+                            "option1": {"a_key": "a_value"},
+                            "option2": ["value1", "value2"],
+                        },
+                    },
+                ],
+            }
+        },
+    }
+    schema_helper = JsonSchemaHelper(schema)
+    variant_paths = schema_helper.find_nodes(keys=keys)
+    assert len(variant_paths) == num_paths
+
+    if variant_paths:
+        values_at_nodes = []
+        for path in variant_paths:
+            values_at_nodes.append(schema_helper.get_node(path))
+        assert last_value in values_at_nodes