code: def test_join_expansion(tmpdir, needed_outputs):
      global _sums
      global _sum_operands
      global _products
      tmpdir.chdir()
  
      # Clean up, just in case some other test modified them
      _products = []
      _sum_operands = []
      _sums = []
  
      prev_state = config.get("execution", "remove_unnecessary_outputs")
      config.set(
          "execution", "remove_unnecessary_outputs", needed_outputs
      )
      # Make the workflow.
      wf = pe.Workflow(name="test")
      # the iterated input node
      inputspec = pe.Node(
          IdentityInterface(fields=["n"]), name="inputspec"
      )
      inputspec.iterables = [("n", [1, 2])]
      # a pre-join node in the iterated path
      pre_join1 = pe.Node(IncrementInterface(), name="pre_join1")
      # another pre-join node in the iterated path
      pre_join2 = pe.Node(IncrementInterface(), name="pre_join2")
      # the join node
      join = pe.JoinNode(
          SumInterface(),
          joinsource="inputspec",
          joinfield="input1",
          name="join",
      )
      # an uniterated post-join node
      post_join1 = pe.Node(IncrementInterface(), name="post_join1")
      # a post-join node in the iterated path
      post_join2 = pe.Node(ProductInterface(), name="post_join2")
  
      wf.connect(
          [
              (inputspec, pre_join1, [("n", "input1")]),
              (pre_join1, pre_join2, [("output1", "input1")]),
              (pre_join1, post_join2, [("output1", "input2")]),
              (pre_join2, join, [("output1", "input1")]),
              (join, post_join1, [("output1", "input1")]),
              (join, post_join2, [("output1", "input1")]),
          ]
      )
      result = wf.run()
  
      # the two expanded pre-join predecessor nodes feed into one join node
      joins = [node for node in result.nodes() if node.name == "join"]
      assert (
          len(joins) == 1
      ), "The number of join result nodes is incorrect."
      # the expanded graph contains 2 * 2 = 4 iteration pre-join nodes, 1 join
      # node, 1 non-iterated post-join node and 2 * 1 iteration post-join nodes.
      # Nipype factors away the IdentityInterface.
      assert (
          len(result.nodes()) == 8
      ), "The number of expanded nodes is incorrect."
  
      # the join Sum result is (1 + 1 + 1) + (2 + 1 + 1)
      assert len(_sums) == 1, "The number of join outputs is incorrect"
      assert _sums[0] == 7, (
          "The join Sum output value is incorrect: %s." % _sums[0]
      )
      # the join input preserves the iterables input order
      assert _sum_operands[0] == [3, 4], (
          "The join Sum input is incorrect: %s." % _sum_operands[0]
      )
      # there are two iterations of the post-join node in the iterable path
      assert (
          len(_products) == 2
      ), "The number of iterated post-join outputs is incorrect"
      config.set("execution", "remove_unnecessary_outputs", prev_state)
  
imports: import pytest
  from ....None import config
  from ...None import engine
  from ....interfaces import base
  from ....interfaces.utility import IdentityInterface, Function, Merge
  from ....interfaces.base import traits, File
label_desc: Use image node_id 1 for calling a global function with example usage: test_join_expansion(tmpdir, needed_outputs) without return types