docstr: Tests whether batches are correctly formed when sampling from the buffer.
docstr_tok: ["Tests","whether","batches","are","correctly","formed","when","sampling","from","the","buffer","."]
code: def test_episode_replay_buffer_sample_logic(self):
      """Tests whether batches are correctly formed when sampling from the buffer."""
      buffer = EpisodeReplayBuffer(capacity=10000)
  
      for _ in range(200):
          episode = self._get_episode()
          buffer.add(episode)
  
      for _ in range(1000):
          sample = buffer.sample(batch_size_B=16, batch_length_T=64)
          (
              obs,
              actions,
              rewards,
              is_first,
              is_last,
              is_terminated,
              is_truncated,
          ) = (
              sample["obs"],
              sample["actions"],
              sample["rewards"],
              sample["is_first"],
              sample["is_last"],
              sample["is_terminated"],
              sample["is_truncated"],
          )
          # Make sure terminated and truncated are never both True.
          assert not np.any(np.logical_and(is_truncated, is_terminated))
  
          # Make sure, is_first and is_last are trivially correct.
          assert np.all(is_last[:, -1])
          assert np.all(is_first[:, 0])
  
          # All fields have same shape.
          assert (
              obs.shape[:2]
              == rewards.shape
              == actions.shape
              == is_first.shape
              == is_last.shape
              == is_terminated.shape
          )
  
          # All rewards match obs.
          assert np.all(np.equal(obs * 0.1, rewards))
          # All actions are always the same as their obs, except when terminated (one
          # less).
          assert np.all(np.where(is_last, True, np.equal(obs, actions)))
          # All actions on is_terminated=True must be the same as the previous ones
          # (we repeat the action b/c the last one is anyways a dummy one (action
          # picked in terminal observation/state)).
          assert np.all(
              np.where(
                  is_terminated[:, 1:],
                  np.equal(actions[:, 1:], actions[:, :-1]),
                  True,
              )
          )
          # Where is_terminated, the next rewards should always be 0.0
          # (reset rewards).
          assert np.all(
              np.where(
                  is_terminated[:, :-1], rewards[:, 1:] == 0.0, True
              )
          )
  
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 3 for calling the TestEpisodeReplayBuffer obj's underlying member method code with example usage: obj.test_episode_replay_buffer_sample_logic() without return types