code: def test_OnDiskDataset_preprocess_homogeneous(edge_fmt):
      """Test preprocess of OnDiskDataset."""
      with tempfile.TemporaryDirectory() as test_dir:
          # All metadata fields are specified.
          dataset_name = "graphbolt_test"
          num_nodes = 4000
          num_edges = 20000
          num_classes = 10
  
          # Generate random graph.
          yaml_content = gbt.random_homo_graphbolt_graph(
              test_dir,
              dataset_name,
              num_nodes,
              num_edges,
              num_classes,
              edge_fmt=edge_fmt,
          )
          yaml_file = os.path.join(test_dir, "metadata.yaml")
          with open(yaml_file, "w") as f:
              f.write(yaml_content)
          output_file = gb.ondisk_dataset.preprocess_ondisk_dataset(
              test_dir, include_original_edge_id=False
          )
  
          with open(output_file, "rb") as f:
              processed_dataset = yaml.load(f, Loader=yaml.Loader)
  
          assert processed_dataset["dataset_name"] == dataset_name
          assert (
              processed_dataset["tasks"][0]["num_classes"]
              == num_classes
          )
          assert "graph" not in processed_dataset
          assert "graph_topology" in processed_dataset
  
          fused_csc_sampling_graph = torch.load(
              os.path.join(
                  test_dir, processed_dataset["graph_topology"]["path"]
              )
          )
          assert fused_csc_sampling_graph.total_num_nodes == num_nodes
          assert fused_csc_sampling_graph.total_num_edges == num_edges
          assert (
              fused_csc_sampling_graph.node_attributes is not None
              and "feat" in fused_csc_sampling_graph.node_attributes
          )
          assert (
              fused_csc_sampling_graph.edge_attributes is not None
              and gb.ORIGINAL_EDGE_ID
              not in fused_csc_sampling_graph.edge_attributes
              and "feat" in fused_csc_sampling_graph.edge_attributes
          )
  
          num_samples = 100
          fanout = 1
          subgraph = fused_csc_sampling_graph.sample_neighbors(
              torch.arange(num_samples),
              torch.tensor([fanout]),
          )
          assert len(subgraph.sampled_csc.indices) <= num_samples
  
      with tempfile.TemporaryDirectory() as test_dir:
          # All metadata fields are specified.
          dataset_name = "graphbolt_test"
          num_nodes = 4000
          num_edges = 20000
          num_classes = 10
  
          # Generate random graph.
          yaml_content = gbt.random_homo_graphbolt_graph(
              test_dir,
              dataset_name,
              num_nodes,
              num_edges,
              num_classes,
              edge_fmt=edge_fmt,
          )
          yaml_file = os.path.join(test_dir, "metadata.yaml")
          with open(yaml_file, "w") as f:
              f.write(yaml_content)
          # Test generating original_edge_id.
          output_file = gb.ondisk_dataset.preprocess_ondisk_dataset(
              test_dir, include_original_edge_id=True
          )
          with open(output_file, "rb") as f:
              processed_dataset = yaml.load(f, Loader=yaml.Loader)
          fused_csc_sampling_graph = torch.load(
              os.path.join(
                  test_dir, processed_dataset["graph_topology"]["path"]
              )
          )
          assert (
              fused_csc_sampling_graph.edge_attributes is not None
              and gb.ORIGINAL_EDGE_ID
              in fused_csc_sampling_graph.edge_attributes
          )
          fused_csc_sampling_graph = None
  
imports: import os
  import pickle
  import random
  import re
  import tempfile
  import unittest
  import warnings
  import numpy
  import pandas
  import pydantic
  import pytest
  import torch
  import yaml
  from dgl import graphbolt
  from dgl.base import DGLWarning
  from ..None import gb_test_utils
label_desc: Use image node_id 19 for calling a global function with example usage: test_OnDiskDataset_preprocess_homogeneous(edge_fmt) without return types