code: def test_suite_clean_cv_predictions():
      kcvp = "keep_cross_validation_predictions"
      nfolds = 5
  
      def setup_and_train(param_enabled=None):
          h2o.remove_all()
          ds = import_dataset()
          state = (
              "enabled"
              if param_enabled is True
              else "disabled"
              if param_enabled is False
              else "default"
          )
          if param_enabled is None:
              aml = H2OAutoML(
                  project_name="keep_cross_validation_predictions_"
                  + state,
                  nfolds=nfolds,
                  max_models=3,
                  seed=1,
              )
          else:
              aml = H2OAutoML(
                  project_name="keep_cross_validation_predictions_"
                  + state,
                  nfolds=nfolds,
                  max_models=8,
                  seed=1,
                  keep_cross_validation_predictions=param_enabled,
              )
  
          aml.train(y=ds.target, training_frame=ds.train)
          # print(aml.leaderboard)
          return aml
  
      def assert_cv_predictions_on_model(model_name, present=True):
          model = h2o.get_model(model_name)
          cv_predictions = model.cross_validation_predictions()
          holdout_predictions = (
              model.cross_validation_holdout_predictions()
          )
  
          # see last comment line below for ideal assertion if cv predictions could be returned as null,
          #  but this can't be done in a clean way for autoML right now due to StackedEnsemble
          # assert not h2o.get_model(m).cross_validation_predictions(), "unexpected cv predictions for model "+m
          for p in cv_predictions:
              if present:
                  assert p is not None, (
                      "missing cv predictions for model " + model_name
                  )
              else:
                  assert not p, (
                      "unexpected cv predictions for model "
                      + model_name
                  )
  
          if present:
              assert holdout_predictions is not None, (
                  "missing holdout predictions for model " + model_name
              )
          else:
              assert not holdout_predictions, (
                  "unexpected holdout predictions for model "
                  + model_name
              )
  
      def test_default_behaviour():
          print("\n=== " + kcvp + " default behaviour ===")
          aml = setup_and_train()
          models = get_partitioned_model_names(aml.leaderboard)
          keys = list_keys_in_memory()
          preds = len(keys["cv_predictions"])
          assert (
              preds == 0
          ), "{preds} CV predictions were not cleaned from memory".format(
              preds=preds
          )
          for m in models.base:
              assert_cv_predictions_on_model(m, False)
          for m in models.se:
              assert not h2o.get_model(
                  h2o.get_model(m).metalearner().model_id
              ).cross_validation_predictions()
  
      def test_param_enabled():
          print("\n=== enabling " + kcvp + " ===")
          aml = setup_and_train(True)
          models = get_partitioned_model_names(aml.leaderboard)
          keys = list_keys_in_memory()
          preds = len(keys["cv_predictions"])
          expected = len(models.all) * (
              nfolds + 1
          )  # +1 for holdout prediction
          assert (
              preds == expected
          ), "missing CV predictions in memory, got {actual}, expected {expected}".format(
              actual=preds, expected=expected
          )
          for m in models.base:
              assert_cv_predictions_on_model(m)
          for m in models.se:
              assert_cv_predictions_on_model(
                  h2o.get_model(m).metalearner().model_id
              )
  
      def test_param_disabled():
          print("\n=== disabling " + kcvp + " ===")
          aml = setup_and_train(False)
          models = get_partitioned_model_names(aml.leaderboard)
          keys = list_keys_in_memory()
          preds = len(keys["cv_predictions"])
          assert (
              preds == 0
          ), "{preds} CV predictions were not cleaned from memory".format(
              preds=preds
          )
          for m in models.base:
              assert_cv_predictions_on_model(m, False)
          for m in models.se:
              assert not h2o.get_model(
                  h2o.get_model(m).metalearner().model_id
              ).cross_validation_predictions()
  
      def test_SE_retraining_fails_when_param_disabled():
          print("\n=== disabling " + kcvp + " and retraining ===")
          total_runs = 4
          aml = setup_and_train(False)  # first run
          first_models = get_partitioned_model_names(aml.leaderboard)
          first_bof = next(
              m
              for m in first_models.se
              if re.search(r"_BestOfFamily_", m)
          )
          ds = import_dataset()
          for i in range(total_runs - 1):
              aml.train(y=ds.target, training_frame=ds.train)
          models = get_partitioned_model_names(aml.leaderboard)
          first_se_all_models = [
              m for m in first_models.se if re.search(r"_AllModels_", m)
          ]
          se_all_models = [
              m for m in models.se if re.search(r"_AllModels_", m)
          ]
          se_best_of_family = [
              m for m in models.se if re.search(r"_BestOfFamily_", m)
          ]
          lb = aml.leaderboard
          print(lb.head(lb.nrows))
  
          assert len(models.se) == len(se_all_models) + len(
              se_best_of_family
          )
          assert len(se_all_models) == len(
              first_se_all_models
          ), "expecting only the {} first StackedEnsemble_AllModels, but got {}".format(
              len(first_se_all_models), len(se_all_models)
          )
          assert (
              se_all_models[0] in first_models.se
          ), "first StackedEnsemble_AllModels got replaced by new one"
          if len(se_best_of_family) > 1:
              assert (
                  first_bof in se_best_of_family
              ), "first StackedEnsemble_BestOfFamily disappeared after multiple runs"
              row_of = lambda id: lb[lb["model_id"] == id]
              first_bof_row = row_of(first_bof)
              assert all(
                  all(
                      row[i] == first_bof_row[i]
                      for i in range(1, lb.ncols)
                  )
                  for row in [row_of(se) for se in se_best_of_family]
              ), "expecting possibly 2+ similar StackedEnsemble_BestOfFamily (corner case), but managed to obtain 2 different ones!"
          else:
              assert (
                  len(se_best_of_family) == 1
              ), "expecting only the first StackedEnsemble_BestOfFamily, but got {}".format(
                  len(se_best_of_family)
              )
              assert (
                  se_best_of_family[0] == first_bof
              ), "first StackedEnsemble_Best_of_Family got replaced by new one"
  
      def test_SE_retraining_works_when_param_enabled():
          print("\n=== enabling " + kcvp + " and retraining ===")
          total_runs = 4
          aml = setup_and_train(True)  # first run
          ds = import_dataset()
          for i in range(total_runs - 1):
              aml.train(y=ds.target, training_frame=ds.train)
          models = get_partitioned_model_names(aml.leaderboard)
          se_all_models = [
              m for m in models.se if re.search(r"_AllModels_", m)
          ]
          se_best_of_family = [
              m for m in models.se if re.search(r"_BestOfFamily_", m)
          ]
          assert len(models.se) == len(se_all_models) + len(
              se_best_of_family
          )
          assert (
              len(se_best_of_family) + len(se_all_models) >= total_runs
          ), "some StackedEnsembles are missing"
  
      return [
          test_default_behaviour,
          test_param_enabled,
          test_param_disabled,
          test_SE_retraining_fails_when_param_disabled,
          test_SE_retraining_works_when_param_enabled,
      ]
  
file_path: h2o-3/h2o-py/tests/testdir_algos/automl/pyunit_automl_memory_cleanup.py
imports: import sys, os
  import re
  import h2o
  from h2o import H2OFrame
  from h2o.automl import H2OAutoML
  from tests import pyunit_utils
  from _automl_utils import import_dataset, get_partitioned_model_names
label_desc: Use image node_id 3 for calling a global function with example usage: test_suite_clean_cv_predictions() and returns: list, aml