CAT: validate stream schemas to be valid json schemas (airbytehq#32787)

Co-authored-by: Roman Yermilov [GL] <86300758+roman-yermilov-gl@users.noreply.github.com>
Co-authored-by: Christo Grabowski <108154848+ChristoGrab@users.noreply.github.com>
Co-authored-by: Anatolii Yatsuk <35109939+tolik0@users.noreply.github.com>
Co-authored-by: Edward Gao <edward.gao@airbyte.io>
Co-authored-by: davydov-d <davydov-d@users.noreply.github.com>
Co-authored-by: Serhii Lazebnyi <serhii.lazebnyi@globallogic.com>

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

@@ -1,6 +1,10 @@
 # Changelog
 
 
+## 2.1.4
+
+Add check to ensure stream schemas are valid json schemas
+
 ## 2.1.3
 
 Remove dockerfile and migrations tools. We are now running CAT with `airbyte-ci` and new migration operation will be done by the `airbyte-ci` tool.

airbyte-integrations/bases/connector-acceptance-test/connector_acceptance_test/tests/test_core.py

@@ -663,6 +663,11 @@ def duplicated_stream_names(self, streams) -> List[str]:
             name_counts[stream.name] = count + 1
         return [k for k, v in name_counts.items() if v > 1]
 
+    def test_streams_have_valid_json_schemas(self, discovered_catalog: Mapping[str, Any]):
+        """Check if all stream schemas are valid json schemas."""
+        for stream_name, stream in discovered_catalog.items():
+            jsonschema.Draft7Validator.check_schema(stream.json_schema)
+
     def test_defined_cursors_exist_in_schema(self, discovered_catalog: Mapping[str, Any]):
         """Check if all of the source defined cursor fields are exists on stream's json schema."""
         for stream_name, stream in discovered_catalog.items():

airbyte-integrations/bases/connector-acceptance-test/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
 name = "connector-acceptance-test"
-version = "2.1.3"
+version = "2.1.4"
 description = "Contains acceptance tests for connectors."
 authors = ["Airbyte <contact@airbyte.io>"]
 license = "MIT"

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

@@ -23,6 +23,7 @@
 )
 from connector_acceptance_test.config import BasicReadTestConfig, Config, ExpectedRecordsConfig, IgnoredFieldsConfiguration
 from connector_acceptance_test.tests import test_core
+from jsonschema.exceptions import SchemaError
 
 from .conftest import does_not_raise
 
@@ -75,6 +76,83 @@ def test_discovery_uniquely_named_streams():
     assert len(t.duplicated_stream_names(streams)) == 0
 
 
+@pytest.mark.parametrize(
+    "schema, should_fail",
+    [
+        (
+            {
+                "$schema": "https://json-schema.org/draft-07/schema#",
+                "type": ["null", "object"],
+                "properties": {
+                    "amount": {
+                        "type": ["null", "integer"]
+                    },
+                    "amount_details": {
+                        "type": ["null", "object"],
+                        "properties": {
+                            "atm_fee": ["null", "integer"]
+                        }
+                    }
+                }
+            },
+            True
+        ),
+        (
+            {
+                "$schema": "https://json-schema.org/draft-07/schema#",
+                "type": ["null", "object"],
+                "properties": {
+                    "amount": "integer",
+                    "amount_details": {
+                        "type": ["null", "object"],
+                        "properties": {
+                            "atm_fee": {
+                                "type": ["null", "integer"]
+                            }
+                        }
+                    }
+                }
+            },
+            True
+        ),
+        (
+            {
+                "$schema": "https://json-schema.org/draft-07/schema#",
+                "type": ["null", "object"],
+                "properties": {
+                    "amount": {
+                        "type": ["null", "integer"]
+                    },
+                    "amount_details": {
+                        "type": ["null", "object"],
+                        "properties": {
+                            "atm_fee": {
+                                "type": ["null", "integer"]
+                            }
+                        }
+                    }
+                }
+            },
+            False
+        )
+    ],
+)
+def test_streams_have_valid_json_schemas(schema, should_fail):
+    t = test_core.TestDiscovery()
+    discovered_catalog = {
+        "test_stream": AirbyteStream.parse_obj(
+            {
+                "name": "test_stream",
+                "json_schema": schema,
+                "supported_sync_modes": ["full_refresh", "incremental"],
+            }
+        )
+    }
+    expectation = pytest.raises(SchemaError) if should_fail else does_not_raise()
+    with expectation:
+        t.test_streams_have_valid_json_schemas(discovered_catalog)
+
+
 @pytest.mark.parametrize(
     "schema, should_fail",
     [