diff --git a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py index 5a20bdf1f3..b4a92e0c7f 100644 --- a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py +++ b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py @@ -12,8 +12,8 @@ import orjson import yaml -from jsonschema.exceptions import ValidationError -from jsonschema.validators import validate +from jsonschema.exceptions import ValidationError, best_match +from jsonschema.validators import validator_for from packaging.version import InvalidVersion, Version from airbyte_cdk.config_observation import create_connector_config_control_message @@ -447,12 +447,23 @@ def _validate_source(self) -> None: Validates the connector manifest against the declarative component schema """ + # The component schema ships with the CDK and is checked against its metaschema in unit + # tests, so a valid manifest skips `check_schema`. On failure the schema is checked first, + # so a broken schema is reported as `SchemaError` rather than blamed on the manifest. + # `best_match` keeps the error `jsonschema.validate` would raise. + validator_class = validator_for(self._declarative_component_schema) try: - validate(self._source_config, self._declarative_component_schema) - except ValidationError as e: + error = best_match( + validator_class(self._declarative_component_schema).iter_errors(self._source_config) + ) + except Exception: + validator_class.check_schema(self._declarative_component_schema) + raise + if error is not None: + validator_class.check_schema(self._declarative_component_schema) raise ValidationError( "Validation against json schema defined in declarative_component_schema.yaml schema failed" - ) from e + ) from error cdk_version_str = metadata.version("airbyte_cdk") cdk_version = self._parse_version(cdk_version_str, "airbyte-cdk") diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index 055576f45f..5db724100e 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -22,8 +22,8 @@ import orjson import yaml from airbyte_protocol_dataclasses.models import AirbyteStreamStatus, Level, StreamDescriptor -from jsonschema.exceptions import ValidationError -from jsonschema.validators import validate +from jsonschema.exceptions import ValidationError, best_match +from jsonschema.validators import validator_for from airbyte_cdk.config_observation import create_connector_config_control_message from airbyte_cdk.connector_builder.models import ( @@ -345,12 +345,23 @@ def _validate_source(self) -> None: Validates the connector manifest against the declarative component schema """ + # The component schema ships with the CDK and is checked against its metaschema in unit + # tests, so a valid manifest skips `check_schema`. On failure the schema is checked first, + # so a broken schema is reported as `SchemaError` rather than blamed on the manifest. + # `best_match` keeps the error `jsonschema.validate` would raise. + validator_class = validator_for(self._declarative_component_schema) try: - validate(self._source_config, self._declarative_component_schema) - except ValidationError as e: + error = best_match( + validator_class(self._declarative_component_schema).iter_errors(self._source_config) + ) + except Exception: + validator_class.check_schema(self._declarative_component_schema) + raise + if error is not None: + validator_class.check_schema(self._declarative_component_schema) raise ValidationError( "Validation against json schema defined in declarative_component_schema.yaml schema failed" - ) from e + ) from error def _migrate_and_transform_config( self, diff --git a/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py b/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py index b565874323..58dce25e5b 100644 --- a/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py +++ b/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py @@ -14,7 +14,8 @@ import pytest import requests import yaml -from jsonschema.exceptions import ValidationError +from jsonschema.exceptions import SchemaError, UnknownType, ValidationError +from jsonschema.validators import Draft7Validator, validate import unit_tests.sources.declarative.external_component # Needed for dynamic imports to work from airbyte_cdk.legacy.sources.declarative.manifest_declarative_source import ( @@ -33,6 +34,7 @@ ) from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( ConcurrentDeclarativeSource, + _get_declarative_component_schema, ) from airbyte_cdk.sources.declarative.parsers.model_to_component_factory import ( ModelToComponentFactory, @@ -2540,3 +2542,105 @@ def capturing_resolve(resolver_self, *args, **kwargs): "HttpComponentsResolver's requester should have api_budget set during dynamic stream " "discovery, but it was None. This means discovery HTTP requests are not rate-limited." ) + + +def _two_stream_schema_validation_manifest() -> dict[str, Any]: + def stream(name: str) -> dict[str, Any]: + return { + "type": "DeclarativeStream", + "name": name, + "primary_key": [], + "schema_loader": { + "type": "InlineSchemaLoader", + "schema": {"type": "object", "properties": {}}, + }, + "retriever": { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url_base": "https://api.test.com", + "path": f"/{name}", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, + }, + } + + return { + "version": "6.0.0", + "type": "DeclarativeSource", + "check": {"type": "CheckStream", "stream_names": ["a"]}, + "streams": [stream("a"), stream("b")], + } + + +def test_invalid_manifest_raises_the_best_match_error_not_the_first(): + manifest = _two_stream_schema_validation_manifest() + manifest["streams"][0]["primary_key"] = 5 + manifest["streams"][1]["retriever"]["requester"]["http_method"] = "FETCH" + schema = _get_declarative_component_schema() + + assert next(Draft7Validator(schema).iter_errors(manifest)).json_path == "$.streams[0]" + + with pytest.raises(ValidationError) as expected_error: + validate(manifest, schema) + expected = expected_error.value + + with pytest.raises(ValidationError) as exc_info: + ManifestDeclarativeSource(source_config=manifest) + + assert ( + exc_info.value.message + == "Validation against json schema defined in declarative_component_schema.yaml schema failed" + ) + cause = exc_info.value.__cause__ + assert cause is not None + assert cause.message == expected.message + assert cause.json_path == expected.json_path + assert list(cause.schema_path) == list(expected.schema_path) + assert cause.validator == expected.validator + assert cause.validator_value == expected.validator_value + assert cause.json_path == "$.streams[1].retriever.requester.http_method" + + +def test_building_a_source_does_not_check_the_component_schema(): + with patch.object( + Draft7Validator, + "check_schema", + side_effect=AssertionError("check_schema must not run at source build"), + ): + ManifestDeclarativeSource(source_config=_two_stream_schema_validation_manifest()) + + +def test_a_broken_component_schema_that_fails_the_manifest_raises_schema_error(): + manifest = _two_stream_schema_validation_manifest() + broken = deepcopy(_get_declarative_component_schema()) + broken["definitions"]["HttpRequester"]["properties"]["http_method"]["enum"] = "GET" + + assert list(Draft7Validator(broken).iter_errors(manifest)) + + with patch( + "airbyte_cdk.legacy.sources.declarative.manifest_declarative_source._get_declarative_component_schema", + return_value=broken, + ): + with pytest.raises(SchemaError): + ManifestDeclarativeSource(source_config=manifest) + + +def test_a_broken_component_schema_that_breaks_validation_raises_schema_error(): + manifest = _two_stream_schema_validation_manifest() + broken = deepcopy(_get_declarative_component_schema()) + broken["definitions"]["HttpRequester"]["properties"]["path"]["type"] = "strng" + + with pytest.raises(UnknownType): + list(Draft7Validator(broken).iter_errors(manifest)) + + with patch( + "airbyte_cdk.legacy.sources.declarative.manifest_declarative_source._get_declarative_component_schema", + return_value=broken, + ): + with pytest.raises(SchemaError): + ManifestDeclarativeSource(source_config=manifest) diff --git a/unit_tests/sources/declarative/test_concurrent_declarative_source.py b/unit_tests/sources/declarative/test_concurrent_declarative_source.py index 686bd54f0b..5718d072d7 100644 --- a/unit_tests/sources/declarative/test_concurrent_declarative_source.py +++ b/unit_tests/sources/declarative/test_concurrent_declarative_source.py @@ -25,7 +25,8 @@ AirbyteTraceMessage, TraceType, ) -from jsonschema.exceptions import ValidationError +from jsonschema.exceptions import SchemaError, UnknownType, ValidationError +from jsonschema.validators import Draft7Validator, validate from typing_extensions import deprecated import unit_tests.sources.declarative.external_component # Needed for dynamic imports to work @@ -52,6 +53,7 @@ from airbyte_cdk.sources.declarative.async_job.job_tracker import ConcurrentJobLimitReached from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( ConcurrentDeclarativeSource, + _get_declarative_component_schema, ) from airbyte_cdk.sources.declarative.extractors.record_filter import ( ClientSideIncrementalRecordFilterDecorator, @@ -6964,3 +6966,116 @@ def test_combined_extractor_skipping_empty_records_reads_every_page( 6, 7, ] + + +def _two_stream_schema_validation_manifest() -> Dict[str, Any]: + def stream(name: str) -> Dict[str, Any]: + return { + "type": "DeclarativeStream", + "name": name, + "primary_key": [], + "schema_loader": { + "type": "InlineSchemaLoader", + "schema": {"type": "object", "properties": {}}, + }, + "retriever": { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url_base": "https://api.test.com", + "path": f"/{name}", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, + }, + } + + return { + "version": "6.0.0", + "type": "DeclarativeSource", + "check": {"type": "CheckStream", "stream_names": ["a"]}, + "streams": [stream("a"), stream("b")], + } + + +def test_invalid_manifest_raises_the_best_match_error_not_the_first(): + manifest = _two_stream_schema_validation_manifest() + manifest["streams"][0]["primary_key"] = 5 + manifest["streams"][1]["retriever"]["requester"]["http_method"] = "FETCH" + schema = _get_declarative_component_schema() + + assert next(Draft7Validator(schema).iter_errors(manifest)).json_path == "$.streams[0]" + + with pytest.raises(ValidationError) as expected_error: + validate(manifest, schema) + expected = expected_error.value + + with pytest.raises(ValidationError) as exc_info: + ConcurrentDeclarativeSource( + source_config=manifest, config={}, catalog=create_catalog("a"), state=None + ) + + assert ( + exc_info.value.message + == "Validation against json schema defined in declarative_component_schema.yaml schema failed" + ) + cause = exc_info.value.__cause__ + assert cause is not None + assert cause.message == expected.message + assert cause.json_path == expected.json_path + assert list(cause.schema_path) == list(expected.schema_path) + assert cause.validator == expected.validator + assert cause.validator_value == expected.validator_value + assert cause.json_path == "$.streams[1].retriever.requester.http_method" + + +def test_building_a_source_does_not_check_the_component_schema(): + with patch.object( + Draft7Validator, + "check_schema", + side_effect=AssertionError("check_schema must not run at source build"), + ): + ConcurrentDeclarativeSource( + source_config=_two_stream_schema_validation_manifest(), + config={}, + catalog=create_catalog("a"), + state=None, + ) + + +def test_a_broken_component_schema_that_fails_the_manifest_raises_schema_error(): + manifest = _two_stream_schema_validation_manifest() + broken = copy.deepcopy(_get_declarative_component_schema()) + broken["definitions"]["HttpRequester"]["properties"]["http_method"]["enum"] = "GET" + + assert list(Draft7Validator(broken).iter_errors(manifest)) + + with patch( + "airbyte_cdk.sources.declarative.concurrent_declarative_source._get_declarative_component_schema", + return_value=broken, + ): + with pytest.raises(SchemaError): + ConcurrentDeclarativeSource( + source_config=manifest, config={}, catalog=create_catalog("a"), state=None + ) + + +def test_a_broken_component_schema_that_breaks_validation_raises_schema_error(): + manifest = _two_stream_schema_validation_manifest() + broken = copy.deepcopy(_get_declarative_component_schema()) + broken["definitions"]["HttpRequester"]["properties"]["path"]["type"] = "strng" + + with pytest.raises(UnknownType): + list(Draft7Validator(broken).iter_errors(manifest)) + + with patch( + "airbyte_cdk.sources.declarative.concurrent_declarative_source._get_declarative_component_schema", + return_value=broken, + ): + with pytest.raises(SchemaError): + ConcurrentDeclarativeSource( + source_config=manifest, config={}, catalog=create_catalog("a"), state=None + ) diff --git a/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py index db7d849bba..3f8dee86b9 100644 --- a/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py +++ b/unit_tests/sources/declarative/test_declarative_component_schema_is_loadable.py @@ -22,6 +22,8 @@ import pkgutil import re +from jsonschema.validators import Draft7Validator, validator_for + from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( _get_declarative_component_schema, ) @@ -51,6 +53,14 @@ def test_the_shipped_component_schema_parses(): assert "HttpRequester" in schema["definitions"] +def test_the_shipped_component_schema_is_a_valid_draft_7_schema(): + """`_validate_source` skips `check_schema` at runtime, so this is where a malformed schema is caught.""" + schema = _get_declarative_component_schema() + validator_class = validator_for(schema) + assert validator_class is Draft7Validator + validator_class.check_schema(schema) + + def test_no_plain_scalar_carries_yaml_punctuation(): """`": "` ends a plain scalar and `" #"` starts a comment, so either one truncates a value written as prose -- the first noisily, the second without a trace. Checked against the source