diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py index 571e40e4e..88d46d11e 100644 --- a/airbyte/cloud/connectors.py +++ b/airbyte/cloud/connectors.py @@ -59,11 +59,13 @@ from pathlib import Path from typing import TYPE_CHECKING, Any, ClassVar, Literal, Protocol +import jsonschema import requests import yaml 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 ( @@ -98,6 +100,7 @@ _SourceResponseLike, ) from airbyte.registry import ApiDocsUrl, get_connector_api_docs_urls +from airbyte.secrets.hydration import hydrate_secrets if TYPE_CHECKING: @@ -1274,7 +1277,7 @@ def _unwrap_builder_manifest(value: object) -> dict[str, Any] | None: return value -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. @@ -1339,6 +1342,40 @@ def manifest(self) -> dict[str, Any] | None: self._definition_info = self._fetch_definition_info() return self._definition_info.manifest + @property + 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: + return None + manifest_spec = manifest.get("spec") + if not isinstance(manifest_spec, dict): + return 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.config_spec_jsonschema + if specification is None: + return + + try: + jsonschema.validate(hydrate_secrets(config), specification) + except jsonschema.ValidationError as ex: + raise exc.AirbyteConnectorValidationFailedError( + connector_name=self.name, + message=( + f"The provided config is not valid at " + f"`{_json_path_from_absolute_path(ex.absolute_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.""" @@ -1810,7 +1847,7 @@ def set_testing_values( if manifest_spec: spec = { "documentationUrl": manifest_spec.get("documentation_url"), - "connectionSpecification": manifest_spec.get("connection_specification", {}), + "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 31c1d2caf..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: @@ -667,12 +667,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 @@ -691,12 +693,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 = hydrate_secrets(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()})" @@ -810,10 +817,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(Exception): + spec_json_schema = self.get_custom_source_definition( + definition_id, + definition_type="yaml", + ).config_spec_jsonschema + 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 db8dece8e..e0c6e7cb2 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -610,26 +610,41 @@ 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, Field( @@ -671,40 +686,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.config_spec_jsonschema, + ) + 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 ( @@ -713,12 +769,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, @@ -730,14 +787,19 @@ 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: + 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) + 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) @@ -748,7 +810,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: @@ -756,7 +818,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 9e7aa872c..2c4cafd44 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, ) @@ -831,6 +833,89 @@ 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.config_spec_jsonschema == { + "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_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"), + 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.config_spec_jsonschema 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..c7a39fab1 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(config_spec_jsonschema=None), + ) def _actor_body(connector_type: str, *, draft: bool | None) -> dict[str, Any]: @@ -359,15 +366,74 @@ 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 + + +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, @@ -466,6 +532,68 @@ 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( + config_spec_jsonschema={ + "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_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 671242b9c..ab9558b24 100644 --- a/tests/unit_tests/test_mcp_cloud.py +++ b/tests/unit_tests/test_mcp_cloud.py @@ -1232,6 +1232,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.config_spec_jsonschema = {"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.config_spec_jsonschema, + } + 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."""