code: def test_sample_neighbors_hetero_pick_number(
      fanouts, replace, labor, probs_name
  ):
      if F._default_context_str == "gpu" and replace == True:
          pytest.skip(
              "Sampling with replacement not yet supported on GPU."
          )
      # Initialize data.
      total_num_nodes = 10
      total_num_edges = 9
      ntypes = {"N0": 0, "N1": 1, "N2": 2, "N3": 3}
      etypes = {
          "N1:R0:N0": 0,
          "N2:R1:N0": 1,
          "N3:R2:N0": 2,
      }
      indptr = torch.LongTensor([0, 9, 9, 9, 9, 9, 9, 9, 9, 9, 9])
      indices = torch.LongTensor([1, 2, 3, 4, 5, 6, 7, 8, 9])
      node_type_offset = torch.LongTensor([0, 1, 4, 7, 10])
      type_per_edge = torch.LongTensor([0, 0, 0, 1, 1, 1, 2, 2, 2])
      assert indptr[-1] == total_num_edges
      assert indptr[-1] == len(indices)
      assert node_type_offset[-1] == total_num_nodes
      assert all(type_per_edge < len(etypes))
  
      edge_attributes = {
          "mask": torch.BoolTensor([1, 1, 0, 1, 1, 1, 0, 0, 0]),
          "all": torch.BoolTensor([1, 1, 1, 1, 1, 1, 1, 1, 1]),
          "zero": torch.BoolTensor([0, 0, 0, 0, 0, 0, 0, 0, 0]),
      }
  
      # Construct FusedCSCSamplingGraph.
      graph = gb.fused_csc_sampling_graph(
          indptr,
          indices,
          edge_attributes=edge_attributes,
          node_type_offset=node_type_offset,
          type_per_edge=type_per_edge,
          node_type_to_id=ntypes,
          edge_type_to_id=etypes,
      )
  
      # Generate subgraph via sample neighbors.
      nodes = torch.LongTensor([0, 1])
  
      sampler = (
          graph.sample_layer_neighbors
          if labor
          else graph.sample_neighbors
      )
  
      # Make sure no exception will be thrown.
      subgraph = sampler(
          nodes,
          fanouts=torch.LongTensor(fanouts),
          replace=replace,
          probs_name=probs_name if probs_name != "none" else None,
      )
      print(subgraph)
      if probs_name == "none":
          for etype, pairs in subgraph.sampled_csc.items():
              assert pairs.indptr.size(0) == 2
              sampled_num = pairs.indices.size(0)
              fanout = fanouts[etypes[etype]]
              if fanout == -1:
                  assert sampled_num == 3
              else:
                  if replace:
                      assert sampled_num == fanout
                  else:
                      assert sampled_num == min(fanout, 3)
      else:
          fanout = fanouts[0]  # Here fanout is the same for all etypes.
          for etype, pairs in subgraph.sampled_csc.items():
              assert pairs.indptr.size(0) == 2
              sampled_num = pairs.indices.size(0)
              if etypes[etype] == 0:
                  # Etype 0: 2 valid neighbors.
                  if fanout == -1:
                      assert sampled_num == 2
                  else:
                      if replace:
                          assert sampled_num == fanout
                      else:
                          assert sampled_num == min(fanout, 2)
              elif etypes[etype] == 1:
                  # Etype 1: 3 valid neighbors.
                  if fanout == -1:
                      assert sampled_num == 3
                  else:
                      if replace:
                          assert sampled_num == fanout
                      else:
                          assert sampled_num == min(fanout, 3)
              else:
                  # Etype 2: 0 valid neighbors.
                  assert sampled_num == 0
  
imports: import os
  import pickle
  import re
  import tempfile
  import unittest
  import backend
  import dgl
  import dgl.graphbolt
  import pytest
  import torch
  import torch.multiprocessing
  from dgl.graphbolt.base import etype_str_to_tuple
  from scipy import sparse
  from ..None import gb_test_utils
label_desc: Use image node_id 42 for calling a global function with example usage: test_sample_neighbors_hetero_pick_number(fanouts, replace, labor, probs_name) without return types