code: def run_storage_lifecycle_test(
      manager: storage.StorageManager,
      post_delete_cb: Optional[Callable] = None,
      caplog: Any = None,
  ) -> None:
      checkpoints = []
      for _ in range(2):
          storage_id = str(uuid.uuid4())
          with manager.store_path(storage_id) as path:
              create_checkpoint(path)
              checkpoints.append(storage_id)
  
      for storage_id in checkpoints:
          # Load checkpoint.
          with manager.restore_path(storage_id) as path:
              validate_checkpoint(path, EXPECTED_FILES)
          # Delete.
          manager.delete(storage_id, ["**/*"])
          # Ensure it is gone.
          with pytest.raises(errors.CheckpointNotFound):
              with manager.restore_path(storage_id) as path:
                  pass
  
          # Ensure second delete does not puke.
          manager.delete(storage_id, ["**/*"])
  
          # Allow for backend-specific inspection.
          if post_delete_cb is not None:
              post_delete_cb(storage_id)
  
      # Again, using upload/download instead of store_path/restore_path.
      checkpoints = []
      for _ in range(2):
          storage_id = str(uuid.uuid4())
          path = pathlib.Path(
              f"/tmp/storage_lifecycle_test-{storage_id}"
          )
          try:
              create_checkpoint(path)
              manager.upload(path, storage_id)
              checkpoints.append(storage_id)
          finally:
              shutil.rmtree(path, ignore_errors=True)
  
      for storage_id in checkpoints:
          path = pathlib.Path(
              f"/tmp/storage_lifecycle_test-{storage_id}"
          )
          try:
              manager.download(storage_id, path)
              validate_checkpoint(path, EXPECTED_FILES)
          finally:
              shutil.rmtree(path, ignore_errors=True)
          manager.delete(storage_id, ["**/*"])
          with pytest.raises(errors.CheckpointNotFound):
              manager.download(storage_id, path)
          if post_delete_cb is not None:
              post_delete_cb(storage_id)
  
      # Upload checkpoint and test restore_path/download with selector.
      checkpoints = []
      for _ in range(2):
          storage_id = str(uuid.uuid4())
          with manager.store_path(storage_id) as path:
              create_checkpoint(path)
              checkpoints.append(storage_id)
  
      expected_files_subset = {
          "subdir/": None,
          "subdir/file1.txt": "nested file 1",
          "empty_dir/": None,
      }
  
      def selector(x: str) -> bool:
          return x in ["subdir/file1.txt", "empty_dir/"]
  
      # Test restore_path with selector
      # clear logs collected up to this point
      if caplog is not None:
          caplog.clear()
      for storage_id in checkpoints:
          with manager.restore_path(
              storage_id, selector=selector
          ) as path:
              if isinstance(
                  manager, storage.shared.SharedFSStorageManager
              ):
                  assert caplog
                  assert (
                      caplog.messages[0]
                      == "Ignoring partial checkpoint download from shared_fs;"
                      " all files will be directly accessible from shared_fs."
                  )
                  validate_checkpoint(path, EXPECTED_FILES)
              else:
                  validate_checkpoint(path, expected_files_subset)
  
      # Test download with selector
      for storage_id in checkpoints:
          path = pathlib.Path(
              f"/tmp/storage_lifecycle_test-{storage_id}"
          )
          try:
              manager.download(storage_id, path, selector=selector)
              validate_checkpoint(path, expected_files_subset)
              manager.delete(storage_id, ["**/*"])
              with pytest.raises(errors.CheckpointNotFound):
                  manager.download(storage_id, path)
              if post_delete_cb is not None:
                  post_delete_cb(storage_id)
          finally:
              shutil.rmtree(path, ignore_errors=True)
  
      # Test upload with selector.
      checkpoints = []
      for _ in range(2):
          storage_id = str(uuid.uuid4())
          path = pathlib.Path(
              f"/tmp/storage_lifecycle_test-{storage_id}"
          )
          try:
              create_checkpoint(path)
              manager.upload(
                  path,
                  storage_id,
                  paths={"subdir/file1.txt", "empty_dir/"},
              )
              checkpoints.append(storage_id)
          finally:
              shutil.rmtree(path, ignore_errors=True)
  
      expected_files_subset = {
          "subdir/": None,
          "subdir/file1.txt": "nested file 1",
          "empty_dir/": None,
      }
  
      for storage_id in checkpoints:
          path = pathlib.Path(
              f"/tmp/storage_lifecycle_test-{storage_id}"
          )
          try:
              manager.download(storage_id, path)
              validate_checkpoint(path, expected_files_subset)
              manager.delete(storage_id, ["**/*"])
              with pytest.raises(errors.CheckpointNotFound):
                  manager.download(storage_id, path)
              if post_delete_cb is not None:
                  post_delete_cb(storage_id)
          finally:
              shutil.rmtree(path, ignore_errors=True)
  
      # Partial delete test.
      cases = [
          (
              ["empty_dir/*", "subdir/*"],
              {"empty_dir/": 0, "subdir/": 0, "root.txt": 9},
          ),
          (
              ["empty_dir"],
              {
                  "root.txt": 9,
                  "subdir/": 0,
                  "subdir/file1.txt": 13,
                  "subdir/file2.txt": 13,
              },
          ),
          (["subdir"], {"root.txt": 9, "empty_dir/": 0}),
          (["**/*.txt"], {"empty_dir/": 0, "subdir/": 0}),
          (
              ["**.txt"],
              {
                  "empty_dir/": 0,
                  "subdir/": 0,
                  "subdir/file1.txt": 13,
                  "subdir/file2.txt": 13,
              },
          ),
      ]
  
      for c in cases:
          storage_id = str(uuid.uuid4())
          path = pathlib.Path(
              f"/tmp/storage_lifecycle_test-{storage_id}"
          )
          create_checkpoint(path)
          try:
              manager.upload(path, storage_id)
          finally:
              shutil.rmtree(path, ignore_errors=True)
  
          assert manager.delete(storage_id, c[0]) == c[1]
  
          manager.download(storage_id, path)
          try:
              validate_checkpoint(
                  path,
                  {
                      k: v
                      for k, v in EXPECTED_FILES.items()
                      if k in c[1]
                  },
              )
          finally:
              shutil.rmtree(path, ignore_errors=True)
  
imports: import os
  import pathlib
  import shutil
  import time
  import uuid
  from typing import Any, Callable, Dict, List, Optional, Tuple
  from unittest import mock
  import pytest
  import requests
  from determined import core, errors
  from determined.common import storage
  from determined.tensorboard.fetchers import base
  from tests import parallel
label_desc: Use image node_id 4 for calling a global function with example usage: run_storage_lifecycle_test(manager, post_delete_cb, caplog) and returns: unknown