diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index d7e5e4260..52f83a644 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -3,6 +3,7 @@ import json import logging import pkgutil +from contextlib import contextmanager from copy import deepcopy from dataclasses import dataclass, field from queue import Queue @@ -101,6 +102,7 @@ DebugSliceLogger, SliceLogger, ) +from airbyte_cdk.utils.airbyte_secrets_utils import add_to_secrets, get_secrets from airbyte_cdk.utils.stream_status_utils import as_airbyte_message from airbyte_cdk.utils.traced_exception import AirbyteTracedException @@ -143,6 +145,13 @@ class ConcurrentDeclarativeSource(Source): # because it has hit the limit of futures but not partition reader is consuming them. _LOWEST_SAFE_CONCURRENCY_LEVEL = 2 + # Component types that hold the connector config and can write it back. `refresh_token_updater` is + # declared on `OAuthAuthenticator` alone; `CustomAuthenticator` is the type the transformer injects + # for a `class_name` component. + _CONFIG_PERSISTING_AUTHENTICATOR_TYPES = frozenset( + {"OAuthAuthenticator", "CustomAuthenticator"} + ) + def __init__( self, catalog: Optional[ConfiguredAirbyteCatalog] = None, @@ -228,6 +237,10 @@ def __init__( self._constructor.create_component(SpecModel, spec, dict()) if spec else None ) self._config = self._migrate_and_transform_config(config_path, config) or {} + # `check` may temporarily overlay values onto `self._config` (see + # `_config_overridden_for_check`). The manifest's `config_validations` express intent about the + # config as supplied, so they must run against it rather than against a check-time overlay. + self._config_for_validation = self._config concurrency_level_from_manifest = self._source_config.get("concurrency_level") if concurrency_level_from_manifest: @@ -413,7 +426,7 @@ def streams(self, config: Mapping[str, Any]) -> List[AbstractStream]: # type: i """ if self._spec_component: - self._spec_component.validate_config(self._config) + self._spec_component.validate_config(self._config_for_validation) api_budget_model = self._source_config.get("api_budget") if api_budget_model: @@ -607,11 +620,219 @@ def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteCon f"Expected to generate a ConnectionChecker component, but received {connection_checker.__class__}" ) - check_succeeded, error = connection_checker.check_connection(self, logger, self._config) + with self._config_overridden_for_check(check.get("config_overrides")): + check_succeeded, error = connection_checker.check_connection(self, logger, self._config) if not check_succeeded: return AirbyteConnectionStatus(status=Status.FAILED, message=repr(error)) return AirbyteConnectionStatus(status=Status.SUCCEEDED) + @contextmanager + def _config_overridden_for_check( + self, config_overrides: Optional[Mapping[str, Any]] + ) -> Iterator[None]: + """Overlay the check component's `config_overrides` onto the config for the duration of a check. + + Rebinding `self._config` reaches the whole component tree, because `streams()` interpolates from + it and ignores its own `config` argument. Not thread safe; `check` is one command per process. + + Semantics: + - values are applied verbatim, never interpolated; + - the merge is one level deep, so an object-valued override replaces rather than merges; + - it happens after `_migrate_and_transform_config`, so derived fields are not recomputed; + - `config_validations` run against `self._config_for_validation`, not the overlay. + + The copy is shallow, so a component writing into a nested path writes through to the config the + source was constructed with and the restore does not undo it. Only top-level writes are + discarded, which is why `_raise_if_config_is_persisted` refuses config-persisting manifests. + + Known limitation: `$parameters` declared on a check component still propagate into object-valued + overrides. See https://github.com/airbytehq/airbyte-internal-issues/issues/16995. + """ + if not config_overrides: + yield + return + + self._raise_on_non_string_override_keys(config_overrides) + self._raise_on_reserved_override_keys(config_overrides) + self._raise_if_config_is_persisted(config_overrides) + self._warn_on_unknown_override_keys(config_overrides) + self._register_override_secrets(config_overrides) + # Keys only. An override may name a secret field, so values must not reach the logs. + self.logger.info( + f"Overriding config keys for the check operation: {', '.join(sorted(config_overrides))}" + ) + + unmodified_config = self._config + self._config = {**self._config, **config_overrides} + try: + yield + finally: + self._config = unmodified_config + + @staticmethod + def _raise_on_reserved_override_keys(config_overrides: Mapping[str, Any]) -> None: + """Refuse keys in the platform's reserved `__airbyte` namespace. + + `CheckStream` reads `__airbyte_check_stream_names` out of the config this overlay writes to, so + an override there would change which streams a check tests. The whole prefix is refused. + """ + reserved = sorted(key for key in config_overrides if key.startswith("__airbyte")) + if reserved: + raise AirbyteTracedException( + message=( + f"This connector's manifest is invalid: its check component overrides the " + f"reserved config key(s) {reserved}. Keys prefixed with `__airbyte` belong to " + "the platform, not to the connector's spec. This is a bug in the connector " + "rather than in this connection's settings." + ), + internal_message=( + f"config_overrides rejected: reserved __airbyte key(s) {reserved}. " + "`CheckStream` reads `__airbyte_check_stream_names` out of the config this " + "overlay writes to, so an override there would change which streams a check " + "tests. Use `stream_names` instead." + ), + failure_type=FailureType.system_error, + ) + + @staticmethod + def _raise_on_non_string_override_keys(config_overrides: Mapping[str, Any]) -> None: + """Reject non-string keys, which YAML allows and every consumer below assumes away.""" + non_strings = [key for key in config_overrides if not isinstance(key, str)] + if non_strings: + raise AirbyteTracedException( + message=( + "This connector's manifest is invalid: its check component has " + f"`config_overrides` key(s) {sorted(map(repr, non_strings))} that are not " + "strings, so they cannot name a field in the connector's spec. This is a bug " + "in the connector rather than in this connection's settings." + ), + internal_message=( + "config_overrides rejected: non-string key(s) " + f"{sorted(map(repr, non_strings))}. Quote them in the manifest so they name a " + "field in the connector's spec." + ), + failure_type=FailureType.system_error, + ) + + def _raise_if_config_is_persisted(self, config_overrides: Mapping[str, Any]) -> None: + """Reject `config_overrides` on a manifest whose authenticator writes the config back. + + A `refresh_token_updater` builds a `DeclarativeSingleUseRefreshTokenOauth2Authenticator`, whose + `_emit_control_message` emits the entire config it was handed as a `CONNECTOR_CONFIG` message. + During a check that is the overlay, and the platform persists it, so a check-only override would + become the connection's saved config. The restore cannot recall a message already on stdout. + """ + if not self._manifest_writes_back_config(self._source_config): + return + raise AirbyteTracedException( + message=( + "This connector's manifest is invalid: `config_overrides` cannot be used by a " + "manifest that declares a `refresh_token_updater`. A token refresh during `check` " + "emits the whole config it was handed as a CONNECTOR_CONFIG control message, which " + f"the platform persists, so the check-only override(s) {sorted(config_overrides)} " + "would be saved as this connection's config and applied to every later sync. This " + "is a bug in the connector rather than in this connection's settings." + ), + internal_message=( + "config_overrides rejected: manifest persists config via refresh_token_updater. " + "Remove `config_overrides` from the check component, or drop the " + "`refresh_token_updater`." + ), + failure_type=FailureType.system_error, + ) + + @staticmethod + def _manifest_writes_back_config(definition: Any) -> bool: + """Whether any component in the manifest emits the connector config back to the platform. + + Walks the whole manifest, since an authenticator can sit under any requester, inside a + `SelectiveAuthenticator`, or in a `ConditionalStreams` branch. Callers pass `self._source_config`, + which is post-`_pre_process_manifest`, so a `$ref`-ed authenticator is already inlined and + carries its own `type`. + + `is not None` rather than truthiness because `refresh_token_updater: {}` takes every default and + still builds a single-use authenticator - the factory tests a model instance, which is truthy. + The type condition is what separates a component from a data blob that happens to use the name. + + Does not detect a `CustomAuthenticator` that persists config without declaring the field. + """ + if isinstance(definition, Mapping): + if ( + definition.get("refresh_token_updater") is not None + and definition.get("type") + in ConcurrentDeclarativeSource._CONFIG_PERSISTING_AUTHENTICATOR_TYPES + ): + return True + return any( + ConcurrentDeclarativeSource._manifest_writes_back_config(value) + for value in definition.values() + ) + if isinstance(definition, list): + return any( + ConcurrentDeclarativeSource._manifest_writes_back_config(item) + for item in definition + ) + return False + + def _warn_on_unknown_override_keys(self, config_overrides: Mapping[str, Any]) -> None: + """Warn about override keys the spec does not declare. + + Nothing validates the overlay - the entrypoint validates what the user supplied, and + `config_validations` run against `self._config_for_validation` - so a typo is a silent no-op. + """ + if not self._spec_component: + return + declared = self._declared_config_properties(self._spec_component.connection_specification) + if declared is None: + return + unknown = sorted(key for key in config_overrides if key not in declared) + if unknown: + self.logger.warning( + f"Check-only config override(s) {unknown} are not declared in the connector spec, so " + "they will have no effect on any component that reads the config by field name." + ) + + @staticmethod + def _declared_config_properties( + connection_specification: Mapping[str, Any], + ) -> Optional[Set[str]]: + """Every field name a spec declares, or `None` when it does not enumerate them. + + `oneOf` and `allOf` composition put declarations one level down, so reading only the top-level + `properties` would report a declared field as undeclared. + """ + if not isinstance(connection_specification, Mapping): + return None + + names: Set[str] = set() + found_any = False + properties = connection_specification.get("properties") + if isinstance(properties, Mapping): + found_any = True + names.update(str(key) for key in properties) + for keyword in ("allOf", "anyOf", "oneOf"): + for branch in connection_specification.get(keyword) or []: + nested = ConcurrentDeclarativeSource._declared_config_properties(branch) + if nested is not None: + found_any = True + names.update(nested) + return names if found_any else None + + def _register_override_secrets(self, config_overrides: Mapping[str, Any]) -> None: + """Register override values landing on an `airbyte_secret` field, so they get redacted. + + The entrypoint builds the secret list from the config the user supplied, so a value substituted + here is unknown to `filter_secrets`. Uses the same discovery as the entrypoint, so the overlay is + redacted exactly where the user's own value at that path would be. + """ + if not self._spec_component: + return + for secret in get_secrets( + self._spec_component.connection_specification, dict(config_overrides) + ): + if secret is not None: + add_to_secrets(str(secret)) + @property def dynamic_streams(self) -> List[Dict[str, Any]]: return self._dynamic_stream_configs( diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 45c6cbfde..016e78bee 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -558,6 +558,14 @@ definitions: type: array items: "$ref": "#/definitions/DynamicStreamCheckConfig" + config_overrides: + title: Config Overrides + description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, a `$ref` inside them is not resolved, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. Keys must be strings, and two combinations are rejected outright. Keys prefixed with `__airbyte` belong to the platform rather than to the connector's spec. And a manifest that declares a `refresh_token_updater` cannot use this field at all, because a token refresh during check emits the whole config it was handed as a CONNECTOR_CONFIG control message, which the platform persists - so a check-only override would become the connection's saved config and apply to every later sync. + type: object + additionalProperties: true + examples: + - max_waiting_time: 0 + - page_size: 1 DynamicStreamCheckConfig: type: object required: @@ -596,6 +604,14 @@ definitions: description: Enables stream check availability. This field is automatically set by the CDK. type: boolean default: true + config_overrides: + title: Config Overrides + description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, a `$ref` inside them is not resolved, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. Keys must be strings, and two combinations are rejected outright. Keys prefixed with `__airbyte` belong to the platform rather than to the connector's spec. And a manifest that declares a `refresh_token_updater` cannot use this field at all, because a token refresh during check emits the whole config it was handed as a CONNECTOR_CONFIG control message, which the platform persists - so a check-only override would become the connection's saved config and apply to every later sync. + type: object + additionalProperties: true + examples: + - max_waiting_time: 0 + - page_size: 1 CompositeErrorHandler: title: Composite Error Handler description: Error handler that sequentially iterates over a list of error handlers. diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 1726c80aa..d09848a23 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -77,6 +77,12 @@ class CheckDynamicStream(BaseModel): description="Enables stream check availability. This field is automatically set by the CDK.", title="Use Check Availability", ) + config_overrides: Optional[Dict[str, Any]] = Field( + None, + description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, a `$ref` inside them is not resolved, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. Keys must be strings, and two combinations are rejected outright. Keys prefixed with `__airbyte` belong to the platform rather than to the connector's spec. And a manifest that declares a `refresh_token_updater` cannot use this field at all, because a token refresh during check emits the whole config it was handed as a CONNECTOR_CONFIG control message, which the platform persists - so a check-only override would become the connection's saved config and apply to every later sync.", + examples=[{"max_waiting_time": 0}, {"page_size": 1}], + title="Config Overrides", + ) class ConcurrencyLevel(BaseModel): @@ -1794,6 +1800,12 @@ class CheckStream(BaseModel): title="Stream Names", ) dynamic_streams_check_configs: Optional[List[DynamicStreamCheckConfig]] = None + config_overrides: Optional[Dict[str, Any]] = Field( + None, + description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, a `$ref` inside them is not resolved, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. Keys must be strings, and two combinations are rejected outright. Keys prefixed with `__airbyte` belong to the platform rather than to the connector's spec. And a manifest that declares a `refresh_token_updater` cannot use this field at all, because a token refresh during check emits the whole config it was handed as a CONNECTOR_CONFIG control message, which the platform persists - so a check-only override would become the connection's saved config and apply to every later sync.", + examples=[{"max_waiting_time": 0}, {"page_size": 1}], + title="Config Overrides", + ) class IncrementingCountCursor(BaseModel): diff --git a/airbyte_cdk/sources/declarative/parsers/manifest_reference_resolver.py b/airbyte_cdk/sources/declarative/parsers/manifest_reference_resolver.py index 1c5ae0485..0294171d9 100644 --- a/airbyte_cdk/sources/declarative/parsers/manifest_reference_resolver.py +++ b/airbyte_cdk/sources/declarative/parsers/manifest_reference_resolver.py @@ -3,6 +3,7 @@ # import re +from copy import deepcopy from typing import Any, Dict, Mapping, Set, Tuple, Union from airbyte_cdk.sources.declarative.parsers.custom_exceptions import ( @@ -12,6 +13,11 @@ REF_TAG = "$ref" +# Manifest fields whose values are connector config rather than components. Any string starting with +# `#/` is a reference, so without this a config value shaped like a pointer would be resolved - or, if +# it resolves to nothing, would raise during preprocessing and break every command. +_FIELDS_HOLDING_CONFIG_VALUES = frozenset({"config_overrides"}) + class ManifestReferenceResolver: """ @@ -109,7 +115,9 @@ def preprocess_manifest(self, manifest: Mapping[str, Any]) -> Dict[str, Any]: def _evaluate_node(self, node: Any, manifest: Mapping[str, Any], visited: Set[Any]) -> Any: if isinstance(node, dict): evaluated_dict = { - k: self._evaluate_node(v, manifest, visited) + k: deepcopy(v) + if k in _FIELDS_HOLDING_CONFIG_VALUES + else self._evaluate_node(v, manifest, visited) for k, v in node.items() if not self._is_ref_key(k) } diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index 7cd635944..62e0a3e62 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -1307,6 +1307,9 @@ def create_check_stream( else [] ) + # `model.config_overrides` is deliberately not read here. The source applies it around the whole + # check operation (`ConcurrentDeclarativeSource._config_overridden_for_check`), which is what makes + # it work for every checker type rather than only this one. Do not wire it in a second time. return CheckStream( stream_names=model.stream_names or [], dynamic_streams_check_configs=dynamic_streams_check_configs, @@ -1321,6 +1324,7 @@ def create_check_dynamic_stream( use_check_availability = model.use_check_availability + # See `create_check_stream`: `model.config_overrides` is applied by the source, not here. return CheckDynamicStream( stream_count=model.stream_count, use_check_availability=use_check_availability, diff --git a/unit_tests/sources/declarative/test_concurrent_declarative_source_config_overrides.py b/unit_tests/sources/declarative/test_concurrent_declarative_source_config_overrides.py new file mode 100644 index 000000000..7c9492a20 --- /dev/null +++ b/unit_tests/sources/declarative/test_concurrent_declarative_source_config_overrides.py @@ -0,0 +1,928 @@ +# +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +# + +"""Tests for `ConcurrentDeclarativeSource._config_overridden_for_check` and its guards. + +The behaviour under test lives in `ConcurrentDeclarativeSource`, not in the check components - the +check factories deliberately ignore `model.config_overrides`. Kept beside +`test_concurrent_declarative_source.py` rather than inside it because that module is already 6k lines. +""" + +import json +import logging +import pkgutil +from copy import deepcopy + +import pytest +import yaml + +from airbyte_cdk.entrypoint import AirbyteEntrypoint +from airbyte_cdk.models import ConnectorSpecification, FailureType, Status, Type +from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( + ConcurrentDeclarativeSource, +) +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + CheckDynamicStream as CheckDynamicStreamModel, +) +from airbyte_cdk.sources.declarative.models.declarative_component_schema import ( + CheckStream as CheckStreamModel, +) +from airbyte_cdk.test.mock_http import HttpMocker, HttpRequest, HttpResponse +from airbyte_cdk.utils.traced_exception import AirbyteTracedException + +logger = logging.getLogger("test") + + +_CONFIG_DRIVEN_PATH_CONFIG = {"resource": "sync"} + +_MANIFEST_WITH_CONFIG_DRIVEN_PATH = { + "version": "6.7.0", + "type": "DeclarativeSource", + "check": {"type": "CheckStream", "stream_names": ["items"]}, + "streams": [ + { + "type": "DeclarativeStream", + "name": "items", + "primary_key": "id", + "schema_loader": { + "type": "InlineSchemaLoader", + "schema": { + "$schema": "http://json-schema.org/schema#", + "type": "object", + "properties": {"id": {"type": "integer"}}, + }, + }, + "retriever": { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url": "https://api.test.com/{{ config['resource'] }}", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, + "paginator": {"type": "NoPagination"}, + }, + } + ], +} + + +def _source_with_check_component(check_component): + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = check_component + return ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + +def test_given_no_config_overrides_when_check_then_components_use_the_user_config(): + source = _source_with_check_component({"type": "CheckStream", "stream_names": ["items"]}) + + with HttpMocker() as http_mocker: + # Only the user-configured path is mocked, so a request to any other path fails the test. + http_mocker.get( + HttpRequest(url="https://api.test.com/sync"), + HttpResponse(body=json.dumps([{"id": 1}])), + ) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + + +def test_given_config_overrides_when_check_then_components_built_during_check_see_them(): + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + ) + + with HttpMocker() as http_mocker: + # Only the overridden path is mocked. Reaching this endpoint proves the overlay was applied to + # the stream the checker built, and not merely stored on the source. + overridden_request = HttpRequest(url="https://api.test.com/check-only") + http_mocker.get(overridden_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + http_mocker.assert_number_of_calls(overridden_request, 1) + + +def test_given_config_overrides_when_check_then_the_config_is_restored_afterwards(): + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + ) + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/check-only"), + HttpResponse(body=json.dumps([{"id": 1}])), + ) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + + assert source._config == _CONFIG_DRIVEN_PATH_CONFIG + + # A sync that follows a check in the same process must go back to the user's value. + with HttpMocker() as http_mocker: + sync_request = HttpRequest(url="https://api.test.com/sync") + http_mocker.get(sync_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + stream = source.streams(_CONFIG_DRIVEN_PATH_CONFIG)[0] + assert stream.check_availability().is_available + + http_mocker.assert_number_of_calls(sync_request, 1) + + +def test_given_config_overrides_when_check_raises_then_the_config_is_restored(): + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["not_in_the_catalog"], + "config_overrides": {"resource": "check-only"}, + } + ) + + with pytest.raises(ValueError): + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + + assert source._config == _CONFIG_DRIVEN_PATH_CONFIG + + +def test_given_config_overrides_when_check_then_config_validations_run_against_the_user_config(): + """An override is authored in the manifest, so it must not be held to validations written for the + user's own input - the user has no way to satisfy them.""" + config = {"resource": "sync", "settings": {"mode": "sync"}} + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only", "settings": {"mode": "check-only"}}, + } + manifest["spec"] = { + "type": "Spec", + "connection_specification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": { + "resource": {"type": "string"}, + "settings": {"type": "object"}, + }, + }, + "config_normalization_rules": { + "type": "ConfigNormalizationRules", + "validations": [ + { + "type": "DpathValidator", + "field_path": ["settings"], + "validation_strategy": { + "type": "ValidateAdheresToSchema", + "base_schema": { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": {"mode": {"type": "string", "enum": ["sync"]}}, + "required": ["mode"], + "additionalProperties": False, + }, + }, + } + ], + }, + } + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=config, + catalog=None, + state=None, + ) + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/check-only"), + HttpResponse(body=json.dumps([{"id": 1}])), + ) + + # `settings.mode` is overridden to a value outside the validator's enum, so this would fail were + # the overlay validated instead of the config the user supplied. + assert source.check(logger, config).status == Status.SUCCEEDED + + +def test_given_config_overrides_when_check_then_values_are_not_interpolated(): + """Pins the verbatim contract. A value containing `{{ }}` reaches components as that literal string, + so turning interpolation on later is a deliberate, test-breaking decision rather than a silent + reinterpretation of overrides already written.""" + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "{{config['resource']}}"}, + } + ) + + with HttpMocker() as http_mocker: + literal_request = HttpRequest(url="https://api.test.com/%7B%7Bconfig%5B'resource'%5D%7D%7D") + http_mocker.get(literal_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + http_mocker.assert_number_of_calls(literal_request, 1) + + +_MANIFEST_WITH_CONFIG_DRIVEN_DYNAMIC_STREAM = { + "version": "6.7.0", + "type": "DeclarativeSource", + "check": {"type": "CheckDynamicStream", "stream_count": 1}, + "streams": [], + "dynamic_streams": [ + { + "type": "DynamicDeclarativeStream", + "name": "dynamic_items", + "stream_template": { + "type": "DeclarativeStream", + "name": "", + "primary_key": [], + "schema_loader": { + "type": "InlineSchemaLoader", + "schema": { + "$schema": "http://json-schema.org/schema#", + "type": "object", + "properties": {"id": {"type": "integer"}}, + }, + }, + "retriever": { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url": "https://api.test.com/{{ config['resource'] }}", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, + "paginator": {"type": "NoPagination"}, + }, + }, + "components_resolver": { + "type": "ConfigComponentsResolver", + "stream_config": { + "type": "StreamConfig", + "configs_pointer": ["custom_streams"], + }, + "components_mapping": [ + { + "type": "ComponentMappingDefinition", + "field_path": ["name"], + "value": "{{components_values['name']}}", + } + ], + }, + } + ], +} + + +def test_given_config_overrides_on_check_dynamic_stream_then_components_see_them(): + """The overlay is read from the raw check definition, so it is checker-agnostic. Without this test a + refactor moving the read into `create_check_stream` would silently drop `CheckDynamicStream`.""" + config = {"resource": "sync", "custom_streams": [{"name": "items"}]} + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_DYNAMIC_STREAM) + manifest["check"] = { + "type": "CheckDynamicStream", + "stream_count": 1, + "config_overrides": {"resource": "check-only"}, + } + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=config, + catalog=None, + state=None, + ) + + with HttpMocker() as http_mocker: + overridden_request = HttpRequest(url="https://api.test.com/check-only") + http_mocker.get(overridden_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, config).status == Status.SUCCEEDED + http_mocker.assert_number_of_calls(overridden_request, 1) + + +_OAUTH_WITH_REFRESH_TOKEN_UPDATER = { + "type": "OAuthAuthenticator", + "token_refresh_endpoint": "https://api.test.com/oauth/token", + "client_id": "{{ config['credentials']['client_id'] }}", + "client_secret": "{{ config['credentials']['client_secret'] }}", + "refresh_token": "{{ config['credentials']['refresh_token'] }}", + "refresh_token_updater": {"type": "RefreshTokenUpdater", "refresh_token_name": "refresh_token"}, +} + + +def _manifest_with_refresh_token_updater(check_component, refresh_token_updater=None): + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = check_component + authenticator = deepcopy(_OAUTH_WITH_REFRESH_TOKEN_UPDATER) + if refresh_token_updater is not None: + authenticator["refresh_token_updater"] = deepcopy(refresh_token_updater) + manifest["streams"][0]["retriever"]["requester"]["authenticator"] = authenticator + return manifest + + +@pytest.mark.parametrize( + "refresh_token_updater", + [ + pytest.param( + {"type": "RefreshTokenUpdater", "refresh_token_name": "refresh_token"}, id="populated" + ), + # Every field of `RefreshTokenUpdater` has a default, so an empty mapping is a valid way to take + # all of them. It builds the same single-use authenticator a populated one does, and the + # transformer injects no `type` into it, so it stays falsy - a truthiness test would miss it. + pytest.param({}, id="empty-taking-all-defaults"), + ], +) +def test_given_refresh_token_updater_when_config_overrides_then_manifest_is_rejected( + refresh_token_updater, +): + """A `refresh_token_updater` emits the whole config it was handed as a CONNECTOR_CONFIG control + message, which the platform persists - so a check-only override would become the connection's saved + config. The restore cannot recall a message already on stdout, so the combination is refused.""" + source = ConcurrentDeclarativeSource( + source_config=_manifest_with_refresh_token_updater( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + }, + refresh_token_updater=refresh_token_updater, + ), + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + with pytest.raises(AirbyteTracedException, match="refresh_token_updater") as raised: + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + assert raised.value.failure_type == FailureType.system_error + + +def test_given_no_refresh_token_updater_when_config_overrides_then_manifest_is_accepted(): + """The scan must not reject every OAuth manifest. The same authenticator without the updater writes + nothing back, so the overlay is allowed.""" + manifest = _manifest_with_refresh_token_updater( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + ) + del manifest["streams"][0]["retriever"]["requester"]["authenticator"]["refresh_token_updater"] + + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + assert source._manifest_writes_back_config(source._source_config) is False + + # Asserted through the overlay rather than a full check: this manifest still authenticates with + # OAuth, and mocking a token exchange would test the handshake rather than the guard. + with source._config_overridden_for_check({"resource": "check-only"}): + assert source._config["resource"] == "check-only" + + +def test_given_spec_property_named_refresh_token_updater_then_overrides_are_allowed(): + """The scan walks the raw manifest looking for the key anywhere, because an authenticator reached + through a `$ref` is only found under `definitions`. A connector whose spec happens to declare a + config field by that name must not be caught by that net - nothing in `spec` is a component.""" + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + manifest["spec"] = { + "type": "Spec", + "connection_specification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": { + "resource": {"type": "string"}, + "refresh_token_updater": {"type": "string"}, + }, + }, + } + + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + assert source._manifest_writes_back_config(source._source_config) is False + + with HttpMocker() as http_mocker: + overridden_request = HttpRequest(url="https://api.test.com/check-only") + http_mocker.get(overridden_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + + +def test_given_override_of_a_field_named_refresh_token_updater_then_it_is_allowed(): + """`config_overrides` holds config values, not components, so a key that collides with the + authenticator field name is just a config field and must not trip the guard.""" + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only", "refresh_token_updater": "check-only"}, + } + + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + assert source._manifest_writes_back_config(source._source_config) is False + + with HttpMocker() as http_mocker: + overridden_request = HttpRequest(url="https://api.test.com/check-only") + http_mocker.get(overridden_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + + +def test_given_refresh_token_updater_behind_a_ref_then_manifest_is_rejected(): + """The coarse walk is what makes a `$ref`-ed authenticator detectable at all: the raw manifest holds + only the reference, and the authenticator itself sits under `definitions`. Narrowing the walk must + not lose that.""" + manifest = _manifest_with_refresh_token_updater( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + ) + manifest["definitions"] = { + "authenticator": manifest["streams"][0]["retriever"]["requester"]["authenticator"] + } + manifest["streams"][0]["retriever"]["requester"]["authenticator"] = { + "$ref": "#/definitions/authenticator" + } + + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + with pytest.raises(AirbyteTracedException, match="refresh_token_updater") as raised: + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + assert raised.value.failure_type == FailureType.system_error + + +def test_given_refresh_token_updater_without_config_overrides_then_nothing_is_rejected(): + """The rejection is scoped to the feature. A manifest that does not use `config_overrides` keeps + working with a `refresh_token_updater` exactly as before, even though the manifest scan detects it.""" + source = ConcurrentDeclarativeSource( + source_config=_manifest_with_refresh_token_updater( + {"type": "CheckStream", "stream_names": ["items"]} + ), + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + assert source._manifest_writes_back_config(source.resolved_manifest) is True + + # No overrides means no overlay, so the guard must not fire and the config must not be copied. + with source._config_overridden_for_check(None): + assert source._config is _CONFIG_DRIVEN_PATH_CONFIG + + +def test_given_config_overrides_when_check_then_overridden_keys_are_logged_without_values(caplog): + """An override may name a secret field, so the log records which keys were overridden but never what + they were set to.""" + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "s3cr3t-value"}, + } + ) + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/s3cr3t-value"), + HttpResponse(body=json.dumps([{"id": 1}])), + ) + + with caplog.at_level(logging.INFO): + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + + assert "Overriding config keys for the check operation: resource" in caplog.text + assert "s3cr3t-value" not in caplog.text + + +def test_given_override_key_absent_from_the_spec_then_a_warning_is_logged(caplog): + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only", "typoed_key": 1}, + } + manifest["spec"] = { + "type": "Spec", + "connection_specification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": {"resource": {"type": "string"}}, + }, + } + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/check-only"), + HttpResponse(body=json.dumps([{"id": 1}])), + ) + + with caplog.at_level(logging.WARNING): + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + + assert "typoed_key" in caplog.text + assert "not declared in the connector spec" in caplog.text + assert "'resource'" not in caplog.text + + +def test_given_an_airbyte_reserved_override_key_then_the_manifest_is_rejected(): + """`__airbyte`-prefixed keys are the platform's channel into the config - `CheckStream` reads + `__airbyte_check_stream_names` from the very config the overlay writes to. The feature refuses to + touch that namespace rather than leaving it available.""" + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"__airbyte_check_stream_names": ["something_else"]}, + } + ) + + with pytest.raises(AirbyteTracedException, match="__airbyte_check_stream_names") as raised: + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + assert raised.value.failure_type == FailureType.system_error + + +def _spec_with(properties): + return { + "type": "Spec", + "connection_specification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": properties, + }, + } + + +def test_given_a_reserved_key_when_check_through_the_entrypoint_then_a_trace_is_emitted_and_it_raises(): + """The guards fire on a manifest authoring mistake, so they raise `system_error`, and + `AirbyteEntrypoint.check` treats anything other than `config_error` as exceptional: it emits the + TRACE and then re-raises rather than reporting a FAILED connection status. That is deliberate -- a + broken manifest is a connector bug and should exit non-zero, not be reported to the user as a bad + connection. What must not regress is the TRACE: a bare `ValueError` would escape `run()` entirely + and throw away the message the guard was written to deliver.""" + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"__airbyte_check_stream_names": ["something_else"]}, + } + ) + entrypoint = AirbyteEntrypoint(source) + + messages = [] + with pytest.raises(AirbyteTracedException) as raised: + for message in entrypoint.check( + ConnectorSpecification(connectionSpecification={}), _CONFIG_DRIVEN_PATH_CONFIG + ): + messages.append(message) + + assert raised.value.failure_type == FailureType.system_error + traces = [message.trace for message in messages if message.type == Type.TRACE] + assert len(traces) == 1 + assert traces[0].error.failure_type == FailureType.system_error + assert "__airbyte_check_stream_names" in traces[0].error.message + assert not [message for message in messages if message.type == Type.CONNECTION_STATUS] + + +def test_given_a_config_override_shaped_like_a_reference_then_it_stays_a_literal(): + """`ManifestReferenceResolver` treats any string starting with `#/` as a reference, wherever it + appears. Override values are connector config, so the resolver has to leave them alone - otherwise a + config value that happens to look like a pointer is silently replaced by whatever it resolves to.""" + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["definitions"] = {"somewhere": {"a": 1}} + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "#/definitions/somewhere"}, + } + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + assert source.resolved_manifest["check"]["config_overrides"] == { + "resource": "#/definitions/somewhere" + } + + with HttpMocker() as http_mocker: + overridden_request = HttpRequest(url="https://api.test.com/#/definitions/somewhere") + http_mocker.get(overridden_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + + +def test_given_an_unresolvable_reference_shaped_override_then_the_source_still_constructs(): + """The worst version of the same bug: an unresolvable pointer raises inside `_pre_process_manifest`, + which runs in `__init__` - so `spec`, `discover` and `read` would all die over a field only `check` + ever reads.""" + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "#/nothing/here"}, + } + + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + assert [stream.name for stream in source.streams(_CONFIG_DRIVEN_PATH_CONFIG)] == ["items"] + + +@pytest.mark.parametrize( + "mutate_manifest, description", + [ + pytest.param( + lambda manifest: manifest["streams"][0]["schema_loader"]["schema"]["properties"].update( + {"refresh_token_updater": {"type": "string"}} + ), + "record schema property", + id="inline-schema-property", + ), + pytest.param( + lambda manifest: manifest["streams"][0]["retriever"]["requester"].update( + {"request_parameters": {"refresh_token_updater": "x"}} + ), + "request parameter name", + id="request-parameter-name", + ), + pytest.param( + lambda manifest: manifest.update( + { + "schemas": { + "items": {"properties": {"refresh_token_updater": {"type": "string"}}} + } + } + ), + "top-level schemas block", + id="top-level-schemas", + ), + pytest.param( + lambda manifest: manifest["check"]["config_overrides"].update( + {"credentials": {"type": "settings", "refresh_token_updater": "a-value"}} + ), + "object-valued override carrying both keys", + id="override-value-with-type-and-key", + ), + ], +) +def test_given_the_name_appears_in_a_data_blob_then_overrides_are_still_allowed( + mutate_manifest, description +): + """The scan looks for `refresh_token_updater` anywhere, because an authenticator can sit under any + requester. What keeps that from catching data is the `type` on the mapping holding the key: a + component always has one, a record schema property or a request body entry does not. Without that + condition these manifests - which declare no authenticator at all - are refused with an error + telling the author to remove something that does not exist.""" + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + mutate_manifest(manifest) + + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + assert source._manifest_writes_back_config(source._source_config) is False + + with HttpMocker() as http_mocker: + overridden_request = HttpRequest( + url="https://api.test.com/check-only", + query_params=manifest["streams"][0]["retriever"]["requester"].get("request_parameters"), + ) + http_mocker.get(overridden_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + + +def test_given_a_nested_override_then_the_object_is_replaced_rather_than_merged(): + """The one-level merge is a documented semantic, not an accident: replacing the object wholesale is + what keeps "remove this nested key during check" expressible. A recursive merge would leave every + sibling key in place, so this asserts a sibling is gone.""" + config = {"resource": "sync", "settings": {"mode": "sync", "sibling": "present"}} + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"settings": {"mode": "check-only"}}, + } + manifest["streams"][0]["retriever"]["requester"]["url"] = ( + "https://api.test.com/{{ config['settings'].get('sibling', 'dropped') }}" + ) + + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=config, + catalog=None, + state=None, + ) + + with HttpMocker() as http_mocker: + replaced_request = HttpRequest(url="https://api.test.com/dropped") + http_mocker.get(replaced_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, config).status == Status.SUCCEEDED + http_mocker.assert_number_of_calls(replaced_request, 1) + + +def test_given_non_string_override_keys_then_the_manifest_is_rejected_cleanly(): + """`type: object` in the schema does not constrain key types, and YAML will happily produce an + integer key. Every consumer downstream treats a key as a string, so this is refused with an + author-facing message rather than a `TypeError` from a join.""" + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {0: "check-only"}, + } + ) + + with pytest.raises(AirbyteTracedException, match="are not strings") as raised: + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + assert raised.value.failure_type == FailureType.system_error + + +def test_given_a_spec_composed_with_all_of_then_no_spurious_warning_is_logged(caplog): + """A spec does not have to enumerate its fields at the top level. Reading only `properties` reports + an `allOf`-composed field as undeclared, which sends an author chasing a warning about a key that is + perfectly valid.""" + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + manifest["spec"] = { + "type": "Spec", + "connection_specification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": {}, + "allOf": [{"properties": {"resource": {"type": "string"}}}], + }, + } + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/check-only"), + HttpResponse(body=json.dumps([{"id": 1}])), + ) + + with caplog.at_level(logging.WARNING): + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + + assert "not declared in the connector spec" not in caplog.text + + +def test_given_an_override_on_a_secret_field_then_the_value_is_registered_for_redaction(): + """The entrypoint builds the secret list from the config the user supplied, so a value substituted + here is unknown to `filter_secrets` - and would print in the clear at a path where the user's own + value prints as `****`.""" + from airbyte_cdk.utils.airbyte_secrets_utils import filter_secrets, update_secrets + + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only-secret"}, + } + manifest["spec"] = _spec_with({"resource": {"type": "string", "airbyte_secret": True}}) + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + update_secrets([]) + try: + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/check-only-secret"), + HttpResponse(body=json.dumps([{"id": 1}])), + ) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + + assert filter_secrets("saw check-only-secret") == "saw ****" + finally: + update_secrets([]) + + +def test_config_overrides_is_published_on_both_check_components(): + """The schema is the contract the Connector Builder and the manifest server read, and the factory + deliberately ignores the field - so nothing at runtime notices if it disappears from the published + surface.""" + schema = yaml.safe_load( + pkgutil.get_data( + "airbyte_cdk.sources.declarative", "declarative_component_schema.yaml" + ).decode() + ) + + for component in ("CheckStream", "CheckDynamicStream"): + assert "config_overrides" in schema["definitions"][component]["properties"] + + assert CheckStreamModel( + type="CheckStream", stream_names=["items"], config_overrides={"a": 1} + ).config_overrides == {"a": 1} + assert CheckDynamicStreamModel( + type="CheckDynamicStream", stream_count=1, config_overrides={"a": 1} + ).config_overrides == {"a": 1} + + +def test_given_a_custom_authenticator_declaring_a_refresh_token_updater_then_it_is_rejected(): + """`refresh_token_updater` is declared on `OAuthAuthenticator` alone, so matching that type would be + enough for the schema as written. `CustomAuthenticator` is matched too because the transformer + injects that type for a `class_name` component, and custom code that declares the field is the shape + most likely to write the config back - the one case of custom code the manifest does name.""" + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + manifest["streams"][0]["retriever"]["requester"]["authenticator"] = { + "type": "CustomAuthenticator", + "class_name": "unit_tests.sources.declarative.test_concurrent_declarative_source_config_overrides.NotBuilt", + "refresh_token_updater": {}, + } + + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + # Matched on the guard's own sentence: a failure to import the custom class also mentions + # `refresh_token_updater`, because the message echoes the component definition. + with pytest.raises( + AirbyteTracedException, match="cannot be used by a manifest that declares" + ) as raised: + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + assert raised.value.failure_type == FailureType.system_error