docstr: Test DataflowHook is created and the right args are passed to
  start_java_pipeline when executing Dataflow runner.
docstr_tok: ["Test","DataflowHook","is","created","and","the","right","args","are","passed","to","start_java_pipeline","when","executing","Dataflow","runner","."]
code: def test_exec_dataflow_runner(
      self,
      gcs_hook,
      dataflow_hook_mock,
      beam_hook_mock,
      persist_link_mock,
  ):
      """
      Test DataflowHook is created and the right args are passed to
      start_java_pipeline when executing Dataflow runner.
      """
  
      dataflow_config = DataflowConfiguration(
          impersonation_chain=TEST_IMPERSONATION_ACCOUNT
      )
      self.operator.runner = "DataflowRunner"
      self.operator.dataflow_config = dataflow_config
      gcs_provide_file = gcs_hook.return_value.provide_file
      magic_mock = mock.MagicMock()
      with pytest.raises(TaskDeferred):
          self.operator.execute(context=magic_mock)
  
      job_name = dataflow_hook_mock.build_dataflow_job_name.return_value
      dataflow_hook_mock.assert_called_once_with(
          gcp_conn_id=dataflow_config.gcp_conn_id,
          poll_sleep=dataflow_config.poll_sleep,
          impersonation_chain=dataflow_config.impersonation_chain,
          drain_pipeline=dataflow_config.drain_pipeline,
          cancel_timeout=dataflow_config.cancel_timeout,
          wait_until_finished=dataflow_config.wait_until_finished,
      )
      expected_options = {
          "project": dataflow_hook_mock.return_value.project_id,
          "job_name": job_name,
          "staging_location": "gs://test/staging",
          "output": "gs://test/output",
          "labels": {"foo": "bar", "airflow-version": TEST_VERSION},
          "region": "us-central1",
          "impersonate_service_account": TEST_IMPERSONATION_ACCOUNT,
      }
      gcs_provide_file.assert_called_once_with(object_url=JAR_FILE)
      persist_link_mock.assert_called_once_with(
          self.operator,
          magic_mock,
          expected_options["project"],
          expected_options["region"],
          self.operator.dataflow_job_id,
      )
      beam_hook_mock.return_value.start_python_pipeline.assert_not_called()
      dataflow_hook_mock.return_value.provide_authorized_gcloud.assert_called_once_with()
  
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 5 for calling the TestBeamRunJavaPipelineOperatorAsync obj's underlying member method code with example usage: obj.test_exec_dataflow_runner(gcs_hook, dataflow_hook_mock, beam_hook_mock, persist_link_mock) without return types