code: def test_poke(
      self, mock_hook, job_current_state, fail_on_terminal_state
  ):
      mock_get_job = mock_hook.return_value.get_job
      mock_fetch_job_messages_by_id = (
          mock_hook.return_value.fetch_job_messages_by_id
      )
      callback = mock.MagicMock()
  
      task = DataflowJobMessagesSensor(
          task_id=TEST_TASK_ID,
          job_id=TEST_JOB_ID,
          callback=callback,
          fail_on_terminal_state=fail_on_terminal_state,
          location=TEST_LOCATION,
          project_id=TEST_PROJECT_ID,
          gcp_conn_id=TEST_GCP_CONN_ID,
          impersonation_chain=TEST_IMPERSONATION_CHAIN,
      )
      mock_get_job.return_value = {
          "id": TEST_JOB_ID,
          "currentState": job_current_state,
      }
  
      results = task.poke(mock.MagicMock())
  
      assert callback.return_value == results
  
      mock_hook.assert_called_once_with(
          gcp_conn_id=TEST_GCP_CONN_ID,
          impersonation_chain=TEST_IMPERSONATION_CHAIN,
      )
      mock_fetch_job_messages_by_id.assert_called_once_with(
          job_id=TEST_JOB_ID,
          project_id=TEST_PROJECT_ID,
          location=TEST_LOCATION,
      )
      callback.assert_called_once_with(
          mock_fetch_job_messages_by_id.return_value
      )
  
file_path: airflow/tests/providers/google/cloud/sensors/test_dataflow.py
imports: from __future__ import annotations
  from unittest import mock
  import pytest
  from airflow.exceptions import AirflowException, AirflowSkipException
  from airflow.providers.google.cloud.hooks.dataflow import DataflowJobStatus
  from airflow.providers.google.cloud.sensors.dataflow import DataflowJobAutoScalingEventsSensor, DataflowJobMessagesSensor, DataflowJobMetricsSensor, DataflowJobStatusSensor
label_desc: Use image node_id 1 for calling the DataflowJobMessagesSensorTest obj's underlying member method code with example usage: obj.test_poke(mock_hook, job_current_state, fail_on_terminal_state) without return types