code: def test_checkpoint_apis(client: _client.Determined) -> None:
      with open(
          conf.fixtures_path("no_op/single-default-ckpt.yaml")
      ) as f:
          config = util.yaml_safe_load(f)
  
      # Test for 100 batches/checkpoint every 10 = 10 checkpoints.
      config["min_checkpoint_period"]["batches"] = 10
      config["min_validation_period"]["batches"] = 10
      config["checkpoint_storage"] = {}
      config["checkpoint_storage"]["save_trial_best"] = 10
  
      exp = client.create_experiment(
          config, conf.fixtures_path("no_op")
      )
  
      # Await first trial is safe to call before a trial has started.
      trial = exp.await_first_trial()
  
      assert exp.wait() == _client.ExperimentState.COMPLETED
      trials = exp.get_trials()
      assert len(trials) == 1, trials
  
      checkpoints = trial.get_checkpoints()
      assert len(checkpoints) == 10
  
      # Validate end (report) time sorting.
      checkpoints = trial.get_checkpoints(
          sort_by=_client.CheckpointSortBy.END_TIME,
          order_by=_client.OrderBy.DESC,
      )
      end_times = [checkpoint.report_time for checkpoint in checkpoints]
      assert all(x >= y for x, y in zip(end_times, end_times[1:]))  # type: ignore
  
      # Validate state sorting.
      checkpoints = trial.get_checkpoints(
          sort_by=_client.CheckpointSortBy.STATE,
          order_by=_client.OrderBy.ASC,
      )
      states = []
      for checkpoint in checkpoints:
          assert checkpoint.state
          states.append(checkpoint.state.value)
      assert all(x <= y for x, y in zip(states, states[1:]))
  
      # Validate UUID sorting.
      checkpoints = trial.get_checkpoints(
          sort_by=_client.CheckpointSortBy.UUID,
          order_by=_client.OrderBy.ASC,
      )
      uuids = [checkpoint.uuid for checkpoint in checkpoints]
      assert all(x <= y for x, y in zip(uuids, uuids[1:]))
  
      # Validate batch number sorting.
      checkpoints = trial.get_checkpoints(
          sort_by=_client.CheckpointSortBy.BATCH_NUMBER,
          order_by=_client.OrderBy.DESC,
      )
      batch_numbers = []
      for checkpoint in checkpoints:
          assert checkpoint.metadata
          batch_numbers.append(checkpoint.metadata["steps_completed"])
      assert all(
          x >= y for x, y in zip(batch_numbers, batch_numbers[1:])
      )
  
      # Validate metric sorting.
      checkpoints = trial.get_checkpoints(
          sort_by="validation_error", order_by=_client.OrderBy.ASC
      )
      validation_metrics = [
          checkpoint.training.validation_metrics["avgMetrics"]["validation_error"]  # type: ignore
          for checkpoint in checkpoints
      ]
      assert all(
          x <= y
          for x, y in zip(validation_metrics, validation_metrics[1:])
      )
  
      # Expect 10 completed checkpoints.
      checkpoints = [
          checkpoint
          for checkpoint in checkpoints
          if checkpoint.state == _client.CheckpointState.COMPLETED
      ]
      assert len(checkpoints) == 10
  
      # Delete first checkpoint.
      deleted_checkpoint = checkpoints[0]
      checkpoints[0].delete()
  
      # Wait for status to be DELETED.
      start = time.time()
      deadline = start + 30
      while True:
          checkpoints = trial.get_checkpoints()
          deleted_checkpoints = [
              checkpoint
              for checkpoint in checkpoints
              if checkpoint.state == _client.CheckpointState.DELETED
          ]
          if deleted_checkpoints:
              break
          assert (
              time.time() < deadline
          ), "experiment took too long to start trials"
          time.sleep(0.1)
  
      assert len(deleted_checkpoints) == 1
      assert deleted_checkpoints[0].uuid == deleted_checkpoint.uuid
  
      # Partially delete first checkpoint.
      partially_deleted_checkpoint = checkpoints[1]
      partially_deleted_checkpoint.remove_files(["*.pkl"])
  
      # Wait for status to be PARTIALLY_DELETED
      partially_deleted_checkpoints = []
      start = time.time()
      deadline = start + 30
      while True:
          checkpoints = trial.get_checkpoints()
          partially_deleted_checkpoints = [
              checkpoint
              for checkpoint in checkpoints
              if checkpoint.state
              == _client.CheckpointState.PARTIALLY_DELETED
          ]
          if partially_deleted_checkpoints:
              break
          assert (
              time.time() < deadline
          ), "checkpoint took too long to partially delete"
          time.sleep(0.1)
      assert len(partially_deleted_checkpoints) == 1
      assert (
          partially_deleted_checkpoints[0].uuid
          == partially_deleted_checkpoint.uuid
      )
      assert partially_deleted_checkpoints[0].resources
      assert (
          "workload_sequencer.pkl"
          not in partially_deleted_checkpoints[0].resources
      )
  
      # Ensure we can download the partially deleted checkpoint.
      temp_dir = tempfile.mkdtemp()
      try:
          downloaded_path = partially_deleted_checkpoints[0].download(
              path=os.path.join(temp_dir, "c")
          )
          files = os.listdir(downloaded_path)
          assert "no_op_checkpoint" in files
          assert "workload_sequencer.pkl" not in files
      finally:
          shutil.rmtree(temp_dir, ignore_errors=False)
  
      # Ensure we can delete a partially deleted checkpoint.
      partially_deleted_checkpoints[0].delete()
      start = time.time()
      deadline = start + 30
      while True:
          checkpoints = trial.get_checkpoints()
          deleted_checkpoints = [
              checkpoint
              for checkpoint in checkpoints
              if checkpoint.state == _client.CheckpointState.DELETED
              and checkpoint.uuid == partially_deleted_checkpoint.uuid
          ]
          if deleted_checkpoints:
              break
          assert (
              time.time() < deadline
          ), "partially deleted checkpoint took too long to delete"
          time.sleep(0.1)
  
file_path: determined/e2e_tests/tests/test_sdk.py
imports: import os
  import random
  import shutil
  import tempfile
  import time
  import pytest
  from determined.common import util
  from determined.common.api import bindings, errors
  from determined.common.experimental import resource_pool
  from determined.common.experimental.metrics import TrialMetrics
  from determined.experimental import client
  from tests import config
label_desc: Use image node_id 2 for calling a global function with example usage: test_checkpoint_apis(client) without return types