class_docstr: Tests for ConcurrencyCapBackpressurePolicy.
class_docstr_tok: ["Tests","for","ConcurrencyCapBackpressurePolicy","."]
code: def test_basic(self):
      op = MagicMock()
      op.metrics = MagicMock(
          num_tasks_running=0,
          num_tasks_finished=0,
      )
      topology = {op: MagicMock()}
  
      init_cap = 4
      cap_multiply_threshold = 0.5
      cap_multiplier = 2.0
  
      with self._patch_config(
          init_cap, cap_multiply_threshold, cap_multiplier
      ):
          policy = ConcurrencyCapBackpressurePolicy(topology)
  
      self.assertEqual(policy._concurrency_caps[op], 4)
      # Gradually increase num_tasks_running to the cap.
      for i in range(1, init_cap + 1):
          self.assertTrue(policy.can_add_input(op))
          op.metrics.num_tasks_running = i
      # Now num_tasks_running reaches the cap, so can_add_input should return False.
      self.assertFalse(policy.can_add_input(op))
  
      # If we increase num_task_finished to the threshold (4 * 0.5 = 2),
      # it should trigger the cap to increase.
      op.metrics.num_tasks_finished = init_cap * cap_multiply_threshold
      self.assertEqual(policy.can_add_input(op), True)
      self.assertEqual(
          policy._concurrency_caps[op], init_cap * cap_multiplier
      )
  
      # Now the cap is 8 (4 * 2).
      # If we increase num_tasks_finished directly to the next-level's threshold
      # (8 * 2 * 0.5 = 8), it should trigger the cap to increase twice.
      op.metrics.num_tasks_finished = (
          policy._concurrency_caps[op]
          * cap_multiplier
          * cap_multiply_threshold
      )
      op.metrics.num_tasks_running = 0
      self.assertEqual(policy.can_add_input(op), True)
      self.assertEqual(
          policy._concurrency_caps[op], init_cap * cap_multiplier**3
      )
  
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 TestConcurrencyCapBackpressurePolicy obj's underlying member method code with example usage: obj.test_basic() without return types