🐛 SAT: Fix test_oneof_usage fail #9860

.pre-commit-config.yaml

@@ -11,11 +11,17 @@ repos:
     rev: 21.11b1
     hooks:
       - id: black
+        args: ["--line-length=140"]
   - repo: https://github.com/timothycrosley/isort
     rev: 5.10.1
     hooks:
       - id: isort
-        args: ["--dont-follow-links", "--jobs=-1"]
+        args:
+          [
+            "--settings-path=tools/python/.isort.cfg",
+            "--dont-follow-links",
+            "--jobs=-1",
+          ]
         additional_dependencies: ["colorama"]
   - repo: https://github.com/pre-commit/mirrors-prettier
     rev: v2.5.0
@@ -34,12 +40,14 @@ repos:
     rev: v0.0.1a2.post1
     hooks:
       - id: pyproject-flake8
+        args: ["--config=tools/python/.flake8"]
         additional_dependencies: ["mccabe"]
         alias: flake8
   - repo: https://github.com/pre-commit/mirrors-mypy
     rev: v0.910-1
     hooks:
       - id: mypy
+        args: ["--config-file=tools/python/.mypy.ini"]
         exclude: |
           (?x)^.*(
               octavia-cli/unit_tests/|

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

@@ -1,5 +1,8 @@
 # Changelog
 
+## 0.1.44
+Fix test_oneof_usage test: [#9861](https://github.com/airbytehq/airbyte/pull/9861)
+
 ## 0.1.44
 Fix incorrect name of primary_keys attribute: [#9768](https://github.com/airbytehq/airbyte/pull/9768)
 

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

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

airbyte-integrations/bases/source-acceptance-test/source_acceptance_test/tests/test_core.py

@@ -76,18 +76,15 @@ def test_oneof_usage(self, actual_connector_spec: ConnectorSpecification):
         docs_msg = f"See specification reference at {docs_url}."
 
         schema_helper = JsonSchemaHelper(actual_connector_spec.connectionSpecification)
-        variant_paths = schema_helper.find_variant_paths()
+        variant_paths = schema_helper.find_nodes(keys=["oneOf", "anyOf"])
 
         for variant_path in variant_paths:
-            top_level_obj = dpath.util.get(self._schema, "/".join(variant_path[:-1]))
-            if "$ref" in top_level_obj:
-                obj_def = top_level_obj["$ref"].split("/")[-1]
-                top_level_obj = self._schema["definitions"][obj_def]
+            top_level_obj = schema_helper.get_node(variant_path[:-1])
             assert (
                 top_level_obj.get("type") == "object"
             ), f"The top-level definition in a `oneOf` block should have type: object. misconfigured object: {top_level_obj}. {docs_msg}"
 
-            variants = dpath.util.get(self._schema, "/".join(variant_path))
+            variants = schema_helper.get_node(variant_path)
             for variant in variants:
                 assert "properties" in variant, f"Each item in the oneOf array should be a property with type object. {docs_msg}"
 

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

@@ -11,7 +11,9 @@
 
 
 class CatalogField:
-    """Field class to represent cursor/pk fields"""
+    """Field class to represent cursor/pk fields.
+    It eases the read of values from records according to schema definition.
+    """
 
     def __init__(self, schema: Mapping[str, Any], path: List[str]):
         self.schema = schema
@@ -48,16 +50,46 @@ def parse(self, record: Mapping[str, Any], path: Optional[List[str]] = None) ->
 
 
 class JsonSchemaHelper:
+    """Helper class to simplify schema validation and read of records according to their schema."""
+
     def __init__(self, schema):
         self._schema = schema
 
-    def get_ref(self, path: str):
+    def get_ref(self, path: str) -> Any:
+        """Resolve reference
+
+        :param path: reference (#/definitions/SomeClass, etc)
+        :return: part of schema that is definition of the reference
+        :raises KeyError: in case path can't be followed
+        """
         node = self._schema
         for segment in path.split("/")[1:]:
             node = node[segment]
         return node
 
     def get_property(self, path: List[str]) -> Mapping[str, Any]:
+        """Get any part of schema according to provided path, resolves $refs if necessary
+
+        schema = {
+                "properties": {
+                    "field1": {
+                        "properties": {
+                            "nested_field": {
+                                <inner_object>
+                            }
+                        }
+                    },
+                    "field2": ...
+                }
+            }
+
+        helper = JsonSchemaHelper(schema)
+        helper.get_property(["field1", "nested_field"]) == <inner_object>
+
+        :param path: list of fields in the order of navigation
+        :return: discovered part of schema
+        :raises KeyError: in case path can't be followed
+        """
         node = self._schema
         for segment in path:
             if "$ref" in node:
@@ -66,17 +98,40 @@ def get_property(self, path: List[str]) -> Mapping[str, Any]:
         return node
 
     def field(self, path: List[str]) -> CatalogField:
+        """Get schema property and wrap it into CatalogField.
+
+        CatalogField is a helper to ease the read of values from records according to schema definition.
+
+        :param path: list of fields in the order of navigation
+        :return: discovered part of schema wrapped in CatalogField
+        :raises KeyError: in case path can't be followed
+        """
         return CatalogField(schema=self.get_property(path), path=path)
 
-    def find_variant_paths(self) -> List[List[str]]:
+    def get_node(self, path: List[str]) -> Any:
+        """Return part of schema by specified path
+
+        :param path: list of fields in the order of navigation
         """
-        return list of json object paths for oneOf or anyOf attributes
+
+        node = self._schema
+        for segment in path:
+            if "$ref" in node:
+                node = self.get_ref(node["$ref"])
+            node = node[segment]
+        return node
+
+    def find_nodes(self, keys: List[str]) -> List[List[str]]:
+        """Get all nodes of schema that has specifies properties
+
+        :param keys:
+        :return: list of json object paths
         """
         variant_paths = []
 
         def traverse_schema(_schema, path=None):
             path = path or []
-            if path and path[-1] in ["oneOf", "anyOf"]:
+            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