docstr: Tests batches getting properly added to buffer and cause proper eviction.
docstr_tok: ["Tests","batches","getting","properly","added","to","buffer","and","cause","proper","eviction","."]
code: def test_add_and_eviction_logic(self):
      """Tests batches getting properly added to buffer and cause proper eviction."""
  
      # Fill a buffer till capacity (100 ts).
      buffer = EpisodeReplayBuffer(capacity=100)
  
      episode = self._get_episode(id_="A", episode_len=50)
      buffer.add(episode)
      self.assertTrue(buffer.get_num_episodes() == 1)
      self.assertTrue(buffer.get_num_timesteps() == 50)
  
      episode = self._get_episode(id_="B", episode_len=25)
      buffer.add(episode)
      self.assertTrue(buffer.get_num_episodes() == 2)
      self.assertTrue(buffer.get_num_timesteps() == 75)
  
      # No eviction yet (but we are full).
      episode = self._get_episode(id_="C", episode_len=25)
      buffer.add(episode)
      self.assertTrue(buffer.get_num_episodes() == 3)
      self.assertTrue(buffer.get_num_timesteps() == 100)
  
      # Trigger eviction of first episode by adding a single timestep episode.
      episode = self._get_episode(id_="D", episode_len=1)
      buffer.add(episode)
  
      self.assertTrue(buffer.get_num_episodes() == 3)
      self.assertTrue(buffer.get_num_timesteps() == 51)
      self.assertTrue(
          {eps.id_ for eps in buffer.episodes} == {"B", "C", "D"}
      )
  
      # Add another big episode and trigger another eviction.
      episode = self._get_episode(id_="E", episode_len=200)
      buffer.add(episode)
      self.assertTrue(buffer.get_num_episodes() == 1)
      self.assertTrue(buffer.get_num_timesteps() == 200)
      self.assertTrue({eps.id_ for eps in buffer.episodes} == {"E"})
  
      # Add another small episode and trigger another eviction.
      episode = self._get_episode(id_="F", episode_len=2)
      buffer.add(episode)
      self.assertTrue(buffer.get_num_episodes() == 1)
      self.assertTrue(buffer.get_num_timesteps() == 2)
      self.assertTrue({eps.id_ for eps in buffer.episodes} == {"F"})
  
      # Add N small episodes.
      for i in range(10):
          episode = self._get_episode(id_=str(i), episode_len=10)
          buffer.add(episode)
      self.assertTrue(buffer.get_num_episodes() == 10)
      self.assertTrue(buffer.get_num_timesteps() == 100)
  
      # Add a 20-ts episode and expect to have evicted 3 episodes.
      episode = self._get_episode(id_="G", episode_len=21)
      buffer.add(episode)
      self.assertTrue(buffer.get_num_episodes() == 8)
      self.assertTrue(buffer.get_num_timesteps() == 91)
      self.assertTrue(
          {eps.id_ for eps in buffer.episodes}
          == {"3", "4", "5", "6", "7", "8", "9", "G"}
      )
  
file_path: ray/rllib/utils/replay_buffers/tests/test_episode_replay_buffer.py
imports: import unittest
  import numpy
  from ray.rllib.env.single_agent_episode import SingleAgentEpisode
  from ray.rllib.utils.replay_buffers.episode_replay_buffer import EpisodeReplayBuffer
label_desc: Use image node_id 2 for calling the TestEpisodeReplayBuffer obj's underlying member method code with example usage: obj.test_add_and_eviction_logic() without return types