diff --git a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py index 2c6d5c8bd..6519e4d5f 100644 --- a/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py +++ b/airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py @@ -12,7 +12,6 @@ import orjson from jsonschema.exceptions import ValidationError -from jsonschema.validators import validate from packaging.version import InvalidVersion, Version from airbyte_cdk.config_observation import create_connector_config_control_message @@ -34,6 +33,7 @@ from airbyte_cdk.models.airbyte_protocol_serializers import AirbyteMessageSerializer from airbyte_cdk.sources.declarative.checks import COMPONENTS_CHECKER_TYPE_MAPPING from airbyte_cdk.sources.declarative.checks.connection_checker import ConnectionChecker +from airbyte_cdk.sources.declarative.concurrent_declarative_source import validate_manifest from airbyte_cdk.sources.declarative.interpolation import InterpolatedBoolean from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( ConditionalStreams as ConditionalStreamsModel, @@ -447,12 +447,7 @@ def _validate_source(self) -> None: Validates the connector manifest against the declarative component schema """ - try: - validate(self._source_config, self._declarative_component_schema) - except ValidationError as e: - raise ValidationError( - "Validation against json schema defined in declarative_component_schema.yaml schema failed" - ) from e + validate_manifest(self._source_config, self._declarative_component_schema) 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 1fafa07a6..24803a72b 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -21,8 +21,8 @@ import orjson 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 ( @@ -122,7 +122,7 @@ class TestLimits: max_streams: int = field(default=DEFAULT_MAX_STREAMS) -def _get_declarative_component_schema() -> Dict[str, Any]: +def get_declarative_component_schema() -> Dict[str, Any]: try: raw_component_schema = pkgutil.get_data( "airbyte_cdk", "sources/declarative/declarative_component_schema.yaml" @@ -140,6 +140,28 @@ def _get_declarative_component_schema() -> Dict[str, Any]: ) +def validate_manifest(manifest: Mapping[str, Any], schema: Mapping[str, Any]) -> None: + """Validate a manifest against the shipped component schema. + + Unit tests check the schema's metaschema, so valid manifests skip `check_schema`. On failure, + check the schema first to surface `SchemaError`, while preserving `jsonschema.validate`'s + best-match error as the cause with a traceback. + """ + validator_class = validator_for(schema) + try: + error = best_match(validator_class(schema).iter_errors(manifest)) + if error is not None: + raise error + except ValidationError as e: + validator_class.check_schema(schema) + raise ValidationError( + "Validation against json schema defined in declarative_component_schema.yaml schema failed" + ) from e + except Exception: + validator_class.check_schema(schema) + raise + + class ConcurrentDeclarativeSource(Source): # By default, we defer to a value of 2. A value lower than could cause a PartitionEnqueuer to be stuck in a state of deadlock # because it has hit the limit of futures but not partition reader is consuming them. @@ -204,7 +226,7 @@ def __init__( self._should_normalize = normalize_manifest self._should_migrate = migrate_manifest - self._declarative_component_schema = _get_declarative_component_schema() + self._declarative_component_schema = get_declarative_component_schema() # If custom components are needed, locate and/or register them. self.components_module: ModuleType | None = get_registered_components_module(config=config) # set additional attributes @@ -345,12 +367,7 @@ def _validate_source(self) -> None: Validates the connector manifest against the declarative component schema """ - try: - validate(self._source_config, self._declarative_component_schema) - except ValidationError as e: - raise ValidationError( - "Validation against json schema defined in declarative_component_schema.yaml schema failed" - ) from e + validate_manifest(self._source_config, self._declarative_component_schema) 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 a2db6ea39..b584f8b71 100644 --- a/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py +++ b/unit_tests/legacy/sources/declarative/test_manifest_declarative_source.py @@ -15,6 +15,7 @@ import requests import yaml from jsonschema.exceptions import ValidationError +from jsonschema.validators import validator_for 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, @@ -43,6 +45,9 @@ from airbyte_cdk.sources.declarative.retrievers.simple_retriever import SimpleRetriever from airbyte_cdk.sources.streams.concurrent.default_stream import DefaultStream from unit_tests.sources.declarative.parsers.test_model_to_component_factory import get_retriever +from unit_tests.sources.declarative.test_concurrent_declarative_source import ( + two_stream_schema_validation_manifest, +) logger = logging.getLogger("airbyte") @@ -2550,3 +2555,28 @@ 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." ) + + +@pytest.mark.parametrize("migrate_manifest", [False, True]) +def test_building_a_source_does_not_check_the_component_schema(migrate_manifest): + with patch.object( + validator_for(get_declarative_component_schema()), + "check_schema", + side_effect=AssertionError("check_schema must not run at source build"), + ) as check_schema: + ManifestDeclarativeSource( + source_config=two_stream_schema_validation_manifest(), + migrate_manifest=migrate_manifest, + ) + check_schema.assert_not_called() + + +def test_legacy_source_delegates_to_public_manifest_validator(): + with patch( + "airbyte_cdk.legacy.sources.declarative.manifest_declarative_source.validate_manifest" + ) as validate_manifest: + source = ManifestDeclarativeSource(source_config=two_stream_schema_validation_manifest()) + + validate_manifest.assert_called_once_with( + source._source_config, source._declarative_component_schema + ) diff --git a/unit_tests/sources/declarative/parsers/test_manifest_normalizer.py b/unit_tests/sources/declarative/parsers/test_manifest_normalizer.py index 3e934db79..246a840b1 100644 --- a/unit_tests/sources/declarative/parsers/test_manifest_normalizer.py +++ b/unit_tests/sources/declarative/parsers/test_manifest_normalizer.py @@ -4,7 +4,7 @@ from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( - _get_declarative_component_schema, + get_declarative_component_schema, ) from airbyte_cdk.sources.declarative.parsers.manifest_normalizer import ( ManifestNormalizer, @@ -25,7 +25,7 @@ def test_when_multiple_url_base_are_resolved_and_most_frequent_is_shared( and the most frequent one is shared. """ - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() resolved_manifest = resolver.preprocess_manifest(manifest_with_multiple_url_base) normalized_manifest = ManifestNormalizer(resolved_manifest, schema).normalize() @@ -41,7 +41,7 @@ def test_with_shared_definitions_url_base_are_present( between the definitions and the `url_base` is present in the manifest. """ - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() resolved_manifest = resolver.preprocess_manifest(manifest_with_url_base_linked_definition) normalized_manifest = ManifestNormalizer(resolved_manifest, schema).normalize() @@ -75,7 +75,7 @@ def test_clean_dangling_fields_removes_unreferenced_definitions() -> None: } ], } - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._clean_dangling_fields() assert normalizer._normalized_manifest == expected @@ -100,7 +100,7 @@ def test_clean_dangling_fields_removes_unreferenced_schemas() -> None: {"name": "stream1", "schema_loader": {"schema": {"$ref": "#/schemas/referenced"}}} ], } - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._clean_dangling_fields() assert normalizer._normalized_manifest == expected @@ -134,7 +134,7 @@ def test_clean_dangling_fields_keeps_parent_paths() -> None: } ], } - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._clean_dangling_fields() assert normalizer._normalized_manifest == expected @@ -150,7 +150,7 @@ def test_clean_dangling_fields_removes_empty_sections() -> None: "streams": [{"name": "stream1", "type": "object", "properties": {}}], } expected = {"streams": [{"name": "stream1", "type": "object", "properties": {}}]} - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._clean_dangling_fields() assert normalizer._normalized_manifest == expected @@ -220,7 +220,7 @@ def test_replace_parent_streams_with_refs_replaces_with_ref(): }, ] } - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._replace_parent_streams_with_refs() assert normalizer._normalized_manifest == expected @@ -276,7 +276,7 @@ def test_replace_parent_streams_with_refs_no_match(): }, ] } - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._replace_parent_streams_with_refs() assert normalizer._normalized_manifest == expected @@ -360,7 +360,7 @@ def test_replace_parent_streams_with_refs_handles_multiple_partition_routers(): }, ] } - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() normalizer = ManifestNormalizer(manifest, schema) normalizer._replace_parent_streams_with_refs() assert normalizer._normalized_manifest == expected diff --git a/unit_tests/sources/declarative/test_concurrent_declarative_source.py b/unit_tests/sources/declarative/test_concurrent_declarative_source.py index 1b8f3fa7d..dc82ca4d2 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, validator_for from typing_extensions import deprecated import unit_tests.sources.declarative.external_component # Needed for dynamic imports to work @@ -54,6 +55,7 @@ from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( ConcurrentDeclarativeSource, TestLimits, + get_declarative_component_schema, ) from airbyte_cdk.sources.declarative.extractors.record_filter import ( ClientSideIncrementalRecordFilterDecorator, @@ -7146,3 +7148,143 @@ 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() + + first = next(Draft7Validator(schema).iter_errors(manifest)) + + 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.__traceback__ 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 != first.json_path + + +@pytest.mark.parametrize("migrate_manifest", [False, True]) +def test_building_a_source_does_not_check_the_component_schema(migrate_manifest): + with patch.object( + validator_for(get_declarative_component_schema()), + "check_schema", + side_effect=AssertionError("check_schema must not run at source build"), + ) as check_schema: + ConcurrentDeclarativeSource( + source_config=two_stream_schema_validation_manifest(), + config={}, + catalog=create_catalog("a"), + state=None, + migrate_manifest=migrate_manifest, + ) + check_schema.assert_not_called() + + +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"]["required"] = "type" + + 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_component_schema_that_passes_check_schema_but_breaks_validation_reraises(): + manifest = two_stream_schema_validation_manifest() + broken = copy.deepcopy(get_declarative_component_schema()) + broken["definitions"]["HttpRequester"]["properties"]["path"] = { + "$ref": "#/definitions/DoesNotExist" + } + + Draft7Validator.check_schema(broken) + with pytest.raises(Exception) as expected: + 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(type(expected.value)) as exc_info: + ConcurrentDeclarativeSource( + source_config=manifest, config={}, catalog=create_catalog("a"), state=None + ) + + assert type(exc_info.value) is type(expected.value) + + +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 db7d849bb..bdab6a30d 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 @@ -4,7 +4,7 @@ """The shipped manifest schema has to parse, and only 194 unrelated failures said so. -`_get_declarative_component_schema()` runs before any low-code source is built, so a malformed +`get_declarative_component_schema()` runs before any low-code source is built, so a malformed `declarative_component_schema.yaml` fails every declarative connector at startup. It is also easy to break from a documentation edit alone: the descriptions are plain YAML scalars, so a `": "` in prose ends the scalar and the file stops being a mapping. That happened in this file's history and @@ -12,18 +12,21 @@ at the wrong layer and with unreadable output. What these guards add is not coverage but a named failure in milliseconds. -Two guards, because the two ways prose breaks a plain scalar look nothing alike. A `": "` is loud: -the file stops parsing, so loading it is enough to catch it. A `" #"` is silent: YAML reads the -rest of the line as a comment, the value keeps parsing as a shorter string, and nothing in the +Three guards, because the two ways prose breaks a plain scalar look nothing alike. A `": "` is +loud: the file stops parsing, so loading it is enough to catch it. A `" #"` is silent: YAML reads +the rest of the line as a comment, the value keeps parsing as a shorter string, and nothing in the parsed tree records that a sentence went missing. Only the source text knows, so the second guard -reads that instead of the loaded schema. +reads that instead of the loaded schema. The third validates the parsed schema against Draft 7, +since `_validate_source` checks it only when a manifest fails validation. """ import pkgutil import re +from jsonschema.validators import Draft7Validator, validator_for + from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( - _get_declarative_component_schema, + get_declarative_component_schema, ) # A plain scalar is one not opened with a quote, a block indicator or a flow collection, and it is @@ -45,12 +48,20 @@ def test_the_shipped_component_schema_parses(): - schema = _get_declarative_component_schema() + schema = get_declarative_component_schema() assert schema["title"] == "DeclarativeSource" assert "HttpRequester" in schema["definitions"] +def test_the_shipped_component_schema_is_a_valid_draft_7_schema(): + """`_validate_source` runs `check_schema` only when a manifest fails validation, so a malformed schema that still accepts manifests is caught only here.""" + 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 diff --git a/unit_tests/utils/test_yaml_loader.py b/unit_tests/utils/test_yaml_loader.py index 13ee02e17..882769672 100644 --- a/unit_tests/utils/test_yaml_loader.py +++ b/unit_tests/utils/test_yaml_loader.py @@ -16,7 +16,7 @@ _get_declarative_component_schema as _legacy_get_declarative_component_schema, ) from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( - _get_declarative_component_schema, + get_declarative_component_schema, ) from airbyte_cdk.sources.declarative.yaml_declarative_source import YamlDeclarativeSource from airbyte_cdk.utils.yaml_loader import _get_safe_loader, safe_load_yaml @@ -77,7 +77,7 @@ def test_component_schema_parses_identically_with_both_loaders() -> None: c_loader_schema = yaml.load(raw, Loader=yaml.CSafeLoader) assert pure_python_schema == c_loader_schema assert repr(pure_python_schema) == repr(c_loader_schema) - assert c_loader_schema == _get_declarative_component_schema() + assert c_loader_schema == get_declarative_component_schema() @requires_libyaml @@ -107,7 +107,7 @@ def __init__(self, stream: Any) -> None: monkeypatch.setattr(yaml, "CSafeLoader", _Spy) - _get_declarative_component_schema() + get_declarative_component_schema() assert len(instantiated) == 1 _legacy_get_declarative_component_schema()