code: def test_configure_output_locality():
      inputs = make_ref_bundles([[x] for x in range(20)])
      o1 = InputDataBuffer(inputs)
      o2 = MapOperator.create(
          make_map_transformer(lambda block: [b * -1 for b in block]),
          o1,
      )
      o3 = MapOperator.create(
          make_map_transformer(lambda block: [b * 2 for b in block]),
          o2,
          compute_strategy=ray.data.ActorPoolStrategy(size=1),
      )
      # No locality.
      build_streaming_topology(
          o3, ExecutionOptions(locality_with_output=False)
      )
      assert o2._ray_remote_args.get("scheduling_strategy") is None
      assert o3._ray_remote_args.get("scheduling_strategy") == "SPREAD"
  
      # Current node locality.
      build_streaming_topology(
          o3, ExecutionOptions(locality_with_output=True)
      )
      s1 = o2._get_runtime_ray_remote_args()["scheduling_strategy"]
      assert isinstance(s1, NodeAffinitySchedulingStrategy)
      assert s1.node_id == ray.get_runtime_context().get_node_id()
      s2 = o3._get_runtime_ray_remote_args()["scheduling_strategy"]
      assert isinstance(s2, NodeAffinitySchedulingStrategy)
      assert s2.node_id == ray.get_runtime_context().get_node_id()
  
      # Multi node locality.
      build_streaming_topology(
          o3, ExecutionOptions(locality_with_output=["node1", "node2"])
      )
      s1a = o2._get_runtime_ray_remote_args()["scheduling_strategy"]
      s1b = o2._get_runtime_ray_remote_args()["scheduling_strategy"]
      s1c = o2._get_runtime_ray_remote_args()["scheduling_strategy"]
      assert s1a.node_id == "node1"
      assert s1b.node_id == "node2"
      assert s1c.node_id == "node1"
      s2a = o3._get_runtime_ray_remote_args()["scheduling_strategy"]
      s2b = o3._get_runtime_ray_remote_args()["scheduling_strategy"]
      s2c = o3._get_runtime_ray_remote_args()["scheduling_strategy"]
      assert s2a.node_id == "node1"
      assert s2b.node_id == "node2"
      assert s2c.node_id == "node1"
  
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 14 for calling a global function with example usage: test_configure_output_locality() without return types