code: def test_placement_group_local_resource_view(
      monkeypatch, ray_start_cluster
  ):
      """Please refer to https://github.com/ray-project/ray/pull/19911
      for more details.
      """
      with monkeypatch.context() as m:
          # Increase broadcasting interval so that node resource will arrive
          # at raylet after local resource all being allocated.
          m.setenv(
              "RAY_raylet_report_resources_period_milliseconds", "2000"
          )
          cluster = ray_start_cluster
  
          cluster.add_node(num_cpus=16, object_store_memory=1e9)
          cluster.wait_for_nodes()
          # We need to init here so that we can make sure it's connecting to
          # the raylet where it only has cpu resources.
          # This is a hacky way to prevent scheduling hanging which will
          # schedule <CPU:1> job to the node with GPU and for <GPU:1, CPU:1> task
          # there is no node has this resource.
          ray.init(address="auto")
          cluster.add_node(num_cpus=16, num_gpus=1)
          cluster.wait_for_nodes()
          NUM_CPU_BUNDLES = 30
  
          @ray.remote(num_cpus=1)
          class Worker(object):
              def __init__(self, i):
                  self.i = i
  
              def work(self):
                  time.sleep(0.1)
                  print("work ", self.i)
  
          @ray.remote(num_cpus=1, num_gpus=1)
          class Trainer(object):
              def __init__(self, i):
                  self.i = i
  
              def train(self):
                  time.sleep(0.2)
                  print("train ", self.i)
  
          bundles = [{"CPU": 1, "GPU": 1}]
          bundles += [{"CPU": 1} for _ in range(NUM_CPU_BUNDLES)]
          pg = placement_group(bundles, strategy="PACK")
          ray.get(pg.ready())
  
          # Local resource will be allocated and here we are to ensure
          # local view is consistent and node resouce updates are discarded
          workers = [
              Worker.options(
                  scheduling_strategy=PlacementGroupSchedulingStrategy(
                      placement_group=pg
                  )
              ).remote(i)
              for i in range(NUM_CPU_BUNDLES)
          ]
          trainer = Trainer.options(
              scheduling_strategy=PlacementGroupSchedulingStrategy(
                  placement_group=pg
              )
          ).remote(0)
          ray.get(
              [workers[i].work.remote() for i in range(NUM_CPU_BUNDLES)]
          )
          ray.get(trainer.train.remote())
  
imports: import sys
  import time
  import pytest
  import ray
  import ray._private.gcs_utils
  import ray.cluster_utils
  import ray.experimental.internal_kv
  from ray._private.ray_constants import DEBUG_AUTOSCALING_ERROR, DEBUG_AUTOSCALING_STATUS
  from ray.autoscaler._private.constants import AUTOSCALER_UPDATE_INTERVAL_S
  from ray._private.test_utils import convert_actor_state, generate_system_config_map, is_placement_group_removed, kill_actor_and_wait_for_failure, reset_autoscaler_v2_enabled_cache, run_string_as_driver, wait_for_condition
  from ray.autoscaler._private.commands import debug_status
  from ray.exceptions import RaySystemError
  from ray.util.client.ray_client_helpers import connect_to_client_or_not
  from ray.util.placement_group import placement_group, remove_placement_group
  from ray.util.scheduling_strategies import PlacementGroupSchedulingStrategy
label_desc: Use image node_id 16 for calling a global function with example usage: test_placement_group_local_resource_view(monkeypatch, ray_start_cluster) without return types