code: def test_shuffle_graph(tileable, fuse):
      # can't test df.groupby and mt.bincount, those chunk graph build depend on ctx.get_chunks_meta/get_chunks_result
      chunk_graph = tileable.build_graph(tile=True)
      assert len(chunk_graph) > 0
      all_bands = [(f"address_{i}", "numa-0") for i in range(5)]
      band_resource = dict(
          (band, Resource(num_cpus=1)) for band in all_bands
      )
      task = Task("mock_task", "mock_session", fuse_enabled=fuse)
      analyzer = GraphAnalyzer(
          chunk_graph,
          band_resource,
          task,
          Config(),
          dict(),
          shuffle_fetch_type=ShuffleFetchType.FETCH_BY_INDEX,
      )
      subtask_graph = analyzer.gen_subtask_graph()
      proxy_subtasks = []
      for subtask in subtask_graph:
          for c in subtask.chunk_graph.results:
              if isinstance(c.op, ShuffleProxy):
                  assert len(subtask.chunk_graph.results) == 1
                  proxy_subtasks.append(subtask)
      proxy_chunks = [
          c
          for subtask in proxy_subtasks
          for c in chunk_graph
          if subtask.chunk_graph.results[0].key == c.key
      ]
      assert len(proxy_subtasks) == len(proxy_chunks)
      assert len(proxy_subtasks) > 0
      assert len(proxy_subtasks) == len(
          subtask_graph.get_shuffle_proxy_subtasks()
      )
      for proxy_chunk, proxy_subtask in zip(
          proxy_chunks, proxy_subtasks
      ):
          reducer_subtasks = subtask_graph.successors(proxy_subtask)
          for reducer_subtask in reducer_subtasks:
              start_chunks = list(
                  reducer_subtask.chunk_graph.iter_indep()
              )
              assert len(start_chunks) == 1
              assert (
                  start_chunks[0].op.shuffle_fetch_type
                  == ShuffleFetchType.FETCH_BY_INDEX
              )
          reducer_chunks = chunk_graph.successors(proxy_chunk)
          # single reducer may have multiple output chunks, see `PSRSShuffle._execute_reduce
          if len(reducer_subtasks) != len(reducer_chunks):
              assert len(reducer_subtasks) == len(
                  set(c.op for c in reducer_chunks)
              )
          mapper_subtasks = subtask_graph.predecessors(proxy_subtask)
          for mapper_subtask in mapper_subtasks:
              assert len(mapper_subtask.chunk_graph.results) == 1
          mapper_chunks = chunk_graph.predecessors(proxy_chunk)
          assert len(mapper_subtasks) == len(mapper_chunks)
  
file_path: mars/mars/services/task/analyzer/tests/test_analyzer.py
imports: import pytest
  from .....None import dataframe
  from .....None import tensor
  from .....config import Config
  from .....core.operand.shuffle import ShuffleFetchType, ShuffleProxy
  from .....resource import Resource
  from ...core import Task
  from ..analyzer import GraphAnalyzer
label_desc: Use image node_id 1 for calling a global function with example usage: test_shuffle_graph(tileable, fuse) without return types