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_run.return_value = mock.MagicMock(
            stdout=f"{TEST_JOB_ID}\n".encode(), stderr=f"{TEST_JOB_ID}\n".encode(), returncode=1
        )
        with pytest.raises(AirflowException):
            self.dataflow_hook.start_sql_job(
                job_name=TEST_SQL_JOB_NAME,
                query=TEST_SQL_QUERY,
                options=TEST_SQL_OPTIONS,
                location=TEST_LOCATION,
                project_id=TEST_PROJECT,
                on_new_job_callback=mock.MagicMock(),
            )
class TestDataflowJob: