code: def test_mg_replicate_edgelist(dask_client, input_combo):
      df = input_combo["graph_file"].get_edgelist()
      distributed = input_combo["distributed"]
  
      use_weights = input_combo["use_weights"]
      use_edge_ids = input_combo["use_edge_ids"]
      use_edge_type_ids = input_combo["use_edge_type_ids"]
  
      columns = [srcCol, dstCol]
      weight = None
      edge_id = None
      edge_type = None
  
      if use_weights:
          df = df.rename(columns={"wgt": edgeWeightCol})
          columns.append(edgeWeightCol)
          weight = edgeWeightCol
      if use_edge_ids:
          df = df.reset_index().rename(columns={"index": edgeIdCol})
          df[edgeIdCol] = df[edgeIdCol].astype(df[srcCol].dtype)
          columns.append(edgeIdCol)
          edge_id = edgeIdCol
      if use_edge_type_ids:
          df[edgeTypeCol] = np.random.randint(0, 10, size=len(df))
          df[edgeTypeCol] = df[edgeTypeCol].astype(df[srcCol].dtype)
          columns.append(edgeTypeCol)
          edge_type = edgeTypeCol
  
      if distributed:
          # Distribute the edges across all ranks
          num_workers = len(dask_client.scheduler_info()["workers"])
          df = dask_cudf.from_cudf(df, npartitions=num_workers)
      ddf = replicate_edgelist(
          df[columns],
          weight=weight,
          edge_id=edge_id,
          edge_type=edge_type,
      )
  
      if distributed:
          df = df.compute()
  
      for i in range(ddf.npartitions):
          result_df = (
              ddf.get_partition(i)
              .compute()
              .sort_values([srcCol, dstCol])
              .reset_index(drop=True)
          )
          expected_df = (
              df[columns]
              .sort_values([srcCol, dstCol])
              .reset_index(drop=True)
          )
  
          assert_frame_equal(
              expected_df, result_df, check_dtype=False, check_like=True
          )
  
file_path: cugraph/python/cugraph/cugraph/tests/internals/test_replicate_edgelist_mg.py
imports: import gc
  import pytest
  import dask_cudf
  import numpy
  from cugraph.testing import UNDIRECTED_DATASETS, karate_disjoint
  from cugraph.structure.replicate_edgelist import replicate_edgelist
  from cudf.testing.testing import assert_frame_equal
  from pylibcugraph.testing.utils import gen_fixture_params_product
label_desc: Use image node_id 3 for calling a global function with example usage: test_mg_replicate_edgelist(dask_client, input_combo) without return types