From f8c916a5313373d8336decf4a138f5c4f494ae86 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:04:19 +0000 Subject: [PATCH 1/6] feat(mcp): deploy sources from custom YAML definitions Co-Authored-By: AJ Steers --- airbyte/cloud/connectors.py | 44 ++++- airbyte/cloud/workspaces.py | 33 +++- airbyte/mcp/cloud.py | 131 ++++++++++---- .../cloud/test_custom_definitions.py | 35 ++++ tests/unit_tests/fixtures/arg_trace_plan.json | 1 + tests/unit_tests/test_cloud_credentials.py | 54 ++++++ .../test_deferred_cloud_credentials.py | 75 +++++++- tests/unit_tests/test_mcp_cloud.py | 162 ++++++++++++++++++ 8 files changed, 488 insertions(+), 47 deletions(-) diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py index b66fdcfa1..73fbc3727 100644 --- a/airbyte/cloud/connectors.py +++ b/airbyte/cloud/connectors.py @@ -59,6 +59,7 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, ClassVar, Literal, Protocol +import jsonschema import requests import yaml from pydantic import ValidationError @@ -1255,7 +1256,7 @@ def _from_destination_response( return result -class CustomCloudSourceDefinition: +class CustomCloudSourceDefinition: # noqa: PLR0904 """A custom source connector definition in Airbyte Cloud. This represents either a YAML (declarative) or Docker-based custom source definition. @@ -1320,6 +1321,45 @@ def manifest(self) -> dict[str, Any] | None: self._definition_info = self._fetch_definition_info() return self._definition_info.manifest + @property + def connection_specification(self) -> dict[str, Any] | None: + """Get the JSON schema for this custom source's configuration.""" + manifest = self.manifest + if manifest is None: + return None + manifest_spec = manifest.get("spec") + if not isinstance(manifest_spec, dict): + return None + connection_specification = manifest_spec.get("connection_specification") + return connection_specification if isinstance(connection_specification, dict) else None + + def validate_config(self, config: dict[str, Any]) -> None: + """Validate the config against the custom source's connection specification.""" + specification = self.connection_specification + if specification is None: + return + + try: + jsonschema.validate(config, specification) + except jsonschema.ValidationError as ex: + json_path = "$" + for part in ex.absolute_path: + if isinstance(part, str): + json_path += f".{part}" + elif isinstance(part, int): + json_path += f"[{part}]" + raise exc.AirbyteConnectorValidationFailedError( + connector_name=self.name, + message=( + f"The provided config is not valid at `{json_path}` " + f"(failed the `{ex.validator}` rule)." + ), + context={ + "error_path": ex.path, + "error_schema": ex.schema, + }, + ) from ex + @property def version(self) -> str | None: """Get the manifest version. Only present for YAML connectors.""" @@ -1795,7 +1835,7 @@ def set_testing_values( if manifest_spec: spec = { "documentationUrl": manifest_spec.get("documentation_url"), - "connectionSpecification": manifest_spec.get("connection_specification", {}), + "connectionSpecification": self.connection_specification or {}, "advancedAuth": manifest_spec.get("advanced_auth"), } diff --git a/airbyte/cloud/workspaces.py b/airbyte/cloud/workspaces.py index 202e9ec12..7c12f1d6f 100644 --- a/airbyte/cloud/workspaces.py +++ b/airbyte/cloud/workspaces.py @@ -656,12 +656,14 @@ def deploy_source( Args: name: The name to use when deploying. - source: The source object to deploy, or (with `defer_credentials=True`) a + source: The source object to deploy, or a dictionary of configuration values + when `definition_id` is provided. With `defer_credentials=True`, pass a dictionary of non-secret configuration values. unique: Whether to require a unique name. If `True`, duplicate names are not allowed. Defaults to `True`. random_name_suffix: Whether to append a random suffix to the name. - definition_id: The source definition ID. Required with `defer_credentials=True`. + definition_id: The source definition ID. Required for a dictionary config (for + example, a custom definition ID) and with `defer_credentials=True`. defer_credentials: Save a draft with partial configuration. A person completes credentials and other missing settings at the returned source's `connector_url`. A successful connection check promotes the draft. Raises @@ -680,12 +682,17 @@ def deploy_source( ), ) if isinstance(source, dict): - raise exc.AirbyteLibInputError( - message="`source` must be a `Source` object unless `defer_credentials=True`.", - ) - - source_config_dict = source._hydrated_config.copy() # noqa: SLF001 (non-public API) - source_config_dict["sourceType"] = source.name.replace("source-", "") + if definition_id is None: + raise exc.AirbyteLibInputError( + message=( + "`source` must be a `Source` object unless `definition_id` or " + "`defer_credentials=True` is given." + ), + ) + source_config_dict = dict(source) + else: + source_config_dict = source._hydrated_config.copy() # noqa: SLF001 (non-public API) + source_config_dict["sourceType"] = source.name.replace("source-", "") if random_name_suffix: name += f" (ID: {text_util.generate_random_suffix()})" @@ -799,10 +806,18 @@ def _deploy_deferred( random_name_suffix: bool, ) -> str: """Create a connector with deferred credentials on the Config API and return its ID.""" + spec_json_schema = _get_deferred_spec(definition_id) if definition_id else None + if spec_json_schema is None and connector_type == "source" and definition_id: + with contextlib.suppress(exc.AirbyteLibError): + spec_json_schema = self.get_custom_source_definition( + definition_id, + definition_type="yaml", + ).connection_specification + config_dict, definition_id = _deferred_credentials_config( config, definition_id=definition_id, - spec_json_schema=_get_deferred_spec(definition_id) if definition_id else None, + spec_json_schema=spec_json_schema, ) if random_name_suffix: diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index 5684d767b..56275d79c 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -606,25 +606,40 @@ def deploy_connector_to_cloud( # noqa: PLR0913 # Mirrors the API surface. TraceArg.FINGERPRINT, ], connector_name: Annotated[ - str, + str | None, Field( description=( "The canonical name of the connector (e.g., 'source-faker' or " "'destination-postgres')." ), + default=None, ), TraceArg.FINGERPRINT, - ], + ] = None, *, connector_type: Annotated[ ConnectorType | None, Field( description=( "Optional: `source` or `destination`. When omitted, inferred from the " - "`source-`/`destination-` prefix of `connector_name`." + "`source-`/`destination-` prefix of `connector_name`. Custom definitions " + "are sources." + ), + default=None, + ), + ] = None, + custom_definition_id: Annotated[ + str | None, + Field( + description=( + "ID of a custom (YAML / Connector Builder) source definition in this " + "workspace, e.g. from `publish_custom_source_definition` or " + "`list_custom_source_definitions`. Pass this instead of `connector_name` " + "to deploy a source from a custom definition." ), default=None, ), + TraceArg.HASH, ] = None, workspace_id: Annotated[ str | None, @@ -667,40 +682,81 @@ def deploy_connector_to_cloud( # noqa: PLR0913 # Mirrors the API surface. ) -> str: """Deploy a source or destination connector to Airbyte Cloud. + Use `custom_definition_id` to deploy a custom YAML source definition. + With `defer_credentials=True`, returns a JSON `DeferredDeployResult` whose `settings_url` is where a person completes the credentials. """ - resolved_type = connector_type or _infer_connector_type_from_name(connector_name) + if (connector_name is None) == (custom_definition_id is None): + raise AirbyteLibInputError( + message="Provide exactly one of `connector_name` or `custom_definition_id`.", + guidance=( + "Use `connector_name` for a registry connector or " + "`custom_definition_id` for a custom source." + ), + ) + if custom_definition_id is not None and connector_type == ConnectorType.DESTINATION: + raise AirbyteLibInputError( + message="Custom source definitions can only be deployed as sources.", + guidance="Pass `connector_type='source'` or omit `connector_type`.", + ) + + if custom_definition_id is not None: + resolved_type = ConnectorType.SOURCE + else: + assert connector_name is not None + resolved_type = connector_type or _infer_connector_type_from_name(connector_name) if defer_credentials: return _deploy_deferred_to_cloud( ctx, connector_type=resolved_type, name=name, connector_name=connector_name, + custom_definition_id=custom_definition_id, workspace_id=workspace_id, config=config, config_secret_name=config_secret_name, unique=unique, ).model_dump_json(indent=2) - connector: Source | Destination = ( - get_source(connector_name, no_executor=True) - if resolved_type == ConnectorType.SOURCE - else get_destination(connector_name, no_executor=True) - ) - config_dict = resolve_connector_config( - config=config, - config_secret_name=config_secret_name, - config_spec_jsonschema=connector.config_spec, - ) - connector.set_config(config_dict, validate=True) + if custom_definition_id is not None: + workspace = _get_cloud_workspace(ctx, workspace_id) + definition = workspace.get_custom_source_definition( + custom_definition_id, + definition_type="yaml", + ) + config_dict = resolve_connector_config( + config=config, + config_secret_name=config_secret_name, + config_spec_jsonschema=definition.connection_specification, + ) + definition.validate_config(config_dict) + deployed: CloudConnector = workspace.deploy_source( + name=name, + source=config_dict, + unique=unique, + definition_id=custom_definition_id, + ) + else: + assert connector_name is not None + connector: Source | Destination = ( + get_source(connector_name, no_executor=True) + if resolved_type == ConnectorType.SOURCE + else get_destination(connector_name, no_executor=True) + ) + config_dict = resolve_connector_config( + config=config, + config_secret_name=config_secret_name, + config_spec_jsonschema=connector.config_spec, + ) + connector.set_config(config_dict, validate=True) - workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id) - deployed: CloudConnector = ( - workspace.deploy_source(name=name, source=connector, unique=unique) - if isinstance(connector, Source) - else workspace.deploy_destination(name=name, destination=connector, unique=unique) - ) + workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id) + deployed = ( + workspace.deploy_source(name=name, source=connector, unique=unique) + if isinstance(connector, Source) + else workspace.deploy_destination(name=name, destination=connector, unique=unique) + ) register_guid_created_in_session(deployed.connector_id) return ( @@ -709,12 +765,13 @@ def deploy_connector_to_cloud( # noqa: PLR0913 # Mirrors the API surface. ) -def _deploy_deferred_to_cloud( +def _deploy_deferred_to_cloud( # noqa: PLR0913 ctx: Context, *, connector_type: ConnectorType, name: str, - connector_name: str, + connector_name: str | None, + custom_definition_id: str | None, workspace_id: str | None, config: dict | str | None, config_secret_name: str | None, @@ -726,14 +783,22 @@ def _deploy_deferred_to_cloud( message="`config_secret_name` cannot be used with `defer_credentials=True`.", guidance="Pass non-secret configuration in `config`; credentials are entered in Cloud.", ) - metadata = get_connector_metadata(connector_name) - if metadata is None or metadata.definition_id is None: - raise AirbyteConnectorNotRegisteredError(connector_name=connector_name) - if metadata.connector_type != connector_type: - raise AirbyteLibInputError( - message=f"`{connector_name}` is not a {connector_type} connector.", - guidance=f"Pass a `{connector_type}-*` connector name.", - ) + if custom_definition_id is not None: + definition_id = custom_definition_id + else: + if connector_name is None: + raise AirbyteLibInputError( + message="Provide a connector name when deploying a registry connector." + ) + metadata = get_connector_metadata(connector_name) + if metadata is None or metadata.definition_id is None: + raise AirbyteConnectorNotRegisteredError(connector_name=connector_name) + if metadata.connector_type != connector_type: + raise AirbyteLibInputError( + message=f"`{connector_name}` is not a {connector_type} connector.", + guidance=f"Pass a `{connector_type}-*` connector name.", + ) + definition_id = metadata.definition_id config_dict = resolve_connector_config(config=config) workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id) @@ -744,7 +809,7 @@ def _deploy_deferred_to_cloud( name=name, source=config_dict, unique=unique, - definition_id=metadata.definition_id, + definition_id=definition_id, defer_credentials=True, ) else: @@ -752,7 +817,7 @@ def _deploy_deferred_to_cloud( name=name, destination=config_dict, unique=unique, - definition_id=metadata.definition_id, + definition_id=definition_id, defer_credentials=True, ) except AirbyteDeferredSetupError as ex: diff --git a/tests/integration_tests/cloud/test_custom_definitions.py b/tests/integration_tests/cloud/test_custom_definitions.py index a852821a7..3bd65b853 100644 --- a/tests/integration_tests/cloud/test_custom_definitions.py +++ b/tests/integration_tests/cloud/test_custom_definitions.py @@ -100,6 +100,41 @@ def test_publish_custom_yaml_source( fetched.permanently_delete(safe_mode=False) +@pytest.mark.requires_creds +def test_deploy_source_from_custom_yaml_definition( + cloud_workspace: CloudWorkspace, +) -> None: + """Deploy a custom YAML source definition by passing its ID and a config dictionary.""" + from airbyte._util import text_util + + name = f"delete-me-custom-yaml-source-{text_util.generate_random_suffix()}" + definition = cloud_workspace.publish_custom_source_definition( + name=name, + manifest_yaml=TEST_YAML_MANIFEST, + unique=True, + pre_validate=True, + ) + deployed_source = None + + try: + try: + deployed_source = cloud_workspace.deploy_source( + name=name, + source={}, + definition_id=definition.definition_id, + ) + + assert deployed_source.definition_id == definition.definition_id + finally: + if deployed_source is not None: + cloud_workspace.permanently_delete_source( + deployed_source, + safe_mode=False, + ) + finally: + definition.permanently_delete(safe_mode=False) + + @pytest.mark.requires_creds def test_yaml_validation_error( cloud_workspace: CloudWorkspace, diff --git a/tests/unit_tests/fixtures/arg_trace_plan.json b/tests/unit_tests/fixtures/arg_trace_plan.json index 8f3059af5..23b989e52 100644 --- a/tests/unit_tests/fixtures/arg_trace_plan.json +++ b/tests/unit_tests/fixtures/arg_trace_plan.json @@ -22,6 +22,7 @@ "config_secret_name": "fingerprint", "connector_name": "fingerprint", "connector_type": "value allowed=[\"destination\",\"source\"]", + "custom_definition_id": "hash", "defer_credentials": "value allowed=[false,true]", "name": "fingerprint", "unique": "value allowed=[false,true]", diff --git a/tests/unit_tests/test_cloud_credentials.py b/tests/unit_tests/test_cloud_credentials.py index 28b4d7bc4..ccd4eaa37 100644 --- a/tests/unit_tests/test_cloud_credentials.py +++ b/tests/unit_tests/test_cloud_credentials.py @@ -28,6 +28,7 @@ ) from airbyte.cloud.models import ( CloudDestinationInfo, + CloudCustomSourceDefinitionInfo, OrganizationFeature, CloudSourceInfo, CloudWorkspaceInfo, @@ -40,6 +41,7 @@ from airbyte.exceptions import ( AirbyteCloudApiError, AirbyteCloudError, + AirbyteConnectorValidationFailedError, AirbyteMissingResourceError, AirbyteLibInputError, ) @@ -801,6 +803,58 @@ def test_custom_cloud_source_definition_docker_rename_is_not_implemented() -> No definition.rename(name="Renamed definition") +def test_custom_cloud_source_definition_validates_config() -> None: + definition = CustomCloudSourceDefinition( + workspace=CloudWorkspace(workspace_id="workspace-id", bearer_token="token"), + definition_id="definition-id", + definition_type="yaml", + ) + definition._definition_info = CloudCustomSourceDefinitionInfo( + definition_id="definition-id", + name="Custom source", + version="0.1.0", + manifest={ + "spec": { + "connection_specification": { + "type": "object", + "required": ["greeting"], + "properties": {"greeting": {"type": "string"}}, + } + } + }, + ) + + assert definition.connection_specification == { + "type": "object", + "required": ["greeting"], + "properties": {"greeting": {"type": "string"}}, + } + definition.validate_config({"greeting": "hi"}) + + with pytest.raises( + AirbyteConnectorValidationFailedError, + match=r"The provided config is not valid at `\$` \(failed the `required` rule\)\.", + ): + definition.validate_config({}) + + +def test_custom_cloud_source_definition_without_spec_skips_config_validation() -> None: + definition = CustomCloudSourceDefinition( + workspace=CloudWorkspace(workspace_id="workspace-id", bearer_token="token"), + definition_id="definition-id", + definition_type="yaml", + ) + definition._definition_info = CloudCustomSourceDefinitionInfo( + definition_id="definition-id", + name="Custom source", + version="0.1.0", + manifest=None, + ) + + assert definition.connection_specification is None + definition.validate_config({"anything": "goes"}) + + def test_cloud_workspace_rename_forwards_inputs( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/unit_tests/test_deferred_cloud_credentials.py b/tests/unit_tests/test_deferred_cloud_credentials.py index 0dc492307..44e2935a3 100644 --- a/tests/unit_tests/test_deferred_cloud_credentials.py +++ b/tests/unit_tests/test_deferred_cloud_credentials.py @@ -5,7 +5,9 @@ import json from dataclasses import dataclass, field +from types import SimpleNamespace from typing import Any, Literal, cast +from unittest.mock import MagicMock import pytest import requests @@ -45,6 +47,11 @@ def _stub_secret_detection( lambda: ["password", "api_key", "token", "secret", "email"], ) monkeypatch.setattr(cloud_workspaces, "_get_deferred_spec", lambda _id: None) + monkeypatch.setattr( + CloudWorkspace, + "get_custom_source_definition", + lambda *_args, **_kwargs: SimpleNamespace(connection_specification=None), + ) def _actor_body(connector_type: str, *, draft: bool | None) -> dict[str, Any]: @@ -359,15 +366,46 @@ def test_deploy_deferred_rejects_invalid_input_before_any_request( def test_deploy_source_rejects_dict_config_without_defer_credentials( monkeypatch: pytest.MonkeyPatch, ) -> None: - """Raw source dictionaries are only accepted in deferred mode.""" + """Raw source dictionaries require a definition ID or deferred credentials.""" call = _stub_create(monkeypatch, "source") - with pytest.raises(exc.AirbyteLibInputError): - _deploy(_workspace(), "source", {"count": 10}, definition_id=DEFINITION_ID) + with pytest.raises( + exc.AirbyteLibInputError, + match=r"`source` must be a `Source` object unless `definition_id` or " + r"`defer_credentials=True` is given\.", + ): + _deploy(_workspace(), "source", {"count": 10}) assert call.kwargs == {} +def test_deploy_source_passes_custom_definition_config_without_source_type( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A custom source config is passed to the public API without a sourceType field.""" + create_kwargs: dict[str, Any] = {} + monkeypatch.setattr( + api_util, + "create_source", + lambda **kwargs: ( + create_kwargs.update(kwargs) or SimpleNamespace(source_id=ACTOR_ID) + ), + ) + monkeypatch.setattr(api_util, "list_sources", lambda **_: []) + workspace = _workspace() + + deployed = workspace.deploy_source( + name="My custom source", + source={"greeting": "hi"}, + definition_id=DEFINITION_ID, + ) + + assert deployed.connector_id == ACTOR_ID + assert create_kwargs["config"] == {"greeting": "hi"} + assert "sourceType" not in create_kwargs["config"] + assert create_kwargs["definition_id"] == DEFINITION_ID + + @pytest.mark.parametrize("connector_type", CONNECTOR_TYPES) def test_deploy_deferred_rejects_plaintext_credentials( monkeypatch: pytest.MonkeyPatch, @@ -466,6 +504,37 @@ def test_deploy_deferred_spec_still_rejects_secret_fields( assert raised.value.context == {"fields": ["password"]} +def test_deploy_deferred_uses_custom_source_spec_for_secret_detection( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A custom source's secret schema supplements the registry and global secret masks.""" + call = _stub_create(monkeypatch, "source") + workspace = _workspace() + get_definition = MagicMock( + return_value=SimpleNamespace( + connection_specification={ + "type": "object", + "properties": { + "my_token_field": {"type": "string", "airbyte_secret": True}, + }, + }, + ) + ) + monkeypatch.setattr(workspace, "get_custom_source_definition", get_definition) + + with pytest.raises(exc.AirbyteLibInputError, match="credential values") as raised: + workspace.deploy_source( + name="My custom source", + source={"my_token_field": "hunter2"}, + definition_id=DEFINITION_ID, + defer_credentials=True, + ) + + get_definition.assert_called_once_with(DEFINITION_ID, definition_type="yaml") + assert call.kwargs == {} + assert raised.value.context == {"fields": ["my_token_field"]} + + def test_get_connector_name_by_definition_id( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py index c4496adf5..578613d49 100644 --- a/tests/unit_tests/test_mcp_cloud.py +++ b/tests/unit_tests/test_mcp_cloud.py @@ -1195,6 +1195,168 @@ def test_deploy_connector_to_cloud_rejects_unknown_prefix() -> None: ) +def test_deploy_connector_to_cloud_supports_custom_definition( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Custom source definitions are validated and deployed with the supplied ID.""" + definition = MagicMock() + definition.connection_specification = {"type": "object"} + definition.validate_config = MagicMock() + deploy_calls: list[dict[str, object]] = [] + + def fake_deploy(**kwargs: object) -> SimpleNamespace: + deploy_calls.append(kwargs) + return SimpleNamespace( + connector_id="deployed-id", + connector_url="https://cloud.airbyte.com/deployed-id", + ) + + workspace = SimpleNamespace( + get_custom_source_definition=MagicMock(return_value=definition), + deploy_source=fake_deploy, + ) + monkeypatch.setattr(cloud_mcp, "_get_cloud_workspace", lambda *_: workspace) + config = {"greeting": "hi"} + resolved_config_args: dict[str, object] = {} + + def fake_resolve_config(**kwargs: object) -> dict[str, Any]: + resolved_config_args.update(kwargs) + return config + + monkeypatch.setattr(cloud_mcp, "resolve_connector_config", fake_resolve_config) + registered: list[str] = [] + monkeypatch.setattr( + cloud_mcp, "register_guid_created_in_session", registered.append + ) + + result = cloud_mcp.deploy_connector_to_cloud( + cast(Context, object()), + name="My custom source", + connector_name=None, + custom_definition_id="custom-definition-id", + connector_type=None, + workspace_id=None, + config=config, + config_secret_name=None, + unique=True, + ) + + workspace.get_custom_source_definition.assert_called_once_with( + "custom-definition-id", + definition_type="yaml", + ) + assert resolved_config_args == { + "config": config, + "config_secret_name": None, + "config_spec_jsonschema": definition.connection_specification, + } + definition.validate_config.assert_called_once_with(config) + assert deploy_calls == [ + { + "name": "My custom source", + "source": config, + "unique": True, + "definition_id": "custom-definition-id", + } + ] + assert registered == ["deployed-id"] + assert "deployed source 'My custom source'" in result + assert "deployed-id" in result + + +@pytest.mark.parametrize( + ("connector_name", "custom_definition_id"), + [ + pytest.param(None, None, id="neither"), + pytest.param("source-faker", "custom-definition-id", id="both"), + ], +) +def test_deploy_connector_to_cloud_requires_one_definition_selector( + connector_name: str | None, + custom_definition_id: str | None, +) -> None: + """Exactly one of a registry connector name or custom definition ID is required.""" + with pytest.raises(AirbyteLibInputError, match="exactly one"): + cloud_mcp.deploy_connector_to_cloud( + cast(Context, object()), + name="My connector", + connector_name=connector_name, + custom_definition_id=custom_definition_id, + connector_type=None, + workspace_id=None, + config=None, + config_secret_name=None, + unique=True, + ) + + +def test_deploy_connector_to_cloud_rejects_custom_destination() -> None: + """Custom source definitions cannot be deployed as destinations.""" + with pytest.raises(AirbyteLibInputError, match="only be deployed as sources"): + cloud_mcp.deploy_connector_to_cloud( + cast(Context, object()), + name="My connector", + connector_name=None, + custom_definition_id="custom-definition-id", + connector_type=ConnectorType.DESTINATION, + workspace_id=None, + config=None, + config_secret_name=None, + unique=True, + ) + + +def test_deploy_connector_to_cloud_deferred_custom_definition_skips_registry( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Deferred custom source deployment uses its definition ID without registry lookup.""" + deploy_calls: list[dict[str, object]] = [] + + def fake_deploy(**kwargs: object) -> SimpleNamespace: + deploy_calls.append(kwargs) + return SimpleNamespace( + connector_id="deployed-id", + connector_url="https://cloud.airbyte.com/deployed-id", + ) + + workspace = SimpleNamespace(workspace_id="workspace-id", deploy_source=fake_deploy) + monkeypatch.setattr(cloud_mcp, "_get_cloud_workspace", lambda *_: workspace) + monkeypatch.setattr( + cloud_mcp, + "get_connector_metadata", + lambda *_: pytest.fail("custom definitions must not look up registry metadata"), + ) + registered: list[str] = [] + monkeypatch.setattr( + cloud_mcp, "register_guid_created_in_session", registered.append + ) + + result = cloud_mcp.deploy_connector_to_cloud( + cast(Context, object()), + name="My custom source", + connector_name=None, + custom_definition_id="custom-definition-id", + connector_type=None, + workspace_id=None, + config={"greeting": "hi"}, + config_secret_name=None, + unique=True, + defer_credentials=True, + ) + + assert deploy_calls == [ + { + "name": "My custom source", + "source": {"greeting": "hi"}, + "unique": True, + "definition_id": "custom-definition-id", + "defer_credentials": True, + } + ] + assert registered == ["deployed-id"] + assert '"connector_id": "deployed-id"' in result + + class _LazyConnector: """Connector double whose `name` raises `AirbyteMissingResourceError` when missing.""" From c27a6d0fb6a4c22e9f1aa058752bbacb678d10ad Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:06:09 +0000 Subject: [PATCH 2/6] refactor(cloud): reuse config json-path helper; drop unreachable guard Co-Authored-By: AJ Steers --- airbyte/cloud/connectors.py | 10 +++------- airbyte/mcp/cloud.py | 5 +---- 2 files changed, 4 insertions(+), 11 deletions(-) diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py index 73fbc3727..260242aa0 100644 --- a/airbyte/cloud/connectors.py +++ b/airbyte/cloud/connectors.py @@ -65,6 +65,7 @@ from pydantic import ValidationError from airbyte import exceptions as exc +from airbyte._connector_base import _json_path_from_absolute_path from airbyte._direct_connectors import api_util as agents_api_util from airbyte._direct_connectors import connector_docs from airbyte._direct_connectors.actions import ( @@ -1342,16 +1343,11 @@ def validate_config(self, config: dict[str, Any]) -> None: try: jsonschema.validate(config, specification) except jsonschema.ValidationError as ex: - json_path = "$" - for part in ex.absolute_path: - if isinstance(part, str): - json_path += f".{part}" - elif isinstance(part, int): - json_path += f"[{part}]" raise exc.AirbyteConnectorValidationFailedError( connector_name=self.name, message=( - f"The provided config is not valid at `{json_path}` " + f"The provided config is not valid at " + f"`{_json_path_from_absolute_path(ex.absolute_path)}` " f"(failed the `{ex.validator}` rule)." ), context={ diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index 56275d79c..c619f64d2 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -786,10 +786,7 @@ def _deploy_deferred_to_cloud( # noqa: PLR0913 if custom_definition_id is not None: definition_id = custom_definition_id else: - if connector_name is None: - raise AirbyteLibInputError( - message="Provide a connector name when deploying a registry connector." - ) + assert connector_name is not None metadata = get_connector_metadata(connector_name) if metadata is None or metadata.definition_id is None: raise AirbyteConnectorNotRegisteredError(connector_name=connector_name) From 4bef05348934d0a8e575232baa444373ab80e949 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 15 Oct 2026 18:14:51 +0000 Subject: [PATCH 3/6] fix(cloud): hydrate secret refs for custom-definition deploys; best-effort custom spec fallback Co-Authored-By: AJ Steers --- airbyte/cloud/workspaces.py | 2 +- airbyte/mcp/cloud.py | 2 + .../test_deferred_cloud_credentials.py | 31 +++++++++++ tests/unit_tests/test_mcp_cloud.py | 54 ++++++++++++++++++- 4 files changed, 87 insertions(+), 2 deletions(-) diff --git a/airbyte/cloud/workspaces.py b/airbyte/cloud/workspaces.py index 7c12f1d6f..6132bdb44 100644 --- a/airbyte/cloud/workspaces.py +++ b/airbyte/cloud/workspaces.py @@ -808,7 +808,7 @@ def _deploy_deferred( """Create a connector with deferred credentials on the Config API and return its ID.""" spec_json_schema = _get_deferred_spec(definition_id) if definition_id else None if spec_json_schema is None and connector_type == "source" and definition_id: - with contextlib.suppress(exc.AirbyteLibError): + with contextlib.suppress(Exception): spec_json_schema = self.get_custom_source_definition( definition_id, definition_type="yaml", diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index c619f64d2..e81400d18 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -102,6 +102,7 @@ get_connector_metadata, get_connector_metadata_by_definition_id, ) +from airbyte.secrets.hydration import hydrate_secrets if TYPE_CHECKING: @@ -730,6 +731,7 @@ def deploy_connector_to_cloud( # noqa: PLR0913 # Mirrors the API surface. config_secret_name=config_secret_name, config_spec_jsonschema=definition.connection_specification, ) + config_dict = hydrate_secrets(config_dict) definition.validate_config(config_dict) deployed: CloudConnector = workspace.deploy_source( name=name, diff --git a/tests/unit_tests/test_deferred_cloud_credentials.py b/tests/unit_tests/test_deferred_cloud_credentials.py index 44e2935a3..3737ade0a 100644 --- a/tests/unit_tests/test_deferred_cloud_credentials.py +++ b/tests/unit_tests/test_deferred_cloud_credentials.py @@ -535,6 +535,37 @@ def test_deploy_deferred_uses_custom_source_spec_for_secret_detection( assert raised.value.context == {"fields": ["my_token_field"]} +def test_deploy_deferred_uses_global_mask_when_custom_spec_fetch_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A failed custom-spec lookup does not block deferred deploy or bypass the global mask.""" + call = _stub_create(monkeypatch, "source") + monkeypatch.setattr(cloud_workspaces, "_get_deferred_spec", lambda _id: None) + workspace = _workspace() + get_definition = MagicMock(side_effect=RuntimeError("custom spec lookup failed")) + monkeypatch.setattr(workspace, "get_custom_source_definition", get_definition) + + deployed = workspace.deploy_source( + name="My custom source", + source={"my_token_field": "plain-text-value"}, + definition_id=DEFINITION_ID, + defer_credentials=True, + ) + + assert deployed.connector_id == ACTOR_ID + assert call.kwargs["config"] == {"my_token_field": "plain-text-value"} + with pytest.raises(exc.AirbyteLibInputError, match="credential values") as raised: + workspace.deploy_source( + name="My custom source with a masked credential", + source={"password": "hunter2"}, + definition_id=DEFINITION_ID, + defer_credentials=True, + ) + + assert raised.value.context == {"fields": ["password"]} + assert get_definition.call_count == 2 + + def test_get_connector_name_by_definition_id( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py index 578613d49..cbb720067 100644 --- a/tests/unit_tests/test_mcp_cloud.py +++ b/tests/unit_tests/test_mcp_cloud.py @@ -36,7 +36,7 @@ ConnectionStatus, JobStatusEnum, ) -from airbyte.constants import MCP_INSIDERS_ENV_VAR +from airbyte.constants import MCP_INSIDERS_ENV_VAR, MCP_TRUSTED_EXECUTION_ENV_VAR from airbyte.mcp import cloud as cloud_mcp from airbyte.mcp import guidance as guidance_mcp from airbyte.mcp._arg_resolvers import resolve_list_of_dicts @@ -1264,6 +1264,58 @@ def fake_resolve_config(**kwargs: object) -> dict[str, Any]: assert "deployed-id" in result +def test_deploy_connector_to_cloud_hydrates_custom_definition_secret_references( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Custom source secret references are hydrated before validation and deployment.""" + monkeypatch.setenv(MCP_TRUSTED_EXECUTION_ENV_VAR, "1") + monkeypatch.setenv("SOME_ENV", "resolved-value") + + definition = MagicMock() + definition.connection_specification = { + "type": "object", + "properties": {"api_key": {"type": "string", "airbyte_secret": True}}, + } + deploy_calls: list[dict[str, object]] = [] + + def fake_deploy(**kwargs: object) -> SimpleNamespace: + deploy_calls.append(kwargs) + return SimpleNamespace( + connector_id="deployed-id", + connector_url="https://cloud.airbyte.com/deployed-id", + ) + + workspace = SimpleNamespace( + get_custom_source_definition=MagicMock(return_value=definition), + deploy_source=fake_deploy, + ) + monkeypatch.setattr(cloud_mcp, "_get_cloud_workspace", lambda *_: workspace) + monkeypatch.setattr(cloud_mcp, "register_guid_created_in_session", lambda _: None) + + cloud_mcp.deploy_connector_to_cloud( + cast(Context, object()), + name="My custom source", + connector_name=None, + custom_definition_id="custom-definition-id", + connector_type=None, + workspace_id=None, + config={"api_key": "secret_reference::SOME_ENV"}, + config_secret_name=None, + unique=True, + ) + + resolved_config = {"api_key": "resolved-value"} + definition.validate_config.assert_called_once_with(resolved_config) + assert deploy_calls == [ + { + "name": "My custom source", + "source": resolved_config, + "unique": True, + "definition_id": "custom-definition-id", + } + ] + + @pytest.mark.parametrize( ("connector_name", "custom_definition_id"), [ From 118fab7dbcbbd8016758f816676b3335ede27ba0 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 16 Oct 2026 00:11:00 +0000 Subject: [PATCH 4/6] refactor: rename connection_specification to config_spec_jsonschema Co-Authored-By: AJ Steers --- airbyte/cloud/connectors.py | 10 +++++----- airbyte/cloud/workspaces.py | 2 +- airbyte/mcp/cloud.py | 2 +- tests/unit_tests/test_cloud_credentials.py | 4 ++-- tests/unit_tests/test_deferred_cloud_credentials.py | 4 ++-- tests/unit_tests/test_mcp_cloud.py | 6 +++--- 6 files changed, 14 insertions(+), 14 deletions(-) diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py index 260242aa0..ec5d066d5 100644 --- a/airbyte/cloud/connectors.py +++ b/airbyte/cloud/connectors.py @@ -1323,7 +1323,7 @@ def manifest(self) -> dict[str, Any] | None: return self._definition_info.manifest @property - def connection_specification(self) -> dict[str, Any] | None: + def config_spec_jsonschema(self) -> dict[str, Any] | None: """Get the JSON schema for this custom source's configuration.""" manifest = self.manifest if manifest is None: @@ -1331,12 +1331,12 @@ def connection_specification(self) -> dict[str, Any] | None: manifest_spec = manifest.get("spec") if not isinstance(manifest_spec, dict): return None - connection_specification = manifest_spec.get("connection_specification") - return connection_specification if isinstance(connection_specification, dict) else None + config_spec_jsonschema = manifest_spec.get("connection_specification") + return config_spec_jsonschema if isinstance(config_spec_jsonschema, dict) else None def validate_config(self, config: dict[str, Any]) -> None: """Validate the config against the custom source's connection specification.""" - specification = self.connection_specification + specification = self.config_spec_jsonschema if specification is None: return @@ -1831,7 +1831,7 @@ def set_testing_values( if manifest_spec: spec = { "documentationUrl": manifest_spec.get("documentation_url"), - "connectionSpecification": self.connection_specification or {}, + "connectionSpecification": self.config_spec_jsonschema or {}, "advancedAuth": manifest_spec.get("advanced_auth"), } diff --git a/airbyte/cloud/workspaces.py b/airbyte/cloud/workspaces.py index 6132bdb44..0e99c5f5e 100644 --- a/airbyte/cloud/workspaces.py +++ b/airbyte/cloud/workspaces.py @@ -812,7 +812,7 @@ def _deploy_deferred( spec_json_schema = self.get_custom_source_definition( definition_id, definition_type="yaml", - ).connection_specification + ).config_spec_jsonschema config_dict, definition_id = _deferred_credentials_config( config, diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index e81400d18..603dc54c4 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -729,7 +729,7 @@ def deploy_connector_to_cloud( # noqa: PLR0913 # Mirrors the API surface. config_dict = resolve_connector_config( config=config, config_secret_name=config_secret_name, - config_spec_jsonschema=definition.connection_specification, + config_spec_jsonschema=definition.config_spec_jsonschema, ) config_dict = hydrate_secrets(config_dict) definition.validate_config(config_dict) diff --git a/tests/unit_tests/test_cloud_credentials.py b/tests/unit_tests/test_cloud_credentials.py index ccd4eaa37..f9ea9bf47 100644 --- a/tests/unit_tests/test_cloud_credentials.py +++ b/tests/unit_tests/test_cloud_credentials.py @@ -824,7 +824,7 @@ def test_custom_cloud_source_definition_validates_config() -> None: }, ) - assert definition.connection_specification == { + assert definition.config_spec_jsonschema == { "type": "object", "required": ["greeting"], "properties": {"greeting": {"type": "string"}}, @@ -851,7 +851,7 @@ def test_custom_cloud_source_definition_without_spec_skips_config_validation() - manifest=None, ) - assert definition.connection_specification is None + assert definition.config_spec_jsonschema is None definition.validate_config({"anything": "goes"}) diff --git a/tests/unit_tests/test_deferred_cloud_credentials.py b/tests/unit_tests/test_deferred_cloud_credentials.py index 3737ade0a..0651df74c 100644 --- a/tests/unit_tests/test_deferred_cloud_credentials.py +++ b/tests/unit_tests/test_deferred_cloud_credentials.py @@ -50,7 +50,7 @@ def _stub_secret_detection( monkeypatch.setattr( CloudWorkspace, "get_custom_source_definition", - lambda *_args, **_kwargs: SimpleNamespace(connection_specification=None), + lambda *_args, **_kwargs: SimpleNamespace(config_spec_jsonschema=None), ) @@ -512,7 +512,7 @@ def test_deploy_deferred_uses_custom_source_spec_for_secret_detection( workspace = _workspace() get_definition = MagicMock( return_value=SimpleNamespace( - connection_specification={ + config_spec_jsonschema={ "type": "object", "properties": { "my_token_field": {"type": "string", "airbyte_secret": True}, diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py index cbb720067..8436c2496 100644 --- a/tests/unit_tests/test_mcp_cloud.py +++ b/tests/unit_tests/test_mcp_cloud.py @@ -1200,7 +1200,7 @@ def test_deploy_connector_to_cloud_supports_custom_definition( ) -> None: """Custom source definitions are validated and deployed with the supplied ID.""" definition = MagicMock() - definition.connection_specification = {"type": "object"} + definition.config_spec_jsonschema = {"type": "object"} definition.validate_config = MagicMock() deploy_calls: list[dict[str, object]] = [] @@ -1248,7 +1248,7 @@ def fake_resolve_config(**kwargs: object) -> dict[str, Any]: assert resolved_config_args == { "config": config, "config_secret_name": None, - "config_spec_jsonschema": definition.connection_specification, + "config_spec_jsonschema": definition.config_spec_jsonschema, } definition.validate_config.assert_called_once_with(config) assert deploy_calls == [ @@ -1272,7 +1272,7 @@ def test_deploy_connector_to_cloud_hydrates_custom_definition_secret_references( monkeypatch.setenv("SOME_ENV", "resolved-value") definition = MagicMock() - definition.connection_specification = { + definition.config_spec_jsonschema = { "type": "object", "properties": {"api_key": {"type": "string", "airbyte_secret": True}}, } From 13c64928e9d36222292813eec1aded5961f3b76a Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 16 Oct 2026 00:15:24 +0000 Subject: [PATCH 5/6] fix: keep manifest property unchanged after merge Co-Authored-By: AJ Steers --- airbyte/cloud/connectors.py | 11 +++++---- tests/unit_tests/test_cloud_credentials.py | 28 ---------------------- 2 files changed, 7 insertions(+), 32 deletions(-) diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py index 1d63683d5..8723406bd 100644 --- a/airbyte/cloud/connectors.py +++ b/airbyte/cloud/connectors.py @@ -1339,7 +1339,7 @@ def manifest(self) -> dict[str, Any] | None: return None if not self._definition_info: self._definition_info = self._fetch_definition_info() - return _unwrap_builder_manifest(self._definition_info.manifest) + return self._definition_info.manifest @property def config_spec_jsonschema(self) -> dict[str, Any] | None: @@ -1835,11 +1835,14 @@ def set_testing_values( }, ) + # Get the spec from the definition info + if not self._definition_info: + self._definition_info = self._fetch_definition_info() + # Build the spec object from the manifest, matching the Builder UI pattern spec: dict[str, Any] = {} - manifest = self.manifest - if manifest: - manifest_spec = manifest.get("spec", {}) + if self._definition_info.manifest: + manifest_spec = self._definition_info.manifest.get("spec", {}) if manifest_spec: spec = { "documentationUrl": manifest_spec.get("documentation_url"), diff --git a/tests/unit_tests/test_cloud_credentials.py b/tests/unit_tests/test_cloud_credentials.py index 92b39475f..a2ad94e5c 100644 --- a/tests/unit_tests/test_cloud_credentials.py +++ b/tests/unit_tests/test_cloud_credentials.py @@ -868,34 +868,6 @@ def test_custom_cloud_source_definition_validates_config() -> None: definition.validate_config({}) -def test_custom_cloud_source_definition_config_spec_jsonschema_unwraps_manifest() -> ( - None -): - definition = CustomCloudSourceDefinition( - workspace=CloudWorkspace(workspace_id="workspace-id", bearer_token="token"), - definition_id="definition-id", - definition_type="yaml", - ) - connection_specification = { - "type": "object", - "properties": {"greeting": {"type": "string"}}, - } - definition._definition_info = CloudCustomSourceDefinitionInfo( - definition_id="definition-id", - name="Custom source", - version="0.1.0", - manifest={ - "manifest": { - "spec": {"connection_specification": connection_specification} - }, - "spec": {"connectionSpecification": {}}, - "version": 1, - }, - ) - - assert definition.config_spec_jsonschema == connection_specification - - def test_custom_cloud_source_definition_without_spec_skips_config_validation() -> None: definition = CustomCloudSourceDefinition( workspace=CloudWorkspace(workspace_id="workspace-id", bearer_token="token"), From eaa850ca967283a938d443f1b723a5919ea087d3 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 16 Oct 2026 01:30:57 +0000 Subject: [PATCH 6/6] refactor: hydrate custom-definition secrets in core, not MCP Co-Authored-By: AJ Steers --- airbyte/cloud/connectors.py | 3 +- airbyte/cloud/workspaces.py | 4 +- airbyte/mcp/cloud.py | 2 - tests/unit_tests/test_cloud_credentials.py | 31 +++++++++++ .../test_deferred_cloud_credentials.py | 28 ++++++++++ tests/unit_tests/test_mcp_cloud.py | 54 +------------------ 6 files changed, 64 insertions(+), 58 deletions(-) diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py index 8723406bd..88d46d11e 100644 --- a/airbyte/cloud/connectors.py +++ b/airbyte/cloud/connectors.py @@ -100,6 +100,7 @@ _SourceResponseLike, ) from airbyte.registry import ApiDocsUrl, get_connector_api_docs_urls +from airbyte.secrets.hydration import hydrate_secrets if TYPE_CHECKING: @@ -1360,7 +1361,7 @@ def validate_config(self, config: dict[str, Any]) -> None: return try: - jsonschema.validate(config, specification) + jsonschema.validate(hydrate_secrets(config), specification) except jsonschema.ValidationError as ex: raise exc.AirbyteConnectorValidationFailedError( connector_name=self.name, diff --git a/airbyte/cloud/workspaces.py b/airbyte/cloud/workspaces.py index 757da92f9..07c447fa4 100644 --- a/airbyte/cloud/workspaces.py +++ b/airbyte/cloud/workspaces.py @@ -70,7 +70,7 @@ from airbyte.exceptions import AirbyteCloudError from airbyte.registry import _get_connector_name_by_definition_id from airbyte.secrets.base import SecretString -from airbyte.secrets.hydration import detect_hardcoded_secrets +from airbyte.secrets.hydration import detect_hardcoded_secrets, hydrate_secrets if TYPE_CHECKING: @@ -700,7 +700,7 @@ def deploy_source( "`defer_credentials=True` is given." ), ) - source_config_dict = dict(source) + source_config_dict = hydrate_secrets(source) else: source_config_dict = source._hydrated_config.copy() # noqa: SLF001 (non-public API) source_config_dict["sourceType"] = source.name.replace("source-", "") diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index b17f40791..e0c6e7cb2 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -102,7 +102,6 @@ get_connector_metadata, get_connector_metadata_by_definition_id, ) -from airbyte.secrets.hydration import hydrate_secrets if TYPE_CHECKING: @@ -735,7 +734,6 @@ def deploy_connector_to_cloud( # noqa: PLR0913 # Mirrors the API surface. config_secret_name=config_secret_name, config_spec_jsonschema=definition.config_spec_jsonschema, ) - config_dict = hydrate_secrets(config_dict) definition.validate_config(config_dict) deployed: CloudConnector = workspace.deploy_source( name=name, diff --git a/tests/unit_tests/test_cloud_credentials.py b/tests/unit_tests/test_cloud_credentials.py index a2ad94e5c..2c4cafd44 100644 --- a/tests/unit_tests/test_cloud_credentials.py +++ b/tests/unit_tests/test_cloud_credentials.py @@ -868,6 +868,37 @@ def test_custom_cloud_source_definition_validates_config() -> None: definition.validate_config({}) +def test_custom_cloud_source_definition_validates_hydrated_secret_reference( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("SOME_ENV", "resolved-value") + definition = CustomCloudSourceDefinition( + workspace=CloudWorkspace(workspace_id="workspace-id", bearer_token="token"), + definition_id="definition-id", + definition_type="yaml", + ) + definition._definition_info = CloudCustomSourceDefinitionInfo( + definition_id="definition-id", + name="Custom source", + version="0.1.0", + manifest={ + "spec": { + "connection_specification": { + "type": "object", + "properties": { + "api_key": {"type": "string", "pattern": "^resolved"} + }, + } + } + }, + ) + config = {"api_key": "secret_reference::SOME_ENV"} + + definition.validate_config(config) + + assert config == {"api_key": "secret_reference::SOME_ENV"} + + def test_custom_cloud_source_definition_without_spec_skips_config_validation() -> None: definition = CustomCloudSourceDefinition( workspace=CloudWorkspace(workspace_id="workspace-id", bearer_token="token"), diff --git a/tests/unit_tests/test_deferred_cloud_credentials.py b/tests/unit_tests/test_deferred_cloud_credentials.py index 0651df74c..c7a39fab1 100644 --- a/tests/unit_tests/test_deferred_cloud_credentials.py +++ b/tests/unit_tests/test_deferred_cloud_credentials.py @@ -406,6 +406,34 @@ def test_deploy_source_passes_custom_definition_config_without_source_type( assert create_kwargs["definition_id"] == DEFINITION_ID +def test_deploy_source_hydrates_custom_definition_secret_references( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Custom source secret references are hydrated without mutating the input config.""" + monkeypatch.setenv("SOME_ENV", "resolved-value") + source_config = {"api_key": "secret_reference::SOME_ENV"} + create_kwargs: dict[str, Any] = {} + monkeypatch.setattr( + api_util, + "create_source", + lambda **kwargs: ( + create_kwargs.update(kwargs) or SimpleNamespace(source_id=ACTOR_ID) + ), + ) + monkeypatch.setattr(api_util, "list_sources", lambda **_: []) + workspace = _workspace() + + deployed = workspace.deploy_source( + name="My custom source", + source=source_config, + definition_id=DEFINITION_ID, + ) + + assert deployed.connector_id == ACTOR_ID + assert create_kwargs["config"] == {"api_key": "resolved-value"} + assert source_config == {"api_key": "secret_reference::SOME_ENV"} + + @pytest.mark.parametrize("connector_type", CONNECTOR_TYPES) def test_deploy_deferred_rejects_plaintext_credentials( monkeypatch: pytest.MonkeyPatch, diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py index 56c5ded17..ab9558b24 100644 --- a/tests/unit_tests/test_mcp_cloud.py +++ b/tests/unit_tests/test_mcp_cloud.py @@ -36,7 +36,7 @@ ConnectionStatus, JobStatusEnum, ) -from airbyte.constants import MCP_INSIDERS_ENV_VAR, MCP_TRUSTED_EXECUTION_ENV_VAR +from airbyte.constants import MCP_INSIDERS_ENV_VAR from airbyte.mcp import cloud as cloud_mcp from airbyte.mcp import guidance as guidance_mcp from airbyte.mcp._arg_resolvers import resolve_list_of_dicts @@ -1301,58 +1301,6 @@ def fake_resolve_config(**kwargs: object) -> dict[str, Any]: assert "deployed-id" in result -def test_deploy_connector_to_cloud_hydrates_custom_definition_secret_references( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """Custom source secret references are hydrated before validation and deployment.""" - monkeypatch.setenv(MCP_TRUSTED_EXECUTION_ENV_VAR, "1") - monkeypatch.setenv("SOME_ENV", "resolved-value") - - definition = MagicMock() - definition.config_spec_jsonschema = { - "type": "object", - "properties": {"api_key": {"type": "string", "airbyte_secret": True}}, - } - deploy_calls: list[dict[str, object]] = [] - - def fake_deploy(**kwargs: object) -> SimpleNamespace: - deploy_calls.append(kwargs) - return SimpleNamespace( - connector_id="deployed-id", - connector_url="https://cloud.airbyte.com/deployed-id", - ) - - workspace = SimpleNamespace( - get_custom_source_definition=MagicMock(return_value=definition), - deploy_source=fake_deploy, - ) - monkeypatch.setattr(cloud_mcp, "_get_cloud_workspace", lambda *_: workspace) - monkeypatch.setattr(cloud_mcp, "register_guid_created_in_session", lambda _: None) - - cloud_mcp.deploy_connector_to_cloud( - cast(Context, object()), - name="My custom source", - connector_name=None, - custom_definition_id="custom-definition-id", - connector_type=None, - workspace_id=None, - config={"api_key": "secret_reference::SOME_ENV"}, - config_secret_name=None, - unique=True, - ) - - resolved_config = {"api_key": "resolved-value"} - definition.validate_config.assert_called_once_with(resolved_config) - assert deploy_calls == [ - { - "name": "My custom source", - "source": resolved_config, - "unique": True, - "definition_id": "custom-definition-id", - } - ] - - @pytest.mark.parametrize( ("connector_name", "custom_definition_id"), [