class_docstr: Tests for StreamOutputBackpressurePolicy.
class_docstr_tok: ["Tests","for","StreamOutputBackpressurePolicy","."]
code: def test_no_deadlock(self):
      # The producer needs all 5 CPUs, and the consumer has no CPU to run.
      # In this case, we shouldn't backpressure the producer and let it run
      # until it finishes.
      producer_timestamps, consumer_timestamps = self._run_dataset(
          producer_num_cpus=5, consumer_num_cpus=1
      )
      assert producer_timestamps[-1] < consumer_timestamps[0], (
          producer_timestamps,
          consumer_timestamps,
      )
  
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 6 for calling the TestStreamOutputBackpressurePolicy obj's underlying member method code with example usage: obj.test_no_deadlock() without return types