code: def run_gc_checkpoints_test(
      checkpoint_storage: Dict[str, str]
  ) -> None:
      fixtures = [
          (
              conf.fixtures_path(
                  "no_op/gc_checkpoints_decreasing.yaml"
              ),
              {
                  (bindings.experimentv1State.COMPLETED.value): {
                      800,
                      900,
                      1000,
                  },
                  (bindings.experimentv1State.DELETED.value): {
                      100,
                      200,
                      300,
                      400,
                      500,
                      600,
                      700,
                  },
              },
          ),
          (
              conf.fixtures_path(
                  "no_op/gc_checkpoints_increasing.yaml"
              ),
              {
                  (bindings.experimentv1State.COMPLETED.value): {
                      100,
                      200,
                      300,
                      900,
                      1000,
                  },
                  (bindings.experimentv1State.DELETED.value): {
                      400,
                      500,
                      600,
                      700,
                      800,
                  },
              },
          ),
      ]
  
      all_checkpoints: List[
          Tuple[Any, List[bindings.v1CheckpointWorkload]]
      ] = []
      for base_conf_path, result in fixtures:
          config = conf.load_config(str(base_conf_path))
          experiment_storage = checkpoint_storage.copy()
          experiment_storage.update(config["checkpoint_storage"])
          config["checkpoint_storage"].update(experiment_storage)
  
          with tempfile.NamedTemporaryFile() as tf:
              with open(tf.name, "w") as f:
                  util.yaml_safe_dump(config, f)
  
              experiment_id = exp.create_experiment(
                  tf.name, conf.fixtures_path("no_op")
              )
  
          exp.wait_for_experiment_state(
              experiment_id, bindings.experimentv1State.COMPLETED
          )
  
          # In some configurations, checkpoint GC will run on an auxillary machine, which may have to
          # be spun up still.  So we'll wait for it to run.
          wait_for_gc_to_finish([experiment_id])
  
          # Checkpoints are not marked as deleted until gc_checkpoint task starts.
          retries = 5
          for retry in range(retries):
              trials = exp.experiment_trials(experiment_id)
              assert len(trials) == 1
  
              cpoints = exp.workloads_with_checkpoint(
                  trials[0].workloads
              )
              sorted_checkpoints = sorted(
                  cpoints,
                  key=lambda ckp: int(ckp.totalBatches),
              )
              assert len(sorted_checkpoints) == 10
              by_state = {}  # type: Dict[str, Set[int]]
              for ckpt in sorted_checkpoints:
                  by_state.setdefault(ckpt.state.value, set()).add(
                      ckpt.totalBatches
                  )
  
              if by_state == result:
                  all_checkpoints.append((config, sorted_checkpoints))
                  break
  
              if retry + 1 == retries:
                  assert by_state == result
  
              time.sleep(1)
  
      # Check that the actual checkpoint storage reflects the
      # deletions. We want to wait for the GC containers to exit, so check
      # repeatedly with a timeout.
      max_checks = 30
      last_checkpoint_uuid = None
      for i in range(max_checks):
          time.sleep(1)
          try:
              storage_states = []
              for config, checkpoints in all_checkpoints:
                  checkpoint_config = config["checkpoint_storage"]
                  storage_manager = storage.build(
                      checkpoint_config, container_path=None
                  )
                  storage_state = {}  # type: Dict[str, Any]
                  for checkpoint in checkpoints:
                      assert checkpoint.uuid is not None
                      last_checkpoint_uuid = (
                          storage_id
                      ) = checkpoint.uuid
                      storage_state[storage_id] = {}
                      if (
                          checkpoint.state
                          == bindings.checkpointv1State.COMPLETED
                      ):
                          storage_state[storage_id]["found"] = False
                          try:
                              with storage_manager.restore_path(
                                  storage_id
                              ):
                                  storage_state[storage_id][
                                      "found"
                                  ] = True
                          except errors.CheckpointNotFound:
                              pass
                      elif (
                          checkpoint.state
                          == bindings.checkpointv1State.DELETED
                      ):
                          storage_state[storage_id] = {
                              "deleted": False,
                              "checkpoint": checkpoint,
                          }
                          try:
                              with storage_manager.restore_path(
                                  storage_id
                              ):
                                  pass
                          except errors.CheckpointNotFound:
                              storage_state[storage_id][
                                  "deleted"
                              ] = True
                          storage_states.append(storage_state)
  
              for storage_state in storage_states:
                  for state in storage_state.values():
                      if state.get("deleted", None) is False:
                          json_states = json.dumps(storage_states)
                          raise AssertionError(
                              f"Some checkpoints were not deleted: JSON:{json_states}"
                          )
                      if state.get("found", None) is False:
                          json_states = json.dumps(storage_states)
                          raise AssertionError(
                              f"Some checkpoints were not found: JSON:{json_states}"
                          )
          except AssertionError:
              if i == max_checks - 1:
                  raise
          else:
              break
  
      cs_type = checkpoint_storage["type"]
      if cs_type != "azure":
          assert type(last_checkpoint_uuid) == str
          run_command_master_checkpoint_download(
              str(last_checkpoint_uuid)
          )
  
imports: import json
  import os
  import random
  import sys
  import tempfile
  import time
  from typing import Any, Dict, List, Set, Tuple
  import pexpect
  import pytest
  from determined import errors
  from determined.common import api, storage, util
  from determined.common.api import authentication, bindings, certs
  from determined.common.api.bindings import checkpointv1State
  from tests import api_utils
  from tests import config
  from tests import experiment
  from .test_users import det_spawn
label_desc: Use image node_id 9 for calling a global function with example usage: run_gc_checkpoints_test(checkpoint_storage) without return types