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.
@pytest.fixture
    def hook(self):
        return AsyncDataflowHook(
            gcp_conn_id=TEST_PROJECT_ID,
        )
    @pytest.mark.asyncio
    @mock.patch("airflow.providers.google.cloud.hooks.dataflow.AsyncDataflowHook.initialize_client")
    async def test_get_job(self, initialize_client_mock, hook, make_mock_awaitable):
        client = initialize_client_mock.return_value
        make_mock_awaitable(client.get_job, None)
        await hook.get_job(
            project_id=TEST_PROJECT_ID,
            job_id=TEST_JOB_ID,
            location=TEST_LOCATION,
        )
        request = GetJobRequest(
            dict(
                project_id=TEST_PROJECT_ID,
                job_id=TEST_JOB_ID,
                location=TEST_LOCATION,
                view=JobView.JOB_VIEW_SUMMARY,
            )
        )
        initialize_client_mock.assert_called_once()
        client.get_job.assert_called_once_with(
            request=request,
        )