diff --git a/airbyte_cdk/sources/declarative/checks/check_stream.py b/airbyte_cdk/sources/declarative/checks/check_stream.py index aa3298b535..2aafc7e910 100644 --- a/airbyte_cdk/sources/declarative/checks/check_stream.py +++ b/airbyte_cdk/sources/declarative/checks/check_stream.py @@ -126,11 +126,14 @@ def _check_dynamic_streams_availability( logger: logging.Logger, ) -> Tuple[bool, Any]: """Checks the availability of dynamic streams.""" - dynamic_streams = source.resolved_manifest.get("dynamic_streams", []) # type: ignore[attr-defined] # The source's resolved_manifest manifest is checked before calling this method - dynamic_stream_name_to_dynamic_stream = { - ds.get("name", f"dynamic_stream_{i}"): ds for i, ds in enumerate(dynamic_streams) - } - generated_streams = self._map_generated_streams(source.dynamic_streams) # type: ignore[attr-defined] # The source's dynamic_streams manifest is checked before calling this method + try: + dynamic_streams = source.resolved_manifest.get("dynamic_streams", []) # type: ignore[attr-defined] # The source's resolved_manifest manifest is checked before calling this method + dynamic_stream_name_to_dynamic_stream = { + ds.get("name", f"dynamic_stream_{i}"): ds for i, ds in enumerate(dynamic_streams) + } + generated_streams = self._map_generated_streams(source.dynamic_streams) # type: ignore[attr-defined] # The source's dynamic_streams manifest is checked before calling this method + except Exception as error: + return self._log_error(logger, "resolving dynamic streams for check", error) for check_config in self.dynamic_streams_check_configs: # type: ignore[union-attr] # None value for self.dynamic_streams_check_configs handled in __post_init__ if check_config.dynamic_stream_name not in dynamic_stream_name_to_dynamic_stream: diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py index 317abb6c98..7cf279e0d8 100644 --- a/unit_tests/sources/declarative/checks/test_check_stream.py +++ b/unit_tests/sources/declarative/checks/test_check_stream.py @@ -13,7 +13,10 @@ from jsonschema.exceptions import ValidationError from airbyte_cdk.models import Status -from airbyte_cdk.sources.declarative.checks.check_stream import CheckStream +from airbyte_cdk.sources.declarative.checks.check_stream import ( + CheckStream, + DynamicStreamCheckConfig, +) from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( ConcurrentDeclarativeSource, ) @@ -692,3 +695,86 @@ def test_check_stream_only_type_provided(): ) with pytest.raises(ValueError): source.check(logger, _CONFIG) + + +class _RaisingManifest: + """Dict-like manifest whose `get` call raises, simulating a failure such as a + `ConfigComponentsResolver` error while resolving `dynamic_streams`.""" + + def get(self, *args, **kwargs): + raise RuntimeError("ConfigComponentsResolver failed to resolve manifest") + + +class _FailingResolvedManifestSource: + """Fake Source whose resolved_manifest fails when `dynamic_streams` is looked up.""" + + resolved_manifest = _RaisingManifest() + dynamic_streams: list = [] + + def __init__(self, streams): + self._streams = streams + + def streams(self, config): + return self._streams + + +class _FailingDynamicStreamsSource: + """Fake Source whose `dynamic_streams` property raises when accessed from inside + `_check_dynamic_streams_availability`. The first access (from the `hasattr` gate + in `check_connection`) succeeds; subsequent accesses raise, simulating a flaky + resolver or a resolver whose second call fails.""" + + resolved_manifest = {"dynamic_streams": [{"name": "some_dynamic_stream"}]} + + def __init__(self, streams): + self._streams = streams + self._access_count = 0 + + def streams(self, config): + return self._streams + + @property + def dynamic_streams(self): + self._access_count += 1 + if self._access_count > 1: + raise RuntimeError("dynamic stream generation failed") + return [] + + +@pytest.mark.parametrize( + "source_factory, expected_error_substring", + [ + pytest.param( + _FailingResolvedManifestSource, + "ConfigComponentsResolver failed to resolve manifest", + id="resolved_manifest_get_raises", + ), + pytest.param( + _FailingDynamicStreamsSource, + "dynamic stream generation failed", + id="dynamic_streams_property_raises", + ), + ], +) +def test_check_dynamic_streams_availability_returns_error_when_resolution_raises( + source_factory, expected_error_substring +): + """`_check_dynamic_streams_availability` must funnel resolution errors through + `_log_error` and return `(False, message)` rather than propagating the exception.""" + static_stream = MagicMock(spec=Stream) + static_stream.name = "static_stream" + source = source_factory([static_stream]) + + check_stream = CheckStream( + stream_names=[], + parameters={}, + dynamic_streams_check_configs=[ + DynamicStreamCheckConfig(dynamic_stream_name="some_dynamic_stream", stream_count=1) + ], + ) + + is_available, message = check_stream.check_connection(source, logger, config) + + assert is_available is False + assert "resolving dynamic streams for check" in message + assert expected_error_substring in message