class_docstr: Tests insertion and (weighted) sampling of the PrioritizedReplayBuffer.
class_docstr_tok: ["Tests","insertion","and","(","weighted",")","sampling","of","the","PrioritizedReplayBuffer","."]
code: def test_add(self):
      buffer = PrioritizedReplayBuffer(capacity=2, alpha=self.alpha)
  
      # Assert indices 0 before insert.
      self.assertEqual(len(buffer), 0)
      self.assertEqual(buffer._next_idx, 0)
  
      # Insert single record.
      data = self._generate_data()
      buffer.add(data, weight=0.5)
      self.assertTrue(len(buffer) == 1)
      self.assertTrue(buffer._next_idx == 1)
  
      # Insert single record.
      data = self._generate_data()
      buffer.add(data, weight=0.1)
      self.assertTrue(len(buffer) == 2)
      self.assertTrue(buffer._next_idx == 0)
  
      # Insert over capacity.
      data = self._generate_data()
      buffer.add(data, weight=1.0)
      self.assertTrue(len(buffer) == 2)
      self.assertTrue(buffer._next_idx == 1)
  
      # Test get_state/set_state.
      state = buffer.get_state()
      new_memory = PrioritizedReplayBuffer(capacity=2, alpha=self.alpha)
      new_memory.set_state(state)
      self.assertTrue(len(new_memory) == 2)
      self.assertTrue(new_memory._next_idx == 1)
  
file_path: ray/python/ray/rllib/utils/replay_buffers/tests/test_prioritized_replay_buffer_replay_buffer_api.py
imports: from collections import Counter
  import numpy
  import unittest
  from ray.rllib.utils.replay_buffers.prioritized_replay_buffer import PrioritizedReplayBuffer
  from ray.rllib.policy.sample_batch import SampleBatch, MultiAgentBatch, concat_samples
  from ray.rllib.utils.test_utils import check
label_desc: Use image node_id 4 for calling the TestPrioritizedReplayBuffer obj's underlying member method code with example usage: obj.test_add() without return types