class_docstr: Stress tests for the replay buffer, especially for new flags.
class_docstr_tok: ["Stress","tests","for","the","replay","buffer",",","especially","for","new","flags","."]
docstr: hard to make a stress test for this, since tracking which indices
  gets replaced would be effectively building a second RB
  so instead opt for simple test...
  stack_size = 2 so there's 1 padding.
docstr_tok: ["hard","to","make","a","stress","test","for","this",",","since","tracking","which","indices","gets","replaced","would","be","effectively","building","a","second","RB","so","instead","opt","for","simple","test","...","stack_size","=","2","so","there","'s","1","padding","."]
code: def test_replay_overflow(self):
      """
      hard to make a stress test for this, since tracking which indices
      gets replaced would be effectively building a second RB
      so instead opt for simple test...
      stack_size = 2 so there's 1 padding.
      """
      multi_steps = 2
      stack_size = 2
      memory = ReplayBuffer(
          stack_size=stack_size,
          replay_capacity=6,
          batch_size=1,
          update_horizon=multi_steps,
          return_everything_as_stack=None,
          return_as_timeline_format=True,
      )
  
      def trans(i):
          return {
              "observation": np.ones(OBS_SHAPE, dtype=OBS_TYPE),
              "action": int(2 * i),
              "reward": float(3 * i),
          }
  
      # Contents of RB
      # start: [X, X, X, X, X, X]
      npt.assert_array_equal(
          memory._is_index_valid,
          [False, False, False, False, False, False],
      )
  
      # t0: [X, s0, X, X, X, X]
      memory.add(**trans(0), terminal=False)
      npt.assert_array_equal(
          memory._is_index_valid,
          [False, False, False, False, False, False],
      )
  
      # t1: [X, s0, s1, X, X, X]
      memory.add(**trans(1), terminal=False)
      npt.assert_array_equal(
          memory._is_index_valid,
          [False, False, False, False, False, False],
      )
  
      # t2: [X, s0, s1, s2, X, X]
      # s0 finally becomes valid as its next state was added
      memory.add(**trans(2), terminal=False)
      npt.assert_array_equal(
          memory._is_index_valid,
          [False, True, False, False, False, False],
      )
      batch = memory.sample_all_valid_transitions()
      npt.assert_array_equal(batch.action, [[0, 0]])
      npt.assert_array_equal(batch.next_action[0], [[0, 2], [2, 4]])
  
      # t3: [X, s0, s1, s2, s3, X]
      # episode termination validates whole episode
      memory.add(**trans(3), terminal=True)
      npt.assert_array_equal(
          memory._is_index_valid, [False, True, True, True, True, False]
      )
      batch = memory.sample_all_valid_transitions()
      npt.assert_array_equal(
          batch.action, [[0, 0], [0, 2], [2, 4], [4, 6]]
      )
      npt.assert_array_equal(batch.next_action[0], [[0, 2], [2, 4]])
      npt.assert_array_equal(batch.next_action[1], [[2, 4], [4, 6]])
      # batch.next_action[2][1] is garbage
      npt.assert_array_equal(batch.next_action[2][0], [4, 6])
      # batch.next_action[3] is [garbage]
  
      # t4: [s4, s0, s1, s2, s3, X]
      # s0 invalidated as its previous frame is corrupted
      memory.add(**trans(4), terminal=False)
      npt.assert_array_equal(
          memory._is_index_valid,
          [False, False, True, True, True, False],
      )
      batch = memory.sample_all_valid_transitions()
      npt.assert_array_equal(batch.action, [[0, 2], [2, 4], [4, 6]])
      npt.assert_array_equal(batch.next_action[0], [[2, 4], [4, 6]])
      npt.assert_array_equal(batch.next_action[1][0], [4, 6])
  
      # t5: [s4, s5, s1, s2, s3, X]
      memory.add(**trans(5), terminal=False)
      npt.assert_array_equal(
          memory._is_index_valid,
          [False, False, False, True, True, False],
      )
      batch = memory.sample_all_valid_transitions()
      npt.assert_array_equal(batch.action, [[2, 4], [4, 6]])
      npt.assert_array_equal(batch.next_action[0][0], [4, 6])
  
      # t6: [s4, s5, s6, s2, s3, X]
      memory.add(**trans(6), terminal=True)
      npt.assert_array_equal(
          memory._is_index_valid, [True, True, True, False, True, False]
      )
      batch = memory.sample_all_valid_transitions()
      npt.assert_array_equal(
          batch.action, [[0, 8], [8, 10], [10, 12], [4, 6]]
      )
      npt.assert_array_equal(batch.next_action[0], [[8, 10], [10, 12]])
      npt.assert_array_equal(batch.next_action[1][0], [10, 12])
      # batch.next_action[2] is [garbage]
      # batch.next_action[3] is [garbage]
  
      logger.info("Overflow test passes!")
  
file_path: ReAgent/reagent/test/replay_memory/extra_replay_buffer_test.py
imports: import logging
  import numpy
  import numpy.testing
  import torch
  from reagent.replay_memory.circular_replay_buffer import ReplayBuffer
  from reagent.test.base.horizon_test_base import HorizonTestBase
label_desc: Use image node_id 3 for calling the ExtraReplayBufferTest obj's underlying member method code with example usage: obj.test_replay_overflow() and returns: dict