class_docstr: Tests for StreamOutputBackpressurePolicy.
class_docstr_tok: ["Tests","for","StreamOutputBackpressurePolicy","."]
docstr: Basic unit test for the policy without real execution.
docstr_tok: ["Basic","unit","test","for","the","policy","without","real","execution","."]
code: def test_policy_basic(self):
      """Basic unit test for the policy without real execution."""
      up_op, up_state = self._create_mock_op_and_op_state("up")
      down_op, down_state = self._create_mock_op_and_op_state("down")
      topology = {}
      topology[up_op] = up_state
      topology[down_op] = down_state
  
      policy = StreamingOutputBackpressurePolicy(topology)
      assert (
          policy._max_num_blocks_in_op_output_queue
          == self._max_blocks_in_op_output_queue
      )
      assert (
          policy._max_num_blocks_in_streaming_gen_buffer
          == self._max_blocks_in_generator_buffer
      )
  
      # Buffers are empty, both ops can read up to the max.
      res = policy.calculate_max_blocks_to_read_per_op(topology)
      assert res == {
          up_state: self._max_blocks_in_op_output_queue,
          down_state: self._max_blocks_in_op_output_queue,
      }
  
      # up_op's buffer is full, but down_up has no active tasks.
      # We'll still allow up_op to read 1 block.
      up_state.outqueue_num_blocks.return_value = (
          self._max_blocks_in_op_output_queue
      )
      res = policy.calculate_max_blocks_to_read_per_op(topology)
      assert res == {
          up_state: 1,
          down_state: self._max_blocks_in_op_output_queue,
      }
  
      # down_op now has 1 active task. So we won't allow up_op to read any more.
      down_op.num_active_tasks.return_value = 1
      res = policy.calculate_max_blocks_to_read_per_op(topology)
      assert res == {
          up_state: 0,
          down_state: self._max_blocks_in_op_output_queue,
      }
  
      # After `MAX_OUTPUT_IDLE_SECONDS` of no outputs from down_up,
      # we'll allow up_op to read 1 block again.
      with patch.object(
          StreamingOutputBackpressurePolicy,
          "MAX_OUTPUT_IDLE_SECONDS",
          0.1,
      ):
          time.sleep(0.11)
          res = policy.calculate_max_blocks_to_read_per_op(topology)
          assert res == {
              up_state: 1,
              down_state: self._max_blocks_in_op_output_queue,
          }
  
          # down_up now has outputs, so we won't allow up_op to read any more.
          down_op.metrics.num_outputs_generated = 1
          res = policy.calculate_max_blocks_to_read_per_op(topology)
          assert res == {
              up_state: 0,
              down_state: self._max_blocks_in_op_output_queue,
          }
  
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 4 for calling the TestStreamOutputBackpressurePolicy obj's underlying member method code with example usage: obj.test_policy_basic() without return types