code: def test_durable_upload(bucket: str):
      """
      Sync trial and experiment checkpoints to cloud, so:
  
          storage_path="s3://"
  
      Expected results after first checkpoint:
  
          - 4 trials are running
          - At least one trial ran on the head node
          - At least one trial ran remotely
          - Driver has NO trial checkpoints from head node trial
            (since they're uploaded directly to storage instead)
          - Driver has trial artifacts from head node trial
          - Driver has NO trial checkpoints from remote node trials
          - Driver has NO trial artifacts from remote node trials
          - Remote trial dirs only have data for one trial
          - Remote trial dirs have NO checkpoints for node-local trials
            (since they're uploaded directly to storage instead)
          - Remote trial dirs have trial artifacts for node-local trials
          - Cloud checkpoint is valid
          - Cloud checkpoint has checkpoints from ALL trials
          - Cloud checkpoint has artifacts from ALL trials (NOT IMPLEMENTED)
  
      Then, remote checkpoint directories are cleaned up.
  
      Expected results after second checkpoint:
  
          - 4 trials are running
          - All trials progressed with training
          - Cloud checkpoint is valid
          - Cloud checkpoint has checkpoints from all trials
          - Cloud checkpoint has updated synced artifacts for all trials (NOT IMPLEMENTED)
  
      """
      if not bucket:
          raise ValueError(
              "The `durable_upload` test requires a `--bucket` argument to be set."
          )
  
      experiment_name = "cloud_durable_upload"
      indicator_file = f"/tmp/{experiment_name}_indicator"
  
      def before_experiments():
          clear_bucket_contents(bucket)
  
      def between_experiments():
          (
              experiment_state,
              driver_dir_cp,
              trial_exp_checkpoint_data,
          ) = get_experiment_and_trial_data(
              experiment_name=experiment_name
          )
  
          # Req: 4 trials are running
          assert all(
              trial.status == "RUNNING"
              for trial in experiment_state.trials
          ), "Not all trials are RUNNING"
  
          # Req: At least one trial ran on driver
          # Req: At least one trial ran remotely
          assert_min_num_trials(
              driver_dir_cp.trial_to_cps.keys(),
              on_driver=1,
              on_worker=1,
          )
  
          # Req: Driver has NO trial checkpoints from head node trial
          # Req: Driver has NO trial checkpoints from remote node trials
          assert_checkpoint_count(
              driver_dir_cp,
              for_driver_trial=0,
              for_worker_trial=0,
              max_additional=0,
          )
  
          # Req: Driver has trial artifacts from head node trial
          # Req: Driver has no trial artifacts from remote node trials
          # assert_artifact_existence_and_validity(
          #     driver_dir_cp,
          #     exists_for_driver_trials=True,
          #     exists_for_worker_trials=False,
          # )
  
          for trial, exp_dir_cp in trial_exp_checkpoint_data.items():
              # Req: Remote trial dirs only have data for one trial
  
              seen = len(exp_dir_cp.trial_to_cps)
  
              if trial.was_on_driver_node:
                  assert seen == 4, (
                      f"Trial {trial.trial_id} was on driver, "
                      f"but observed too few trials ({seen}) "
                      f"in experiment dir."
                  )
              else:
                  assert seen == 1, (
                      f"Trial {trial.trial_id} was not on driver, "
                      f"but observed not exactly 1 trials ({seen}) "
                      f"in experiment dir."
                  )
  
                  assert_checkpoint_count(
                      exp_dir_cp,
                      for_driver_trial=0,
                      for_worker_trial=0,
                      max_additional=0,
                  )
  
                  # Req: Remote trial dirs have artifacts for node-local trials
                  # assert_artifact_existence_and_validity(
                  #     exp_dir_cp,
                  #     exists_for_driver_trials=False,
                  #     exists_for_worker_trials=True,
                  # )
  
          bucket_state_cp, bucket_dir_cp = get_bucket_data(
              bucket, experiment_name
          )
  
          # Req: Cloud checkpoint is valid
          assert_experiment_checkpoint_validity(bucket_state_cp)
  
          # Req: Cloud checkpoint has checkpoints from all trials
          assert_checkpoint_count(
              bucket_dir_cp,
              for_driver_trial=2,
              for_worker_trial=2,
              max_additional=2,
          )
  
          # Req: Cloud checkpoint has artifacts from all trials
          # assert_artifact_existence_and_validity(
          #     bucket_dir_cp,
          #     exists_for_driver_trials=True,
          #     exists_for_worker_trials=True,
          # )
  
          # Delete remote checkpoints before resume
          print("Deleting remote checkpoints before resume")
          cleanup_remote_node_experiment_dir(experiment_name)
  
      def after_experiments():
          (experiment_state, _, _) = get_experiment_and_trial_data(
              experiment_name=experiment_name
          )
  
          # Req: 4 trials are running
          assert all(
              trial.status == "RUNNING"
              for trial in experiment_state.trials
          ), "Not all trials are RUNNING"
  
          for trial in experiment_state.trials:
              assert_trial_progressed_training(trial)
  
          bucket_state_cp, bucket_dir_cp = get_bucket_data(
              bucket, experiment_name
          )
  
          # Req: Cloud checkpoint is valid
          assert_experiment_checkpoint_validity(bucket_state_cp)
  
          # Req: Cloud checkpoint has checkpoints from all trials
          assert_checkpoint_count(
              bucket_dir_cp,
              for_driver_trial=2,
              for_worker_trial=2,
              max_additional=2,
          )
  
          # assert_artifact_existence_and_validity(
          #     bucket_dir_cp,
          #     exists_for_driver_trials=True,
          #     exists_for_worker_trials=True,
          #     # TODO(ml-team): Set this flag to True after restoration w/
          #     # artifacts is supported.
          #     skip_validation=True,
          # )
  
          # clear_bucket_contents(bucket)
  
      run_time = int(os.getenv("TUNE_RUN_TIME", "180")) or 180
  
      run_start_timeout = (
          600 if "rllib" in os.environ["TUNE_TRAINABLE"] else 30
      )
  
      run_resume_flow(
          experiment_name=experiment_name,
          indicator_file=indicator_file,
          no_syncer=False,
          storage_path=bucket,
          first_run_time=run_time,
          second_run_time=run_time,
          run_start_timeout=run_start_timeout,
          before_experiments_callback=before_experiments,
          between_experiments_callback=between_experiments,
          after_experiments_callback=after_experiments,
      )
  
imports: import argparse
  import csv
  import io
  import tarfile
  from dataclasses import dataclass
  import json
  import os
  import pickle
  import platform
  import re
  import shutil
  import signal
  import subprocess
  import tempfile
  import time
  from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple
  import ray
  from ray.train._internal.storage import StorageContext
  from ray.tune.execution.experiment_state import _find_newest_experiment_checkpoint
  from ray.tune.result import _get_defaults_results_dir
  from ray.tune.utils.serialization import TuneFunctionDecoder
label_desc: Use image node_id 29 for calling a global function with example usage: test_durable_upload(bucket) without return types