As discussed in  https://github.com/apache/airflow/pull/15534#discussion_r620500075, we could possibly rewrite Beam Hook to remove the need of storing state in it.
@mock.patch("airflow.providers.apache.beam.hooks.beam.BeamCommandRunner.log")
    @mock.patch("subprocess.Popen")
    @mock.patch("select.select")
    def test_beam_wait_for_done_logging(self, mock_select, mock_popen, mock_logging):
        cmd = ["test", "cmd"]
        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
        beam = BeamCommandRunner(cmd)
        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
        )
        with pytest.raises(Exception):
            beam.wait_for_done()
class TestBeamOptionsToArgs: