code: def test_rampup_bs_schedule(
      self, gpt_model, trainer_cfg, rampup_batch_size_schedule
  ):
      num_microbatch_calculator = (
          apex.transformer.pipeline_parallel.utils._GLOBAL_NUM_MICROBATCHES_CALCULATOR
      )
      micro_batch_size = gpt_model.cfg.micro_batch_size
      num_devices = trainer_cfg["devices"]
      num_nodes = trainer_cfg["num_nodes"]
      max_steps = trainer_cfg["max_steps"]
  
      global_batch_size_schedule = []
      step, consumed_samples = 0, 0
      while step <= max_steps:
          step += 1
          current_global_batch_size = (
              get_num_microbatches()
              * micro_batch_size
              * num_devices
              * num_nodes
          )
          consumed_samples += current_global_batch_size
          num_microbatch_calculator.update(
              consumed_samples=consumed_samples, consistency_check=True
          )
  
          if (
              current_global_batch_size
              not in global_batch_size_schedule
          ):
              global_batch_size_schedule.append(
                  current_global_batch_size
              )
  
      reset_microbatch_calculator()
  
      assert global_batch_size_schedule == rampup_batch_size_schedule
  
file_path: NeMo/tests/collections/nlp/test_rampup_batch_size.py
imports: import os
  import pytest
  import torch
  from omegaconf import DictConfig
  from pytorch_lightning import Trainer
  from nemo.collections.nlp.models.language_modeling.megatron_gpt_model import MegatronGPTModel
  from nemo.collections.nlp.parts.nlp_overrides import NLPDDPStrategy
label_desc: Use image node_id 2 for calling the TestRampupBatchSize obj's underlying member method code with example usage: obj.test_rampup_bs_schedule(gpt_model, trainer_cfg, rampup_batch_size_schedule) without return types