code: def test_parmetis_preprocessing():
      with tempfile.TemporaryDirectory() as root_dir:
          num_chunks = 2
          g = create_chunked_dataset(root_dir, num_chunks)
  
          # Trigger ParMETIS pre-processing here.
          input_dir = os.path.join(root_dir, "chunked-data")
          results_dir = os.path.join(root_dir, "parmetis-data")
          os.system(
              f"mpirun -np {num_chunks} python3 tools/distpartitioning/parmetis_preprocess.py "
              f"--schema {metadata.json} "
              f"--input_dir {input_dir} "
              f"--output_dir {results_dir} "
              f"--num_parts {num_chunks}"
          )
  
          # Now add all the tests and check whether the test has passed or failed.
          # Read parmetis_nfiles and ensure all files are present.
          parmetis_data_dir = os.path.join(root_dir, "parmetis-data")
          assert os.path.isdir(parmetis_data_dir)
          parmetis_nodes_file = os.path.join(
              parmetis_data_dir, "parmetis_nfiles.txt"
          )
          assert os.path.isfile(parmetis_nodes_file)
  
          # `parmetis_nfiles.txt` should have each line in the following format.
          # <filename> <global_id_start> <global_id_end>
          with open(parmetis_nodes_file, "r") as nodes_metafile:
              lines = nodes_metafile.readlines()
              total_node_count = 0
              for line in lines:
                  tokens = line.split(" ")
                  assert len(tokens) == 3
                  assert os.path.isfile(tokens[0])
                  assert int(tokens[1]) == total_node_count
  
                  # check contents of each of the nodes files here
                  with open(tokens[0], "r") as nodes_file:
                      node_lines = nodes_file.readlines()
                      for line in node_lines:
                          val = line.split(" ")
                          # <ntype_id> <weight_list> <mask_list> <type_node_id>
                          assert len(val) == 8
                      node_count = len(node_lines)
                      total_node_count += node_count
                  assert int(tokens[2]) == total_node_count
  
          # Meta_data object.
          output_dir = os.path.join(root_dir, "chunked-data")
          json_file = os.path.join(output_dir, "metadata.json")
          assert os.path.isfile(json_file)
          with open(json_file, "rb") as f:
              meta_data = json.load(f)
  
          # Count the total no. of nodes.
          true_node_count = 0
          num_nodes_per_chunk = meta_data["num_nodes_per_chunk"]
          for i in range(len(num_nodes_per_chunk)):
              node_per_part = num_nodes_per_chunk[i]
              for j in range(len(node_per_part)):
                  true_node_count += node_per_part[j]
          assert total_node_count == true_node_count
  
          # Read parmetis_efiles and ensure all files are present.
          # This file contains a list of filenames.
          parmetis_edges_file = os.path.join(
              parmetis_data_dir, "parmetis_efiles.txt"
          )
          assert os.path.isfile(parmetis_edges_file)
  
          with open(parmetis_edges_file, "r") as edges_metafile:
              lines = edges_metafile.readlines()
              total_edge_count = 0
              for line in lines:
                  edges_filename = line.strip()
                  assert os.path.isfile(edges_filename)
  
                  with open(edges_filename, "r") as edges_file:
                      edge_lines = edges_file.readlines()
                      total_edge_count += len(edge_lines)
                      for line in edge_lines:
                          val = line.split(" ")
                          assert len(val) == 2
  
          # Count the total no. of edges
          true_edge_count = 0
          num_edges_per_chunk = meta_data["num_edges_per_chunk"]
          for i in range(len(num_edges_per_chunk)):
              edges_per_part = num_edges_per_chunk[i]
              for j in range(len(edges_per_part)):
                  true_edge_count += edges_per_part[j]
          assert true_edge_count == total_edge_count
  
file_path: dgl/tests/tools/test_parmetis.py
imports: import argparse
  import json
  import os
  import sys
  import tempfile
  import unittest
  import dgl
  import numpy
  import torch
  from dgl.data.utils import load_graphs, load_tensors
  from partition_algo.base import load_partition_meta
  from pytest_utils import create_chunked_dataset
label_desc: Use image node_id 1 for calling a global function with example usage: test_parmetis_preprocessing() without return types