class_docstr: Tests for StreamOutputBackpressurePolicy.
class_docstr_tok: ["Tests","for","StreamOutputBackpressurePolicy","."]
code: def _create_mock_op_and_op_state(
      self,
      name,
      outqueue_num_blocks=0,
      num_active_tasks=0,
      num_outputs_generated=0,
  ):
      op = MagicMock()
      op.__str__.return_value = f"Op({name})"
      op.num_active_tasks.return_value = num_active_tasks
      op.metrics.num_outputs_generated = num_outputs_generated
  
      state = MagicMock()
      state.__str__.return_value = f"OpState({name})"
      state.outqueue_num_blocks.return_value = outqueue_num_blocks
  
      state.op = op
      return op, state
  
imports: import functools
  import time
  import unittest
  from collections import defaultdict
  from contextlib import contextmanager
  from unittest.mock import MagicMock, patch
  import numpy
  import pytest
  import ray
  from ray.data._internal.execution.backpressure_policy import ENABLED_BACKPRESSURE_POLICIES_CONFIG_KEY, ConcurrencyCapBackpressurePolicy, StreamingOutputBackpressurePolicy
  from ray.data.tests.conftest import restore_data_context
  from ray.data.tests.conftest import CoreExecutionMetrics, assert_core_execution_metrics_equals, get_initial_core_execution_metrics_snapshot
  from ray.tests.conftest import shutdown_only
label_desc: Use image node_id 3 for calling the TestStreamOutputBackpressurePolicy obj's underlying member method code with example usage: obj._create_mock_op_and_op_state(name, outqueue_num_blocks, num_active_tasks, num_outputs_generated) and returns: op,  state