code: def test_eval_set_no_early_stopping(
      task, output, eval_sizes, eval_names_prefix, cluster
  ):
      if task == "ranking" and output == "scipy_csr_matrix":
          pytest.skip(
              "LGBMRanker is not currently tested on sparse matrices"
          )
  
      with Client(cluster) as client:
          # Use larger trainset to prevent premature stopping due to zero loss, causing num_trees() < n_estimators.
          # Use small chunk_size to avoid single-worker allocation of eval data partitions.
          n_samples = 1000
          chunk_size = 10
          n_eval_sets = len(eval_sizes)
          eval_set = []
          eval_sample_weight = []
          eval_class_weight = None
          eval_init_score = None
  
          if eval_names_prefix:
              eval_names = [
                  f"{eval_names_prefix}_{i}"
                  for i in range(len(eval_sizes))
              ]
          else:
              eval_names = None
  
          X, y, w, g, dX, dy, dw, dg = _create_data(
              objective=task,
              n_samples=n_samples,
              output=output,
              chunk_size=chunk_size,
          )
  
          if task == "ranking":
              eval_metrics = ["ndcg"]
              eval_at = (5, 6)
              eval_metric_names = [f"ndcg@{k}" for k in eval_at]
              eval_group = []
          else:
              # test eval_class_weight, eval_init_score on binary-classification task.
              # Note: objective's default `metric` will be evaluated in evals_result_ in addition to all eval_metrics.
              if task == "binary-classification":
                  eval_metrics = ["binary_error", "auc"]
                  eval_metric_names = [
                      "binary_logloss",
                      "binary_error",
                      "auc",
                  ]
                  eval_class_weight = []
                  eval_init_score = []
              elif task == "multiclass-classification":
                  eval_metrics = ["multi_error"]
                  eval_metric_names = ["multi_logloss", "multi_error"]
              elif task == "regression":
                  eval_metrics = ["l1"]
                  eval_metric_names = ["l2", "l1"]
  
          # create eval_sets by creating new datasets or copying training data.
          for eval_size in eval_sizes:
              if eval_size == 1:
                  y_e = y
                  dX_e = dX
                  dy_e = dy
                  dw_e = dw
                  dg_e = dg
              else:
                  n_eval_samples = max(
                      chunk_size, int(n_samples * eval_size)
                  )
                  _, y_e, _, _, dX_e, dy_e, dw_e, dg_e = _create_data(
                      objective=task,
                      n_samples=n_eval_samples,
                      output=output,
                      chunk_size=chunk_size,
                  )
  
              eval_set.append((dX_e, dy_e))
              eval_sample_weight.append(dw_e)
              if task == "ranking":
                  eval_group.append(dg_e)
  
              if task == "binary-classification":
                  n_neg = np.sum(y_e == 0)
                  n_pos = np.sum(y_e == 1)
                  eval_class_weight.append(
                      {0: n_neg / n_pos, 1: n_pos / n_neg}
                  )
                  init_score_value = np.log(
                      np.mean(y_e) / (1 - np.mean(y_e))
                  )
                  if "dataframe" in output:
                      d_init_score = dy_e.map_partitions(
                          lambda x, val=init_score_value: pd.Series(
                              [val] * x.size
                          )
                      )
                  else:
                      d_init_score = dy_e.map_blocks(
                          lambda x, val=init_score_value: np.repeat(
                              val, x.size
                          )
                      )
  
                  eval_init_score.append(d_init_score)
  
          fit_trees = 50
          params = {
              "random_state": 42,
              "n_estimators": fit_trees,
              "num_leaves": 2,
          }
  
          model_factory = task_to_dask_factory[task]
          dask_model = model_factory(client=client, **params)
  
          fit_params = {
              "X": dX,
              "y": dy,
              "eval_set": eval_set,
              "eval_names": eval_names,
              "eval_sample_weight": eval_sample_weight,
              "eval_init_score": eval_init_score,
              "eval_metric": eval_metrics,
          }
          if task == "ranking":
              fit_params.update(
                  {
                      "group": dg,
                      "eval_group": eval_group,
                      "eval_at": eval_at,
                  }
              )
          elif task == "binary-classification":
              fit_params.update(
                  {"eval_class_weight": eval_class_weight}
              )
  
          if eval_sizes == [0]:
              with pytest.warns(
                  UserWarning,
                  match="Worker (.*) was not allocated eval_set data. Therefore evals_result_ and best_score_ data may be unreliable.",
              ):
                  dask_model.fit(**fit_params)
          else:
              dask_model = dask_model.fit(**fit_params)
  
              # total number of trees scales up for ova classifier.
              if task == "multiclass-classification":
                  model_trees = fit_trees * dask_model.n_classes_
              else:
                  model_trees = fit_trees
  
              # check that early stopping was not applied.
              assert dask_model.booster_.num_trees() == model_trees
              assert dask_model.best_iteration_ == 0
  
              # checks that evals_result_ and best_score_ contain expected data and eval_set names.
              evals_result = dask_model.evals_result_
              best_scores = dask_model.best_score_
              assert len(evals_result) == n_eval_sets
              assert len(best_scores) == n_eval_sets
  
              for eval_name in evals_result:
                  assert eval_name in dask_model.best_score_
                  if eval_names:
                      assert eval_name in eval_names
  
                  # check that each eval_name and metric exists for all eval sets, allowing for the
                  # case when a worker receives a fully-padded eval_set component which is not evaluated.
                  if evals_result[eval_name] != {}:
                      for metric in eval_metric_names:
                          assert metric in evals_result[eval_name]
                          assert metric in best_scores[eval_name]
                          assert (
                              len(evals_result[eval_name][metric])
                              == fit_trees
                          )
  
imports: import inspect
  import random
  import socket
  from itertools import groupby
  from os import getenv
  from platform import machine
  from sys import platform
  from urllib.parse import urlparse
  import pytest
  from sklearn.metrics import accuracy_score, r2_score
  import lightgbm
  from .utils import sklearn_multiclass_custom_objective
  import dask.array
  import dask.dataframe
  import numpy
  import pandas
  import sklearn.utils.estimator_checks
  from dask.array.utils import assert_eq
  from dask.distributed import Client, LocalCluster, default_client, wait
  from scipy.sparse import csc_matrix, csr_matrix
  from scipy.stats import spearmanr
  from sklearn.datasets import make_blobs, make_regression
  from .utils import make_ranking, pickle_obj, unpickle_obj
label_desc: Use image node_id 24 for calling a global function with example usage: test_eval_set_no_early_stopping(task, output, eval_sizes, eval_names_prefix, cluster) without return types