code: def test_BatchLifecycle(self) -> None:
      # Check that state of trial statuses mapping on experiment: there should only be
      # one index, 0, among the `CANDIDATE` trials.
      trial_idcs_by_status = iter(
          self.experiment.trial_indices_by_status.values()
      )
      self.assertEqual(
          next(trial_idcs_by_status), {0}
      )  # `CANDIDATE` trial indices
      # ALl other trial statuses should not yet have trials carry them.
      self.assertTrue(
          all(len(idcs) == 0 for idcs in trial_idcs_by_status)
      )
      staging_mock = PropertyMock()
      with patch.object(
          SyntheticRunner, "staging_required", staging_mock
      ):
          mock_runner = SyntheticRunner()
          staging_mock.return_value = True
          self.batch.runner = mock_runner
          self.batch.run()
          self.assertEqual(self.batch.status, TrialStatus.STAGED)
          # Check that the trial statuses mapping on experiment has been updated.
          self.assertEqual(
              self.experiment.trial_indices_by_status[
                  TrialStatus.STAGED
              ],
              {0},
          )
          self.assertTrue(
              # pyre-fixme[6]: For 1st param expected `Iterable[object]` but got
              #  `bool`.
              all(len(idcs) == 0)
              for status, idcs in self.experiment.trial_indices_by_status.items()
              if status != TrialStatus.STAGED
          )
          self.assertIsNotNone(self.batch.time_staged)
          self.assertTrue(self.batch.status.is_deployed)
          self.assertFalse(self.batch.status.expecting_data)
  
          # Cannot change arms or runner once run
          with self.assertRaises(ValueError):
              self.batch.add_arms_and_weights(
                  arms=self.arms, weights=self.weights
              )
  
          with self.assertRaises(ValueError):
              self.batch.runner = None
  
          # Cannot run batch that was already run
          with self.assertRaises(ValueError):
              self.batch.run()
  
          self.batch.mark_running()
          self.assertEqual(self.batch.status, TrialStatus.RUNNING)
          # Check that the trial statuses mapping on experiment has been updated.
          self.assertEqual(
              self.experiment.trial_indices_by_status[
                  TrialStatus.RUNNING
              ],
              {0},
          )
          self.assertTrue(
              # pyre-fixme[6]: For 1st param expected `Iterable[object]` but got
              #  `bool`.
              all(len(idcs) == 0)
              for status, idcs in self.experiment.trial_indices_by_status.items()
              if status != TrialStatus.RUNNING
          )
          self.assertIsNotNone(self.batch.time_run_started)
          self.assertTrue(self.batch.status.expecting_data)
  
          self.batch.complete()
          # Cannot complete that which is already completed
          with self.assertRaises(ValueError):
              self.batch.complete()
  
          # Verify trial is completed
          self.assertEqual(self.batch.status, TrialStatus.COMPLETED)
          # Check that the trial statuses mapping on experiment has been updated.
          self.assertEqual(
              self.experiment.trial_indices_by_status[
                  TrialStatus.COMPLETED
              ],
              {0},
          )
          self.assertTrue(
              # pyre-fixme[6]: For 1st param expected `Iterable[object]` but got
              #  `bool`.
              all(len(idcs) == 0)
              for status, idcs in self.experiment.trial_indices_by_status.items()
              if status != TrialStatus.COMPLETED
          )
          self.assertIsNotNone(self.batch.time_completed)
          self.assertTrue(self.batch.status.is_terminal)
  
          # Cannot change status after BatchTrial is completed
          with self.assertRaises(ValueError):
              self.batch.mark_staged()
  
          with self.assertRaises(ValueError):
              self.batch.mark_completed()
  
          with self.assertRaises(ValueError):
              self.batch.mark_running()
  
          with self.assertRaises(ValueError):
              self.batch.mark_abandoned()
  
          with self.assertRaises(ValueError):
              self.batch.mark_failed()
  
          # Check that the trial statuses mapping on experiment is updated when
          # trial status is set hackily / directly, without using `mark_X`.
          self.batch._status = TrialStatus.CANDIDATE
          self.assertEqual(
              self.experiment.trial_indices_by_status[
                  TrialStatus.CANDIDATE
              ],
              {0},
          )
          self.assertTrue(
              # pyre-fixme[6]: For 1st param expected `Iterable[object]` but got
              #  `bool`.
              all(len(idcs) == 0)
              for status, idcs in self.experiment.trial_indices_by_status.items()
              if status != TrialStatus.CANDIDATE
          )
  
imports: import math
  from time import sleep
  from unittest import mock
  from unittest.mock import patch, PropertyMock
  import numpy
  from ax.core.arm import Arm
  from ax.core.base_trial import TrialStatus
  from ax.core.batch_trial import BatchTrial, GeneratorRunStruct
  from ax.core.experiment import Experiment
  from ax.core.generator_run import GeneratorRun, GeneratorRunType
  from ax.core.parameter import FixedParameter, ParameterType
  from ax.core.search_space import SearchSpace
  from ax.runners.synthetic import SyntheticRunner
  from ax.utils.common.testutils import TestCase
  from ax.utils.common.typeutils import checked_cast
  from ax.utils.testing.core_stubs import get_abandoned_arm, get_arm, get_arm_weights1, get_arms, get_experiment, get_generator_run, get_generator_run2, get_weights
label_desc: Use image node_id 13 for calling the BatchTrialTest obj's underlying member method code with example usage: obj.test_BatchLifecycle() without return types