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.
@_fallback_to_project_id_from_variables
            def test_fn(self, *args, **kwargs):
                mock_instance(*args, **kwargs)
        with pytest.raises(
            AirflowException, match="You must use keyword arguments in this methods rather than positional"
        ):
            FixtureFallback().test_fn({"project": "TEST"}, "TEST2")
class TestDataflowHook:
    def setup_method(self):
        self.dataflow_hook = DataflowHook(gcp_conn_id="google_cloud_default")
        self.dataflow_hook.beam_hook = MagicMock()
    @mock.patch("airflow.providers.google.cloud.hooks.dataflow.DataflowHook._authorize")
    @mock.patch("airflow.providers.google.cloud.hooks.dataflow.build")
    def test_dataflow_client_creation(self, mock_build, mock_authorize):
        result = self.dataflow_hook.get_conn()
        mock_build.assert_called_once_with(
            "dataflow", "v1b3", http=mock_authorize.return_value, cache_discovery=False
        )
        assert mock_build.return_value == result
    @mock.patch(DATAFLOW_STRING.format("uuid.uuid4"))
    @mock.patch(DATAFLOW_STRING.format("DataflowHook.wait_for_done"))
    @mock.patch(DATAFLOW_STRING.format("process_line_and_extract_dataflow_job_id_callback"))
    def test_start_python_dataflow(self, mock_callback_on_job_id, mock_dataflow_wait_for_done, mock_uuid):
        mock_beam_start_python_pipeline = self.dataflow_hook.beam_hook.start_python_pipeline
        mock_uuid.return_value = MOCK_UUID
        on_new_job_id_callback = MagicMock()
        py_requirements = ["pandas", "numpy"]
        job_name = f"{JOB_NAME}-{MOCK_UUID_PREFIX}"