code: def test_shuffle_assign():
      band_num = 8
      all_bands = [(f"address_{i}", "numa-0") for i in range(band_num)]
  
      pdf = pd.DataFrame(np.random.rand(32, 4))
      df = md.DataFrame(pdf, chunk_size=4)
      r = df.groupby(0).sum(method="shuffle")
      chunk_graph = build_graph([r], tile=True)
  
      band_resource = dict(
          (band, Resource(num_cpus=1)) for band in all_bands
      )
  
      reassign_worker_ops = [
          chunk.op
          for chunk in chunk_graph
          if need_reassign_worker(chunk.op)
      ]
      start_ops = list(GraphAnalyzer._iter_start_ops(chunk_graph))
      to_assign_ops = start_ops + reassign_worker_ops
  
      assigner = GraphAssigner(
          chunk_graph, to_assign_ops, band_resource
      )
      assigns = assigner.assign()
      assert len(assigns) == 16
      init_assigns = set()
      reducer_assigns = set()
      for chunk, assign in assigns.items():
          if chunk.op.stage == OperandStage.reduce:
              reducer_assigns.add(assign)
          else:
              init_assigns.add(assign)
      # init and reducers are assigned on all bands
      assert len(init_assigns) == len(reducer_assigns) == 8
  
file_path: mars/mars/services/task/analyzer/tests/test_assigner.py
imports: import numpy
  import pandas
  from .....None import dataframe
  from .....config import Config
  from .....core import ChunkGraph
  from .....core.graph.builder.utils import build_graph
  from .....core.operand import OperandStage
  from .....tensor.random import TensorRand
  from .....tensor.arithmetic import TensorAdd
  from .....tensor.fetch import TensorFetch
  from .....resource import Resource
  from ...core import Task
  from ..analyzer import GraphAnalyzer, need_reassign_worker
  from ..assigner import GraphAssigner
label_desc: Use image node_id 2 for calling a global function with example usage: test_shuffle_assign() without return types