code: def test_execution_allowed():
      op = InputDataBuffer([])
  
      def stub(res: ExecutionResources) -> TopologyResourceUsage:
          return TopologyResourceUsage(res, EMPTY_DOWNSTREAM_USAGE)
  
      # CPU.
      op.incremental_resource_usage = MagicMock(
          return_value=ExecutionResources(cpu=1)
      )
      assert _execution_allowed(
          op, stub(ExecutionResources(cpu=1)), ExecutionResources(cpu=2)
      )
      assert not _execution_allowed(
          op, stub(ExecutionResources(cpu=2)), ExecutionResources(cpu=2)
      )
      assert _execution_allowed(
          op, stub(ExecutionResources(cpu=2)), ExecutionResources(gpu=2)
      )
  
      # GPU.
      op.incremental_resource_usage = MagicMock(
          return_value=ExecutionResources(cpu=0, gpu=1)
      )
      assert _execution_allowed(
          op, stub(ExecutionResources(gpu=1)), ExecutionResources(gpu=2)
      )
      assert not _execution_allowed(
          op, stub(ExecutionResources(gpu=2)), ExecutionResources(gpu=2)
      )
  
      # Test conversion to indicator (0/1).
      op.incremental_resource_usage = MagicMock(
          return_value=ExecutionResources(cpu=0, gpu=100)
      )
      assert _execution_allowed(
          op, stub(ExecutionResources(gpu=1)), ExecutionResources(gpu=2)
      )
      assert _execution_allowed(
          op,
          stub(ExecutionResources(gpu=1.5)),
          ExecutionResources(gpu=2),
      )
      assert not _execution_allowed(
          op, stub(ExecutionResources(gpu=2)), ExecutionResources(gpu=2)
      )
  
      # Test conversion to indicator (0/1).
      op.incremental_resource_usage = MagicMock(
          return_value=ExecutionResources(cpu=0, gpu=0.1)
      )
      assert _execution_allowed(
          op, stub(ExecutionResources(gpu=1)), ExecutionResources(gpu=2)
      )
      assert _execution_allowed(
          op,
          stub(ExecutionResources(gpu=1.5)),
          ExecutionResources(gpu=2),
      )
      assert not _execution_allowed(
          op, stub(ExecutionResources(gpu=2)), ExecutionResources(gpu=2)
      )
  
imports: import collections
  import time
  from unittest.mock import MagicMock
  import pytest
  import ray
  from ray._private.test_utils import run_string_as_driver_nonblocking, wait_for_condition
  from ray.data._internal.execution.interfaces import ExecutionOptions, ExecutionResources, PhysicalOperator
  from ray.data._internal.execution.interfaces.physical_operator import MetadataOpTask
  from ray.data._internal.execution.operators.input_data_buffer import InputDataBuffer
  from ray.data._internal.execution.operators.map_operator import MapOperator
  from ray.data._internal.execution.operators.map_transformer import create_map_transformer_from_block_fn
  from ray.data._internal.execution.streaming_executor import _debug_dump_topology, _validate_dag
  from ray.data._internal.execution.streaming_executor_state import AutoscalingState, DownstreamMemoryInfo, OpState, TopologyResourceUsage, _execution_allowed, build_streaming_topology, process_completed_tasks, select_operator_to_run, update_operator_states
  from ray.data._internal.execution.util import make_ref_bundles
  from ray.data.tests.conftest import *
  from ray.util.scheduling_strategies import NodeAffinitySchedulingStrategy
label_desc: Use image node_id 11 for calling a global function with example usage: test_execution_allowed() and returns: TopologyResourceUsage