docstr: Test start_go_pipeline_from_binary is called with an executable launcher binary downloaded from GCS.
docstr_tok: ["Test","start_go_pipeline_from_binary","is","called","with","an","executable","launcher","binary","downloaded","from","GCS","."]
code: def test_exec_direct_runner_with_gcs_launcher_binary(
      self, mock_tmp_dir, mock_beam_hook, mock_gcs_hook, tmp_path
  ):
      """
      Test start_go_pipeline_from_binary is called with an executable launcher binary downloaded from GCS.
      """
  
      def tmp_dir_side_effect(prefix: str) -> str:
          sub_dir = tmp_path / mock_tmp_dir.call_args.kwargs["prefix"]
          sub_dir.mkdir()
          return str(sub_dir)
  
      mock_tmp_dir.return_value.__enter__.side_effect = (
          tmp_dir_side_effect
      )
  
      def gcs_download_side_effect(
          bucket_name: str, object_name: str, filename: str
      ) -> None:
          open(filename, "wb").close()
  
      gcs_download_method = mock_gcs_hook.return_value.download
      gcs_download_method.side_effect = gcs_download_side_effect
  
      start_go_pipeline_method = (
          mock_beam_hook.return_value.start_go_pipeline_with_binary
      )
  
      operator = BeamRunGoPipelineOperator(
          task_id=TASK_ID,
          launcher_binary="gs://bucket/path/to/main",
          default_pipeline_options=DEFAULT_OPTIONS,
          pipeline_options=ADDITIONAL_OPTIONS,
      )
      operator.execute({})
  
      expected_binary = f"{tmp_path}/apache-beam-go/launcher-main"
      expected_options = {
          "project": "test",
          "staging_location": "gs://test/staging",
          "output": "gs://test/output",
          "labels": {"foo": "bar", "airflow-version": TEST_VERSION},
      }
      mock_beam_hook.assert_called_once_with(runner=DEFAULT_RUNNER)
      mock_tmp_dir.assert_called_once_with(prefix="apache-beam-go")
      gcs_download_method.assert_called_once_with(
          bucket_name="bucket",
          object_name="path/to/main",
          filename=expected_binary,
      )
      assert os.access(expected_binary, os.X_OK)
      start_go_pipeline_method.assert_called_once_with(
          variables=expected_options,
          launcher_binary=expected_binary,
          worker_binary=expected_binary,
          process_line_callback=None,
      )
  
imports: from __future__ import annotations
  import os
  from unittest import mock
  from unittest.mock import MagicMock, call
  import pytest
  from airflow.exceptions import AirflowException, TaskDeferred
  from airflow.providers.apache.beam.operators.beam import BeamBasePipelineOperator, BeamRunGoPipelineOperator, BeamRunJavaPipelineOperator, BeamRunPythonPipelineOperator
  from airflow.providers.apache.beam.triggers.beam import BeamJavaPipelineTrigger, BeamPythonPipelineTrigger
  from airflow.providers.google.cloud.operators.dataflow import DataflowConfiguration
  from airflow.version import version
label_desc: Use image node_id 8 for calling the TestBeamRunGoPipelineOperator obj's underlying member method code with example usage: obj.test_exec_direct_runner_with_gcs_launcher_binary(mock_tmp_dir, mock_beam_hook, mock_gcs_hook, tmp_path) and returns: str