code: def test_process_completed_tasks():
      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,
      )
      topo, _ = build_streaming_topology(
          o2, ExecutionOptions(verbose_progress=True)
      )
  
      # Test processing output bundles.
      assert len(topo[o1].outqueue) == 0, topo
      process_completed_tasks(topo, [], 0)
      update_operator_states(topo)
      assert len(topo[o1].outqueue) == 20, topo
  
      # Test processing completed work items.
      sleep_task = MetadataOpTask(0, sleep.remote(), lambda: None)
      done_task_callback = MagicMock()
      done_task = MetadataOpTask(0, ray.put("done"), done_task_callback)
      o2.get_active_tasks = MagicMock(
          return_value=[sleep_task, done_task]
      )
      o2.all_inputs_done = MagicMock()
      o1.all_dependents_complete = MagicMock()
      process_completed_tasks(topo, [], 0)
      update_operator_states(topo)
      done_task_callback.assert_called_once()
      o2.all_inputs_done.assert_not_called()
      o1.all_dependents_complete.assert_not_called()
  
      # Test input finalization.
      done_task_callback = MagicMock()
      done_task = MetadataOpTask(0, ray.put("done"), done_task_callback)
      o2.get_active_tasks = MagicMock(return_value=[done_task])
      o2.all_inputs_done = MagicMock()
      o1.all_dependents_complete = MagicMock()
      o1.completed = MagicMock(return_value=True)
      topo[o1].outqueue.clear()
      process_completed_tasks(topo, [], 0)
      update_operator_states(topo)
      done_task_callback.assert_called_once()
      o2.all_inputs_done.assert_called_once()
      o1.all_dependents_complete.assert_not_called()
  
      # Test dependents completed.
      o2.need_more_inputs = MagicMock(return_value=False)
      o1.all_dependents_complete = MagicMock()
      process_completed_tasks(topo, [], 0)
      update_operator_states(topo)
      o1.all_dependents_complete.assert_called_once()
  
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 6 for calling a global function with example usage: test_process_completed_tasks() without return types