diff --git a/airbyte/_util/api_util.py b/airbyte/_util/api_util.py index ab67a76f1..60ad8f969 100644 --- a/airbyte/_util/api_util.py +++ b/airbyte/_util/api_util.py @@ -171,6 +171,19 @@ def _wrap_sdk_error( ) +def _get_sdk_error_detail(error: SDKError) -> str | None: + """Return the human-readable message from an Airbyte API error body, if present.""" + try: + body = json.loads(error.body) + except (json.JSONDecodeError, TypeError): + return None + if not isinstance(body, dict): + return None + data = body.get("data") + candidates = (data.get("message") if isinstance(data, dict) else None, body.get("detail")) + return next((c for c in candidates if isinstance(c, str) and c), None) + + def _infer_config_api_root(api_root: str) -> str | None: """Infer the configuration API root from a public API root.""" normalized_api_root = api_root.rstrip("/") @@ -1906,15 +1919,33 @@ def create_connection( # noqa: PLR0913 # Too many arguments api_root=api_root, ) stream_configurations_obj = build_stream_configurations(selected_stream_names) - response = airbyte_instance.connections.create_connection( - models.ConnectionCreateRequest( - name=name, - source_id=source_id, - destination_id=destination_id, - configurations=stream_configurations_obj, - prefix=prefix, - ), - ) + try: + response = airbyte_instance.connections.create_connection( + models.ConnectionCreateRequest( + name=name, + source_id=source_id, + destination_id=destination_id, + configurations=stream_configurations_obj, + prefix=prefix, + ), + ) + except SDKError as e: + base_context = { + "source_id": source_id, + "destination_id": destination_id, + "selected_stream_names": selected_stream_names, + } + if e.status_code == HTTPStatus.BAD_REQUEST: + raise AirbyteLibInputError( + message=_get_sdk_error_detail(e) + or "The Airbyte API rejected the connection request (HTTP 400).", + guidance=( + "Check that every name in `selected_streams` is a stream of the " + "source, and that `source_id` and `destination_id` are correct." + ), + context={**base_context, **_get_sdk_error_context(e)}, + ) from e + raise _wrap_sdk_error(e, base_context) from e if not status_ok(response.status_code) or response.connection_response is None: raise AirbyteCloudError( context={ diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index aee8cecef..717b305fe 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -754,17 +754,15 @@ def _deploy_deferred_to_cloud( def _get_suggested_streams_for_source( - workspace: CloudWorkspace, - source_id: str, + source: CloudSource, ) -> list[str]: """Resolve the source connector's suggested streams from the connector registry.""" - source = workspace.get_source(source_id) metadata = get_connector_metadata_by_definition_id(source.definition_id) suggested_streams = metadata.suggested_streams if metadata else None if not suggested_streams: raise AirbyteLibInputError( message=( - f"No `selected_streams` provided and source '{source_id}' has no " + f"No `selected_streams` provided and source '{source.connector_id}' has no " "suggested streams in the connector registry." ), guidance=( @@ -834,9 +832,26 @@ def create_connection_on_cloud( """ resolved_streams_list: list[str] = resolve_list_of_strings(selected_streams) or [] workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id) + + source = workspace.get_source(source_id) + destination = workspace.get_destination(destination_id) + for arg_name, connector in (("source_id", source), ("destination_id", destination)): + try: + _ = connector.name # Forces the connector info fetch; objects are lazy. + except AirbyteMissingResourceError as ex: + raise AirbyteMissingResourceError( + message=( + f"`{arg_name}` '{connector.connector_id}' was not found in workspace " + f"'{workspace.workspace_id}', or these credentials can't access it." + ), + guidance=api_util.FORBIDDEN_RESOURCE_GUIDANCE, + resource_type=arg_name.removesuffix("_id"), + resource_name_or_id=connector.connector_id, + ) from ex + used_suggested_streams = False if not resolved_streams_list: - resolved_streams_list = _get_suggested_streams_for_source(workspace, source_id) + resolved_streams_list = _get_suggested_streams_for_source(source) used_suggested_streams = True deployed_connection = workspace.deploy_connection( diff --git a/tests/unit_tests/test_cloud_api_util.py b/tests/unit_tests/test_cloud_api_util.py index e268d865c..72b711938 100644 --- a/tests/unit_tests/test_cloud_api_util.py +++ b/tests/unit_tests/test_cloud_api_util.py @@ -2243,6 +2243,133 @@ def test_api_util_calls_wrap_sdk_errors_with_status_context( get_job.assert_not_called() +def _call_create_connection() -> None: + """Invoke `create_connection` with canned IDs for SDKError tests.""" + api_util.create_connection( + "connection-name", + source_id="source-id", + destination_id="destination-id", + api_root="https://api.airbyte.com/v1", + client_id=None, + client_secret=None, + bearer_token=SecretString("token"), + prefix="", + selected_stream_names=["no_such_stream"], + ) + + +def test_create_connection_400_raises_input_error_with_api_message( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A 400 surfaces the API's `data.message` as an `AirbyteLibInputError`.""" + api_message = ( + "No streams found with name [no_such_stream] and namespace [null]. " + "Please call the https://reference.airbyte.com/reference/getstreamproperties " + "endpoint to see the valid stream name and namespace combinations." + ) + body = json.dumps({ + "status": 400, + "type": "https://reference.airbyte.com/reference/errors#bad-request", + "title": "bad-request", + "detail": "The request could not be understood by the server " + "due to malformed syntax.", + "documentationUrl": None, + "data": {"message": api_message}, + }) + raw_response = requests.Response() + raw_response.status_code = 400 + raw_response.url = "https://api.airbyte.com/v1/connections" + sdk_error = SDKError("API error occurred: Status 400", 400, body, raw_response) + create_connection = Mock(side_effect=sdk_error) + airbyte_instance = SimpleNamespace( + connections=SimpleNamespace(create_connection=create_connection), + ) + monkeypatch.setattr( + api_util, + "get_airbyte_server_instance", + lambda **_: airbyte_instance, + ) + + with pytest.raises(AirbyteLibInputError) as exc_info: + _call_create_connection() + + assert exc_info.value.get_message() == api_message + assert exc_info.value.context["source_id"] == "source-id" + assert exc_info.value.context["destination_id"] == "destination-id" + assert exc_info.value.context["status_code"] == 400 + assert exc_info.value.__cause__ is sdk_error + + +def test_create_connection_400_with_non_json_body_uses_fallback_message( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A 400 with an unparseable body falls back to a generic input error.""" + raw_response = requests.Response() + raw_response.status_code = 400 + raw_response.url = "https://api.airbyte.com/v1/connections" + sdk_error = SDKError( + "API error occurred: Status 400", + 400, + "Bad Request", + raw_response, + ) + airbyte_instance = SimpleNamespace( + connections=SimpleNamespace( + create_connection=Mock(side_effect=sdk_error), + ), + ) + monkeypatch.setattr( + api_util, + "get_airbyte_server_instance", + lambda **_: airbyte_instance, + ) + + with pytest.raises(AirbyteLibInputError) as exc_info: + _call_create_connection() + + assert exc_info.value.get_message() == ( + "The Airbyte API rejected the connection request (HTTP 400)." + ) + assert exc_info.value.__cause__ is sdk_error + + +@pytest.mark.parametrize( + ("status_code", "expect_missing_resource"), + [ + pytest.param(403, True, id="403-missing-resource"), + pytest.param(500, False, id="500-cloud-error"), + ], +) +def test_create_connection_non_400_sdk_errors_keep_status_context( + monkeypatch: pytest.MonkeyPatch, + status_code: int, + expect_missing_resource: bool, +) -> None: + """Non-400 SDK errors go through `_wrap_sdk_error` with the create context.""" + sdk_error = _sdk_status_error(status_code) + airbyte_instance = SimpleNamespace( + connections=SimpleNamespace( + create_connection=Mock(side_effect=sdk_error), + ), + ) + monkeypatch.setattr( + api_util, + "get_airbyte_server_instance", + lambda **_: airbyte_instance, + ) + + with pytest.raises(AirbyteCloudError) as exc_info: + _call_create_connection() + + assert isinstance(exc_info.value, AirbyteMissingResourceError) == ( + expect_missing_resource + ) + assert exc_info.value.context["status_code"] == status_code + assert exc_info.value.context["source_id"] == "source-id" + assert exc_info.value.context["destination_id"] == "destination-id" + assert exc_info.value.__cause__ is sdk_error + + @pytest.mark.parametrize( "source_status", [ diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py index 18afc7dda..15640b586 100644 --- a/tests/unit_tests/test_mcp_cloud.py +++ b/tests/unit_tests/test_mcp_cloud.py @@ -1194,17 +1194,46 @@ def test_deploy_connector_to_cloud_rejects_unknown_prefix() -> None: ) +class _LazyConnector: + """Connector double whose `name` raises `AirbyteMissingResourceError` when missing.""" + + def __init__(self, connector_id: str, *, missing: bool = False) -> None: + self.connector_id = connector_id + self.definition_id = "definition-id" + self._missing = missing + + @property + def name(self) -> str: + """Return the connector name, or raise when the connector is marked missing.""" + if self._missing: + raise AirbyteMissingResourceError( + resource_type="connector", + resource_name_or_id=self.connector_id, + ) + return "connector-name" + + class _DeployConnectionWorkspace: - """Fake `CloudWorkspace` capturing `deploy_connection` and `get_source` calls.""" + """Fake `CloudWorkspace` capturing `deploy_connection` and connector lookups.""" - def __init__(self) -> None: + def __init__(self, missing_ids: set[str] | None = None) -> None: self.deploy_calls: list[dict[str, object]] = [] self.get_source_calls: list[str] = [] + self.get_destination_calls: list[str] = [] + self.workspace_id = "workspace-id" + self._missing_ids = missing_ids or set() - def get_source(self, source_id: str) -> SimpleNamespace: + def get_source(self, source_id: str) -> _LazyConnector: """Return a source double exposing the connector definition ID.""" self.get_source_calls.append(source_id) - return SimpleNamespace(definition_id="definition-id") + return _LazyConnector(source_id, missing=source_id in self._missing_ids) + + def get_destination(self, destination_id: str) -> _LazyConnector: + """Return a destination double.""" + self.get_destination_calls.append(destination_id) + return _LazyConnector( + destination_id, missing=destination_id in self._missing_ids + ) def deploy_connection(self, **kwargs: object) -> SimpleNamespace: """Capture the deploy request and return a connection double.""" @@ -1217,8 +1246,9 @@ def deploy_connection(self, **kwargs: object) -> SimpleNamespace: def _patch_deploy_connection_workspace( monkeypatch: pytest.MonkeyPatch, + missing_ids: set[str] | None = None, ) -> _DeployConnectionWorkspace: - workspace = _DeployConnectionWorkspace() + workspace = _DeployConnectionWorkspace(missing_ids=missing_ids) monkeypatch.setattr(cloud_mcp, "_get_cloud_workspace", lambda *_: workspace) monkeypatch.setattr( cloud_mcp, "register_guid_created_in_session", lambda *args: None @@ -1249,7 +1279,8 @@ def fail_lookup(_definition_id: str) -> None: table_prefix=None, ) - assert workspace.get_source_calls == [] + assert workspace.get_source_calls == ["source-id"] + assert workspace.get_destination_calls == ["destination-id"] assert workspace.deploy_calls == [ { "connection_name": "My Connection", @@ -1285,6 +1316,7 @@ def test_create_connection_on_cloud_defaults_to_suggested_streams( ) assert workspace.get_source_calls == ["source-id"] + assert workspace.get_destination_calls == ["destination-id"] assert workspace.deploy_calls[0]["selected_streams"] == ["users"] assert "suggested streams" in result @@ -1322,6 +1354,50 @@ def test_create_connection_on_cloud_without_suggested_streams_raises( assert workspace.deploy_calls == [] +def test_create_connection_on_cloud_unknown_destination_names_arg( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An unknown destination raises `AirbyteMissingResourceError` naming `destination_id`.""" + workspace = _patch_deploy_connection_workspace( + monkeypatch, missing_ids={"destination-id"} + ) + + with pytest.raises(AirbyteMissingResourceError, match="`destination_id`"): + cloud_mcp.create_connection_on_cloud( + cast(Context, object()), + connection_name="My Connection", + source_id="source-id", + destination_id="destination-id", + workspace_id=None, + selected_streams=["users"], + table_prefix=None, + ) + + assert workspace.deploy_calls == [] + + +def test_create_connection_on_cloud_unknown_source_names_arg( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An unknown source raises `AirbyteMissingResourceError` naming `source_id`.""" + workspace = _patch_deploy_connection_workspace( + monkeypatch, missing_ids={"source-id"} + ) + + with pytest.raises(AirbyteMissingResourceError, match="`source_id`"): + cloud_mcp.create_connection_on_cloud( + cast(Context, object()), + connection_name="My Connection", + source_id="source-id", + destination_id="destination-id", + workspace_id=None, + selected_streams=["users"], + table_prefix=None, + ) + + assert workspace.deploy_calls == [] + + class _CombinedListingWorkspace: """Fake `CloudWorkspace` returning one source and one destination."""