test_connection async ¶

async def test_connection(
    self,
    connector_id: "capo_transfer.types.connector_id.ConnectorId",
    *,
    config_overrides: Optional[AsyncTransferClientConfig] = None,
) -> "capo_transfer.types.test_connection_response.TestConnectionResponse":
    """<p>Tests whether your SFTP connector is set up successfully. We highly recommend that you call this operation to test your ability to transfer files between local Amazon Web Services storage and a trading partner's SFTP server.</p>

    Args:
        connector_id: <p>The unique identifier for the connector.</p>

    Raises:
        capo_transfer.errors.internal_service_error.InternalServiceError: <p>This exception is thrown when an error occurs in the Transfer Family service.</p>
        capo_transfer.errors.invalid_request_exception.InvalidRequestException: <p>This exception is thrown when the client submits a malformed request.</p>
        capo_transfer.errors.resource_not_found_exception.ResourceNotFoundException: <p>This exception is thrown when a resource is not found by the Amazon Web ServicesTransfer Family service.</p>
        capo_transfer.errors.service_unavailable_exception.ServiceUnavailableException: <p>The request has failed because the Amazon Web ServicesTransfer Family service is not available.</p>
        capo_transfer.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    async def _handler(
        req: "AsyncOperationRequest[capo_transfer.types.test_connection_request.TestConnectionRequest]",
    ) -> AsyncOperationResponse[
        "capo_transfer.types.test_connection_response.TestConnectionResponse"
    ]:
        import capo_transfer._operations.transfer_service.test_connection

        (
            output,
            http_response,
        ) = await capo_transfer._operations.transfer_service.test_connection.async_test_connection(
            req.options, req.input
        )
        return AsyncOperationResponse(output=output, response=http_response)

    interceptors_, options_ = self.operation_options(
        config_overrides, OperationLogging("TestConnection")
    )
    input_: capo_transfer.types.test_connection_request.TestConnectionRequest = {
        "connector_id": connector_id
    }

    response = await aexecute_pipeline(
        AsyncOperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    await response.response.aclose()
    return response.output