code: def test_poll_for_termination_fail_deferrable(
      self, mock_livy, mock_dump_logs
  ):
      state_list = 2 * [BatchState.RUNNING] + [BatchState.ERROR]
  
      def side_effect(_, retry_args):
          if state_list:
              return state_list.pop(0)
          # fail if does not stop right before
          raise AssertionError()
  
      mock_livy.side_effect = side_effect
  
      task = LivyOperator(
          file="sparkapp",
          polling_interval=1,
          dag=self.dag,
          task_id="livy_example",
          deferrable=True,
      )
  
      with pytest.raises(AirflowException):
          task.poll_for_termination(BATCH_ID)
  
      mock_livy.assert_called_with(BATCH_ID, retry_args=None)
      mock_dump_logs.assert_called_with(BATCH_ID)
      assert mock_livy.call_count == 3
  
imports: from __future__ import annotations
  import logging
  from unittest.mock import MagicMock, patch
  import pytest
  from airflow.exceptions import AirflowException
  from airflow.models import Connection
  from airflow.models.dag import DAG
  from airflow.providers.apache.livy.hooks.livy import BatchState
  from airflow.providers.apache.livy.operators.livy import LivyOperator
  from airflow.utils import db, timezone
label_desc: Use image node_id 9 for calling the TestLivyOperator obj's underlying member method code with example usage: obj.test_poll_for_termination_fail_deferrable(mock_livy, mock_dump_logs) and returns: state_list