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.
variables=copy.deepcopy(BEAM_VARIABLES_PY),
            py_file=PY_FILE,
            py_options=PY_OPTIONS,
            py_interpreter=py_interpreter,
            process_line_callback=process_line_callback,
        )
        expected_cmd = [
            py_interpreter,
            "-m",
            PY_FILE,
            f"--runner={DEFAULT_RUNNER}",
            "--output=gs://test/output",
            "--labels=foo=bar",
        ]
        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()
    @pytest.mark.parametrize(
        "current_py_requirements, current_py_system_site_packages",
        [
            pytest.param("foo-bar", False, id="requirements without system site-packages"),
            pytest.param("foo-bar", True, id="requirements with system site-packages"),
            pytest.param([], True, id="only system site-packages"),
        ],
    )
    @mock.patch(BEAM_STRING.format("prepare_virtualenv"))
    @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_non_empty_py_requirements_and_without_system_packages(