class_docstr: Tests for ConcurrencyCapBackpressurePolicy.
class_docstr_tok: ["Tests","for","ConcurrencyCapBackpressurePolicy","."]
docstr: Test setting the multiplier to 1.0, which means no ramping up of the
  concurrency cap.
docstr_tok: ["Test","setting","the","multiplier","to","1.0",",","which","means","no","ramping","up","of","the","concurrency","cap","."]
code: def test_e2e_no_ramping_up(self):
      """Test setting the multiplier to 1.0, which means no ramping up of the
      concurrency cap."""
      with self._patch_config(1, 1, 1):
          actor = self._create_record_time_actor()
          map_func1 = self._get_map_func(actor, 1)
          N = self.__class__._cluster_cpus
          ds = ray.data.range(N, parallelism=N)
          ds = ds.map_batches(map_func1, batch_size=None, num_cpus=1)
          res = ds.take_all()
          self.assertEqual(len(res), N)
  
          start, end = ray.get(
              actor.get_start_and_end_time_for_all_tasks_of_op.remote(1)
          )
          for i in range(len(start) - 1):
              assert start[i] < end[i] < start[i + 1], (i, start, end)
  
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 9 for calling the TestConcurrencyCapBackpressurePolicy obj's underlying member method code with example usage: obj.test_e2e_no_ramping_up() without return types