diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index d7e5e4260..60c51e4cb 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 @@ -216,6 +217,8 @@ def __init__( AlwaysLogSliceLogger() if emit_connector_builder_messages else DebugSliceLogger() ) + # Populated by `_pre_process_manifest` from the check component, before parameter propagation. + self._check_config_overrides: Optional[Mapping[str, Any]] = None # resolve all components in the manifest self._source_config = self._pre_process_manifest(dict(source_config)) # validate resolved manifest against the declarative component schema @@ -228,6 +231,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: @@ -275,6 +282,10 @@ def _pre_process_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]: manifest = self._fix_source_type(manifest) # Resolve references in the manifest resolved_manifest = ManifestReferenceResolver().preprocess_manifest(manifest) + # The check component's `config_overrides` holds connector config values rather than + # components, so it is read here: references are resolved, so a `check` behind a `$ref` is + # found, but parameters have not been propagated yet, so the values are still pristine. + self._check_config_overrides = self._extract_check_config_overrides(resolved_manifest) # Propagate types and parameters throughout the manifest propagated_manifest = ManifestComponentTransformer().propagate_types_and_parameters( "", resolved_manifest, {} @@ -282,6 +293,27 @@ def _pre_process_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]: return propagated_manifest + @staticmethod + def _extract_check_config_overrides( + resolved_manifest: Mapping[str, Any], + ) -> Optional[Mapping[str, Any]]: + """Take the check component's `config_overrides` before parameter propagation mangles it. + + `ManifestComponentTransformer` injects `$parameters` and the parameters themselves into every + nested mapping that carries a truthy `type` key, because everywhere else in a manifest such a + mapping is a component. `config_overrides` values come from the connector's own spec, where + `type` is an ordinary field name - so an override like `credentials: {type: oauth, ...}` would + otherwise reach the check with stray keys in it. + """ + check = resolved_manifest.get("check") + if not isinstance(check, Mapping): + return None + config_overrides = check.get("config_overrides") + if not isinstance(config_overrides, Mapping): + return None + # Deep copy so the propagation that follows, which mutates in place, cannot reach these values. + return deepcopy(dict(config_overrides)) + def _fix_source_type(self, manifest: Dict[str, Any]) -> Dict[str, Any]: """ Fix the source type in the manifest. This is necessary because the source type is not always set in the manifest. @@ -413,7 +445,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 +639,167 @@ 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(self._check_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. + + A check and a sync legitimately want different behaviour from the same manifest: a check is + interactive and should fail fast with an actionable message, while a sync can afford to wait out + a rate limit window. Before this existed, expressing that difference required a Python + `check_connection` override that built a second component tree from a modified config, which is + not available to a manifest-only connector. + + Overlaying here is enough to reach every component the checker builds, because `streams()` + interpolates from `self._config` and ignores its own `config` argument. + + Semantics, all deliberate: + - Values are applied verbatim. They are not interpolated, so a value containing `{{ }}` reaches + components as that literal string. + - The merge is one level deep. Overriding a key whose value is an object replaces that object + rather than merging into it, which keeps removing a nested key expressible. + - The overlay happens long after `_migrate_and_transform_config`, so an override is neither + normalised itself nor propagated to fields derived from it by a `ConfigTransformation`. + - `config_validations` run against `self._config_for_validation`, so an override cannot fail a + validation written for the config as supplied. + + Two consequences of the copy being shallow and of `self._config` being shared state: + - Rebinding `self._config` is not thread safe. Acceptable only because `check` is one command + per process. + - The copy is shallow, so every nested object is the *same* object as in the config the source + was constructed with. A component that writes into a nested path - `dpath.new(config, + ("credentials", "access_token"), ...)` - therefore writes through to that shared dict, and the + restore does not undo it, because the restore only rebinds the top level. Only a write to a + top-level key is discarded. + + The write-through above is why a manifest that declares a `refresh_token_updater` is rejected + outright rather than documented: see `_raise_if_config_is_persisted`. + """ + if not config_overrides: + yield + return + + self._raise_on_reserved_override_keys(config_overrides) + self._raise_if_config_is_persisted(config_overrides) + self._warn_on_unknown_override_keys(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 to overlay keys in the platform's reserved `__airbyte` namespace. + + These are not connector config, they are the platform's channel into it: `CheckStream` reads + `__airbyte_check_stream_names` out of the very config this overlay writes to, so an override there + would change which streams a check tests. A manifest wanting that has `stream_names` for it. The + prefix is refused wholesale so future internal keys are covered too. + """ + reserved = sorted(key for key in config_overrides if key.startswith("__airbyte")) + if reserved: + raise ValueError( + f"`config_overrides` may not set the reserved key(s) {reserved}. Keys prefixed with " + "`__airbyte` belong to the platform, not to the connector's spec." + ) + + 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` turns an `OAuthAuthenticator` into a + `DeclarativeSingleUseRefreshTokenOauth2Authenticator`, which is handed the config the component + tree was built with - during a check, the overlay. When it refreshes, `_emit_control_message` + prints that *entire* dict as a `CONNECTOR_CONFIG` control message, and the platform persists it. + So a check-only override would become the connection's saved config and apply to every later sync. + + Restoring `self._config` afterwards cannot help: the control message is already on stdout. And no + amount of threading the config through properly would help either, because the hazard is inherent + to handing an overridden config to something whose job is to write the config back. Until the + emitter is fixed to emit the config it was given plus only the token fields it owns, refusing the + combination is the honest answer. + """ + if not self._manifest_writes_back_config(self._source_config): + return + raise ValueError( + "`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 the platform persists, so the check-only " + f"override(s) {sorted(config_overrides)} would be saved as this connection's config and " + "applied to every later sync. Remove `config_overrides`, or drop the `refresh_token_updater`." + ) + + # Subtrees of a manifest that hold data rather than components, so a `refresh_token_updater` key + # found inside one is a name collision and not an authenticator. Neither key is declared on any + # component that could contain an authenticator: `spec` exists only at the top level of a manifest, + # and `config_overrides` only on `CheckStream` and `CheckDynamicStream`. + _NON_COMPONENT_MANIFEST_KEYS = frozenset({"spec", "config_overrides"}) + + @staticmethod + def _manifest_writes_back_config(definition: Any) -> bool: + """Whether any component in the manifest emits the connector config back to the platform. + + The scan is deliberately coarse - it looks for the key anywhere rather than only under a + recognised authenticator - because it runs on the raw manifest, before references are resolved. + An authenticator reached through a `$ref` is only found because the walk also visits + `definitions`, where it lives under its own key rather than under a requester. The two subtrees + in `_NON_COMPONENT_MANIFEST_KEYS` are the exception: they hold config values, so a matching key + there means a connector whose spec happens to declare a field by that name, not a token refresh. + + Tested with `is not None` rather than for truthiness, to match the factory. Every field of + `RefreshTokenUpdater` has a default, so `refresh_token_updater: {}` is a valid way to take all of + them - and it builds a single-use authenticator just like a populated one, because the factory's + `if model.refresh_token_updater:` sees a model instance, which is always truthy. A truthiness test + here would see an empty dict and let that manifest through. + """ + if isinstance(definition, Mapping): + if definition.get("refresh_token_updater") is not None: + return True + return any( + ConcurrentDeclarativeSource._manifest_writes_back_config(value) + for key, value in definition.items() + if key not in ConcurrentDeclarativeSource._NON_COMPONENT_MANIFEST_KEYS + ) + 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. + + The overlay is the one part of the config nothing validates - the entrypoint validates what the + user supplied, and `config_validations` deliberately run against `self._config_for_validation`. So + a typo or a since-renamed field is a silent no-op that surfaces much later as "why does check no + longer fail fast". + """ + if not self._spec_component: + return + declared = self._spec_component.connection_specification.get("properties") + if not isinstance(declared, Mapping): + 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." + ) + @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 091c694d2..1de34175d 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -549,6 +549,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, 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. 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: @@ -587,6 +595,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, 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. 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 37eb50b7b..25b50bdd9 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, 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. 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): @@ -1781,6 +1787,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, 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. 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/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index a96c406d8..1aa729d18 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -1306,6 +1306,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, @@ -1320,6 +1323,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/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py index 8bc9571d1..5085a024c 100644 --- a/unit_tests/sources/declarative/checks/test_check_stream.py +++ b/unit_tests/sources/declarative/checks/test_check_stream.py @@ -907,3 +907,587 @@ def test_check_stream_only_type_provided(): ) with pytest.raises(ValueError): source.check(logger, _CONFIG) + + +_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) + + +def test_given_object_valued_config_override_when_check_then_parameters_are_not_injected_into_it(): + """`config_overrides` values come from the connector's spec, where `type` is an ordinary field name. + + `ManifestComponentTransformer` injects the enclosing component's parameters, plus a `$parameters` + key, into every nested mapping carrying a truthy `type` key - everywhere else in a manifest such a + mapping is a component. So a check component that declares `$parameters` used to hand the checker a + `credentials` object with stray keys in it. The request echoes the override's keys, so the mocked + request matches only if the object reached the stream exactly as authored. + """ + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["streams"][0]["retriever"]["requester"]["request_parameters"] = { + "credential_keys": "{{ config['credentials'].keys() | list | join(',') }}" + } + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + # A no-op on a check component - `create_check_stream` passes `parameters={}` - but it is what + # the propagation walks with. + "$parameters": {"injected": "parameter"}, + "config_overrides": { + "resource": "check-only", + "credentials": {"type": "oauth", "client_id": "client-id"}, + }, + } + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + with HttpMocker() as http_mocker: + pristine_request = HttpRequest( + url="https://api.test.com/check-only", + query_params={"credential_keys": "type,client_id"}, + ) + http_mocker.get(pristine_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + http_mocker.assert_number_of_calls(pristine_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(ValueError, match="refresh_token_updater"): + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + + +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 + + +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 + + +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(ValueError, match="refresh_token_updater"): + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + + +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): + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + + 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): + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + + 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(ValueError, match="__airbyte_check_stream_names"): + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG)