code: def test_create_mg_graph(dask_client, input_combo):
      G = input_combo["MGGraph"]
      ddf = input_combo["input_df"]
      df = ddf.compute()
  
      # ensure graph exists
      assert G._plc_graph is not None
  
      # ensure graph is partitioned correctly
      assert len(G._plc_graph) == len(dask_client.has_what())
  
      start = dask_cudf.from_cudf(
          cudf.Series([1], dtype="int32"), len(G._plc_graph)
      )
      vertex_dtype = start.dtype
  
      if G.renumbered:
          start = G.lookup_internal_vertex_id(start, None)
      data_start = persist_dask_df_equal_parts_per_worker(
          start, dask_client, return_type="dict"
      )
  
      res = [
          dask_client.submit(
              lambda sID, mg_graph_x, st_x: pylibcugraph_bfs(
                  ResourceHandle(Comms.get_handle(sID).getHandle()),
                  mg_graph_x,
                  st_x,
                  False,
                  0,
                  True,
                  False,
              ),
              Comms.get_session_id(),
              G._plc_graph[w],
              st[0] if st else cudf.Series(dtype=vertex_dtype),
              workers=[w],
          )
          for w, st in data_start.items()
      ]
  
      wait(res)
  
      cudf_result = [
          dask_client.submit(convert_to_cudf, cp_arrays)
          for cp_arrays in res
      ]
      wait(cudf_result)
  
      result_dist = dask_cudf.from_delayed(cudf_result)
  
      if G.renumbered:
          result_dist = G.unrenumber(result_dist, "vertex")
          result_dist = G.unrenumber(result_dist, "predecessor")
          result_dist = result_dist.fillna(-1)
  
      result_dist = result_dist.compute()
  
      g = cugraph.Graph(directed=G.properties.directed)
      g.from_cudf_edgelist(df, "src", "dst")
      expected_dist = cugraph.bfs(g, cudf.Series([1], dtype="int32"))
  
      compare_dist = expected_dist.merge(
          result_dist, on="vertex", suffixes=["_local", "_dask"]
      )
  
      err = 0
  
      for i in range(len(compare_dist)):
          if (
              compare_dist["distance_local"].iloc[i]
              != compare_dist["distance_dask"].iloc[i]
          ):
              err = err + 1
      assert err == 0
  
imports: import gc
  import random
  import pytest
  import copy
  import cupy
  import cudf
  import cugraph
  import dask_cudf
  import cugraph.dask
  import cugraph.dask.comms.comms
  from cugraph.testing import utils
  from dask.distributed import wait
  from pylibcugraph import ResourceHandle
  from pylibcugraph import bfs
  from cudf.testing.testing import assert_frame_equal
  from cugraph.dask.traversal.bfs import convert_to_cudf
  from cugraph.dask.common.input_utils import get_distributed_data
  from pylibcugraph.testing.utils import gen_fixture_params_product
  from cugraph.dask.common.part_utils import persist_dask_df_equal_parts_per_worker
label_desc: Use image node_id 5 for calling a global function with example usage: test_create_mg_graph(dask_client, input_combo) without return types