code: def test_reduce_with_many_common_dependents(
      optimize, keep_self, ndeps, n_reducers
  ):
      da = pytest.importorskip("dask.array")
      import numpy as np
  
      def random(**kwargs):
          assert len(kwargs) == ndeps
          return np.random.random((10, 10))
  
      trivial_deps = {
          f"k{i}": delayed(object(), name=f"object-{i}")
          for i in range(ndeps)
      }
      x = da.blockwise(
          random,
          "yx",
          new_axes={"y": (10,) * n_reducers, "x": (10,) * n_reducers},
          dtype=float,
          **trivial_deps,
      )
      graph = x.sum(axis=1, split_every=20)
      from dask.order import order
  
      if keep_self:
          # Keeping self adds a layer that cannot be fused
          dsk = collections_to_dsk([x, graph], optimize_graph=optimize)
      else:
          dsk = collections_to_dsk([graph], optimize_graph=optimize)
      dependencies, dependents = get_deps(dsk)
      # Verify assumptions
      before = len(dsk)
      before_dsk = dsk.copy()
      o = order(dsk)
      assert_topological_sort(dsk, o)
      assert before == len(o) == len(dsk)
      assert before_dsk == dsk
      # Verify assumptions (specifically that the reducers are sum-aggregate)
      assert ({"object", "sum", "sum-aggregate"}).issubset(
          {key_split(k) for k in o}
      )
      reducers = {k for k in o if key_split(k) == "sum-aggregate"}
      assert (p := max(diagnostics(dsk)[1])) <= n_reducers + ndeps, p
      # With optimize the metrics below change since there are many layers that
      # are being fused but the pressure above should already be a strong
      # indicator if something is wrong
      if optimize:
          for r in reducers:
              prios_deps = []
              for dep in dependencies[r]:
                  prios_deps.append(o[dep])
              assert max(prios_deps) - min(prios_deps) == (
                  len(dependencies[r]) - 1
              ) * (2 if keep_self else 1)
  
imports: from __future__ import annotations
  import inspect
  import pytest
  import dask
  from dask import delayed
  from dask.base import collections_to_dsk, key_split, visualize_dsk
  from dask.core import get_deps
  from dask.order import _connecting_to_roots, diagnostics, ndependencies, order
  from dask.utils_test import add, inc
label_desc: Use image node_id 41 for calling a global function with example usage: test_reduce_with_many_common_dependents(optimize, keep_ndeps, n_reducers) and returns: np