code: def test_calculate_topology_usage():
      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
      )
      o2.current_resource_usage = MagicMock(
          return_value=ExecutionResources(
              cpu=5, object_store_memory=500
          )
      )
      o3.current_resource_usage = MagicMock(
          return_value=ExecutionResources(
              cpu=10, object_store_memory=1000
          )
      )
      topo, _ = build_streaming_topology(o3, ExecutionOptions())
      inputs[0].size_bytes = MagicMock(return_value=200)
      topo[o2].add_output(inputs[0])
      usage = TopologyResourceUsage.of(topo)
      assert len(usage.downstream_memory_usage) == 3, usage
      assert usage.overall == ExecutionResources(15, 0, 1700)
      assert (
          usage.downstream_memory_usage[o1].object_store_memory == 1700
      ), usage
      assert (
          usage.downstream_memory_usage[o1].topology_fraction == 1
      ), usage
      assert (
          usage.downstream_memory_usage[o2].object_store_memory == 1700
      ), usage
      assert (
          usage.downstream_memory_usage[o2].topology_fraction == 1
      ), usage
      assert (
          usage.downstream_memory_usage[o3].object_store_memory == 1000
      ), usage
      assert (
          usage.downstream_memory_usage[o3].topology_fraction == 0.5
      ), usage
  
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 15 for calling a global function with example usage: test_calculate_topology_usage() without return types