code: def _assert_storage_contents(
      local_inspect_dir: Path,
      exp_name: str,
      checkpoint_config: train.CheckpointConfig,
      trainable_name: str,
      test_trainer: bool,
      no_checkpoint_ranks: List[int] = None,
      constants: type = TestConstants,
  ):
      no_checkpoint_ranks = no_checkpoint_ranks or []
  
      # Second, inspect the contents of the storage path
      storage_path_ls = list(local_inspect_dir.glob("*"))
      assert len(storage_path_ls) == 1  # Only expect 1 experiment dir
      exp_dir = storage_path_ls[0]
      assert exp_dir.name == exp_name
  
      # Files synced by the driver
      assert len(list(exp_dir.glob("tuner.pkl"))) == 1
      if test_trainer:
          assert len(list(exp_dir.glob("trainer.pkl"))) == 1
      # 2 copies of these files:
      # 1 for the initial run, and 1 for the manually restored run.
      assert len(list(exp_dir.glob("basic-variant-state-*"))) == 2
      assert len(list(exp_dir.glob("experiment_state-*"))) == 2
  
      # Files synced by the worker
      assert (
          len(list(exp_dir.glob(f"{trainable_name}*"))) == 1
          if test_trainer
          else constants.NUM_TRIALS
      )
      for trial_dir in exp_dir.glob(f"{trainable_name}*"):
          # If set, expect num_to_keep. Otherwise, expect to see all of them.
          expected_num_checkpoints = (
              checkpoint_config.num_to_keep or constants.NUM_ITERATIONS
          )
  
          assert (
              len(list(trial_dir.glob("checkpoint_*")))
              == expected_num_checkpoints
          )
          checkpoint_idxs = sorted(
              [
                  _get_checkpoint_index(checkpoint_dir.name)
                  for checkpoint_dir in trial_dir.glob("checkpoint_*")
              ]
          )
          # Ex: If num_to_keep=2 out of 6 total checkpoints,
          # expect checkpoint_004 and checkpoint_005.
          assert checkpoint_idxs == list(
              range(
                  constants.NUM_ITERATIONS - expected_num_checkpoints,
                  constants.NUM_ITERATIONS,
              )
          )
  
          for checkpoint_dir in trial_dir.glob("checkpoint_*"):
              # 1 shared checkpoint.pkl file, written by the trainable / all workers.
              assert (
                  len(list(checkpoint_dir.glob("checkpoint.pkl"))) == 1
                  # NOTE: Dict checkpoint is only for the ClassTrainable.
                  or len(
                      list(
                          checkpoint_dir.glob(
                              _DICT_CHECKPOINT_FILE_NAME
                          )
                      )
                  )
                  == 1
              )
              if test_trainer:
                  # 1 checkpoint shard per worker.
                  # Unless the worker did not report a checkpoint (no_checkpoint_ranks).
                  assert {
                      _get_checkpoint_shard_rank(checkpoint_shard.name)
                      for checkpoint_shard in checkpoint_dir.glob(
                          "checkpoint_shard-*.pkl"
                      )
                  } == {
                      i
                      for i in range(constants.NUM_WORKERS)
                      if i not in no_checkpoint_ranks
                  }
  
          if test_trainer:
              expected_num_artifacts = (
                  constants.NUM_ITERATIONS * constants.NUM_WORKERS
              )
          else:
              expected_num_artifacts = constants.NUM_ITERATIONS
          assert (
              len(list(trial_dir.glob("artifact-*")))
              == expected_num_artifacts
          )
  
          # NOTE: This result file is synced by the driver.
          assert len(list(trial_dir.glob(EXPR_RESULT_FILE))) == 1
  
imports: import logging
  import os
  import pickle
  import re
  import shutil
  import tempfile
  import time
  from contextlib import contextmanager
  from pathlib import Path
  from typing import List, Optional, Tuple
  import pyarrow.fs
  import pytest
  import ray
  from ray import train, tune
  from ray._private.test_utils import simulate_storage
  from ray.air._internal.uri_utils import URI
  from ray.air.constants import EXPR_RESULT_FILE
  from ray.train._checkpoint import Checkpoint
  from ray.train._internal.storage import StorageContext, _delete_fs_path, _download_from_fs_path
  from ray.train.base_trainer import TrainingFailedError
  from ray.train.data_parallel_trainer import DataParallelTrainer
  from ray.tune.trainable.trainable import _DICT_CHECKPOINT_FILE_NAME
label_desc: Use image node_id 13 for calling a global function with example usage: _assert_storage_contents(local_inspect_dir, exp_name, checkpoint_config, trainable_name, test_trainer, no_checkpoint_ranks, constants) without return types