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_runner.assert_called_once_with(
            cmd=expected_cmd, process_line_callback=process_line_callback, working_directory=None
        )
        wait_for_done.assert_called_once_with()
        mock_virtualenv.assert_called_once_with(
            venv_directory=mock.ANY,
            python_bin="python3",
            system_site_packages=current_py_system_site_packages,
            requirements=current_py_requirements,
        )
    @mock.patch(BEAM_STRING.format("BeamCommandRunner"))
    @mock.patch("airflow.providers.apache.beam.hooks.beam.subprocess.check_output", return_value=b"2.39.0")
    def test_start_python_pipeline_with_empty_py_requirements_and_without_system_packages(
        self, mock_check_output, mock_runner
    ):
        hook = BeamHook(runner=DEFAULT_RUNNER)
        wait_for_done = mock_runner.return_value.wait_for_done
        process_line_callback = MagicMock()
        with pytest.raises(AirflowException, match=r"Invalid method invocation\."):
            hook.start_python_pipeline(
                variables=copy.deepcopy(BEAM_VARIABLES_PY),
                py_file=PY_FILE,
                py_options=PY_OPTIONS,
                py_requirements=[],
                process_line_callback=process_line_callback,
            )
        mock_runner.assert_not_called()
        wait_for_done.assert_not_called()
    @mock.patch(BEAM_STRING.format("BeamCommandRunner"))
    def test_start_java_pipeline(self, mock_runner):