code: def test_cellpx_error_stream(self):
      self.minimum_engines(6)
      ip = get_ipython()
      v = self.client[:]
      v.block = True
      v.activate()
      v.scatter("rank", range(len(v)), flatten=True, block=True)
  
      with capture_output(display=False) as io:
          with pytest.raises(
              ipp.error.AlreadyDisplayedError
          ) as exc_info:
              ip.run_cell_magic(
                  "px",
                  "--stream",
                  "import time; time.sleep(rank); raise RuntimeError(f'oops! {rank}')",
              )
  
      print(io.stdout)
      print(io.stderr, file=sys.stderr)
      assert "RuntimeError" in io.stderr
      assert len(v) <= io.stderr.count("RuntimeError:") < len(v) * 2
      printed_tb = "\n".join(exc_info.value.render_traceback())
      assert printed_tb == f"{len(v)} errors"
  
imports: import re
  import signal
  import sys
  import time
  import pytest
  from IPython import get_ipython
  from IPython.utils.io import capture_output
  import ipyparallel
  from ipyparallel import AsyncResult
  from .clienttest import ClusterTestCase, generate_output, raises_remote
label_desc: Use image node_id 10 for calling the TestParallelMagics obj's underlying member method code with example usage: obj.test_cellpx_error_stream() without return types