code: def test_poke_raise_exception(
      self, mock_hook, soft_fail, expected_exception
  ):
      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=True,
          location=TEST_LOCATION,
          project_id=TEST_PROJECT_ID,
          gcp_conn_id=TEST_GCP_CONN_ID,
          impersonation_chain=TEST_IMPERSONATION_CHAIN,
          soft_fail=soft_fail,
      )
      mock_get_job.return_value = {
          "id": TEST_JOB_ID,
          "currentState": DataflowJobStatus.JOB_STATE_DONE,
      }
  
      with pytest.raises(
          expected_exception,
          match=f"Job with id '{TEST_JOB_ID}' is already in terminal state: "
          f"{DataflowJobStatus.JOB_STATE_DONE}",
      ):
          task.poke(mock.MagicMock())
  
      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_not_called()
      callback.assert_not_called()
  
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 2 for calling the DataflowJobMessagesSensorTest obj's underlying member method code with example usage: obj.test_poke_raise_exception(mock_hook, soft_fail, expected_exception) without return types