docstr: Test adding to segmentation buffer.
docstr_tok: ["Test","adding","to","segmentation","buffer","."]
code: def test_add(self):
      """Test adding to segmentation buffer."""
      for buffer_cls in [
          SegmentationBuffer,
          MultiAgentSegmentationBuffer,
      ]:
          max_seq_len = 3
          max_ep_len = 10
          capacity = 1
          buffer = buffer_cls(capacity, max_seq_len, max_ep_len)
  
          # generate batch
          episode_batches = []
          for i in range(4):
              episode_batches.append(
                  _generate_episode_batch(max_ep_len, i)
              )
          batch = concat_samples(episode_batches)
  
          # add to buffer and check that only last one is kept (due to replacement)
          buffer.add(batch)
  
          self.assertEqual(
              len(_get_internal_buffer(buffer)),
              1,
              "The internal buffer should only contain one SampleBatch since"
              " the capacity is 1.",
          )
          test_utils.check(
              episode_batches[-1], _get_internal_buffer(buffer)[0]
          )
  
          # add again
          buffer.add(episode_batches[0])
  
          test_utils.check(
              episode_batches[0], _get_internal_buffer(buffer)[0]
          )
  
          # make buffer of enough capacity
          capacity = len(episode_batches)
          buffer = buffer_cls(capacity, max_seq_len, max_ep_len)
  
          # add to buffer and make sure all are in
          buffer.add(batch)
          self.assertEqual(
              len(_get_internal_buffer(buffer)),
              len(episode_batches),
              "internal buffer doesn't have the right number of episodes.",
          )
          for i in range(len(episode_batches)):
              test_utils.check(
                  episode_batches[i], _get_internal_buffer(buffer)[i]
              )
  
          # add another one and make sure it replaced one of them
          new_batch = _generate_episode_batch(max_ep_len, 12345)
          buffer.add(new_batch)
          self.assertEqual(
              len(_get_internal_buffer(buffer)),
              len(episode_batches),
              "internal buffer doesn't have the right number of episodes.",
          )
          found = False
          for episode_batch in _get_internal_buffer(buffer):
              if episode_batch[SampleBatch.EPS_ID][0] == 12345:
                  test_utils.check(episode_batch, new_batch)
                  found = True
                  break
          assert found, "new_batch not added to buffer."
  
          # test that adding too long an episode errors
          long_batch = _generate_episode_batch(max_ep_len + 1, 123)
          with self.assertRaises(ValueError):
              buffer.add(long_batch)
  
imports: import unittest
  from typing import List, Union
  import numpy
  from rllib_dt.dt.segmentation_buffer import MultiAgentSegmentationBuffer, SegmentationBuffer
  import ray
  from ray.rllib.policy.sample_batch import DEFAULT_POLICY_ID, MultiAgentBatch, SampleBatch, concat_samples
  from ray.rllib.utils import test_utils
  from ray.rllib.utils.framework import try_import_tf, try_import_torch
  from ray.rllib.utils.typing import PolicyID
label_desc: Use image node_id 3 for calling the TestSegmentationBuffer obj's underlying member method code with example usage: obj.test_add() without return types