code: def _run_train_with_config(time_budget, test_data, tmpdir, **kwargs):
      input_features, output_features, dataset_csv = test_data
      config = {
          "input_features": input_features,
          "output_features": output_features,
          "trainer": {"epochs": 2},
          "hyperopt": {
              "search_alg": {
                  "type": "hyperopt",
                  "random_state_seed": 42,
              },
              "executor": {
                  "type": "ray",
                  "time_budget_s": time_budget,
                  "cpu_resources_per_trial": 1,
                  "scheduler": {
                      "type": "async_hyperband",
                      "max_t": time_budget,
                      "time_attr": "time_total_s",
                      "grace_period": min(72, time_budget),
                      "reduction_factor": 5,
                  },
              },
              "parameters": {
                  "trainer.batch_size": {
                      "space": "choice",
                      "categories": [64, 128, 256],
                  },
                  "trainer.learning_rate": {
                      "space": "loguniform",
                      "lower": 0.001,
                      "upper": 0.1,
                  },
              },
          },
      }
  
      fn = RayTuneExecutor._evaluate_best_model
      with mock.patch(
          "ludwig.hyperopt.execution.RayTuneExecutor._evaluate_best_model"
      ) as mock_fn:
          # We need to check that _evaluate_best_model is called when the time_budget is low
          # as this code path should be triggered when the trial was early stopped
          mock_fn.side_effect = fn
  
          outdir = os.path.join(tmpdir, "output")
          results = train_with_config(
              dataset_csv, config, output_directory=outdir, **kwargs
          )
          try:
              best_model = results.best_model
          except ValueError:
              # ValueError is raised when best_model can't be found. This typically
              # happens when the time_budget is low and the trial is stopped early,
              # resulting in no evaluations happening (and no scores being reported back to RayTune).
              # So RayTune has no way of determining what the best model is.
              best_model = None
  
          if time_budget > 1:
              assert isinstance(best_model, LudwigModel)
              assert best_model.config_obj.trainer.early_stop == -1
              # assert mock_fn.call_count == 1
          else:
              assert best_model is None
              assert mock_fn.call_count == 0
  
imports: import os
  import tempfile
  from typing import List, Set
  from unittest import mock
  import numpy
  import pandas
  import pytest
  from ludwig.api import LudwigModel
  from ludwig.constants import COLUMN, ENCODER, INPUT_FEATURES, NAME, OUTPUT_FEATURES, PREPROCESSING, SPLIT, TYPE
  from ludwig.schema.model_types.base import ModelConfig
  from ludwig.types import FeatureConfigDict, ModelConfigDict
  from ludwig.utils.misc_utils import merge_dict
  from tests.integration_tests.utils import binary_feature, category_feature, generate_data, image_feature, minio_test_creds, number_feature, private_param, remote_tmpdir, text_feature
  import dask.dataframe
  from ray.tune.experiment.trial import Trial
  from ludwig.automl import auto_train, create_auto_config, train_with_config
  from ludwig.automl.automl import OUTPUT_DIR
  from ludwig.hyperopt.execution import RayTuneExecutor
label_desc: Use image node_id 18 for calling a global function with example usage: _run_train_with_config(time_budget, test_data, tmpdir) without return types