code: def test_beam_wait_for_done_logging(self, mock_select, mock_popen):
      cmd = ["test", "cmd"]
      mock_logging = MagicMock()
      mock_logging.info = MagicMock()
      mock_logging.warning = MagicMock()
      mock_proc = MagicMock()
      mock_proc.stderr = MagicMock()
      mock_proc.stderr.readlines = MagicMock(
          return_value=["test\n", "error\n"]
      )
      mock_stderr_fd = MagicMock()
      mock_proc.stderr.fileno = MagicMock(return_value=mock_stderr_fd)
      mock_proc_poll = MagicMock()
      mock_select.return_value = [[mock_stderr_fd]]
  
      def poll_resp_error():
          mock_proc.return_code = 1
          return True
  
      mock_proc_poll.side_effect = [None, poll_resp_error]
      mock_proc.poll = mock_proc_poll
      mock_popen.return_value = mock_proc
      with pytest.raises(Exception):
          run_beam_command(cmd, None, None, mock_logging)
          mock_logging.info.assert_called_once_with(
              "Running command: %s", " ".join(cmd)
          )
          mock_popen.assert_called_once_with(
              cmd,
              shell=False,
              stdout=subprocess.PIPE,
              stderr=subprocess.PIPE,
              close_fds=True,
              cwd=None,
          )
  
imports: from __future__ import annotations
  import copy
  import os
  import re
  import subprocess
  from unittest import mock
  from unittest.mock import ANY, AsyncMock, MagicMock
  import pytest
  from airflow.exceptions import AirflowException
  from airflow.providers.apache.beam.hooks.beam import BeamAsyncHook, BeamHook, beam_options_to_args, run_beam_command
label_desc: Use image node_id 1 for calling the TestBeamRunner obj's underlying member method code with example usage: obj.test_beam_wait_for_done_logging(mock_select, mock_popen) and returns: True