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.
def setup_method(self):
        self.dataflow_hook = DataflowHook(gcp_conn_id="google_cloud_default")
    @mock.patch(DATAFLOW_STRING.format("uuid.uuid4"), return_value=MOCK_UUID)
    @mock.patch(DATAFLOW_STRING.format("_DataflowJobsController"))
    @mock.patch(DATAFLOW_STRING.format("DataflowHook.get_conn"))
    def test_start_template_dataflow(self, mock_conn, mock_controller, mock_uuid):
        launch_method = (
            mock_conn.return_value.projects.return_value.locations.return_value.templates.return_value.launch
        )
        launch_method.return_value.execute.return_value = {"job": {"id": TEST_JOB_ID}}
        variables = {"zone": "us-central1-f", "tempLocation": "gs://test/temp"}
        self.dataflow_hook.start_template_dataflow(
            job_name=JOB_NAME,
            variables=copy.deepcopy(variables),
            parameters=PARAMETERS,
            dataflow_template=TEST_TEMPLATE,
            project_id=TEST_PROJECT,
        )
        launch_method.assert_called_once_with(
            body={
                "jobName": f"test-dataflow-pipeline-{MOCK_UUID_PREFIX}",
                "parameters": PARAMETERS,
                "environment": variables,
            },
            gcsPath="gs://dataflow-templates/wordcount/template_file",
            projectId=TEST_PROJECT,
            location=DEFAULT_DATAFLOW_LOCATION,
        )