code: def test_longer_random_sequence_of_queue_ops(buffer_type):
      """A long random sequence of added and retrieved values"""
      q = buffer_type(100, 80)
  
      for _ in six.moves.xrange(10000):
          if q.can_add():
              _add_many(
                  q, np.random.random((np.random.randint(1, 10),))
              )
          assert q.size < 100 + 10
          for _ in range(np.random.randint(1, 10)):
              if not q.can_retrieve():
                  break
              # Make sure never get to less than `min_after_retrieve` elements
              assert 80 <= q.size
              _retrieve(q)
  
file_path: petastorm/petastorm/tests/test_shuffling_buffer.py
imports: import numpy
  import pytest
  import six
  from petastorm.reader_impl.shuffling_buffer import NoopShufflingBuffer, RandomShufflingBuffer
  from petastorm.reader_impl.pytorch_shuffling_buffer import BatchedNoopShufflingBuffer, BatchedRandomShufflingBuffer
label_desc: Use image node_id 11 for calling a global function with example usage: test_longer_random_sequence_of_queue_ops(buffer_type) without return types