code: def test_pipeline_graph_with_mixed_step_type_acyclic(
      sagemaker_session_mock, job_config
  ):
      delayed_return_a = step(
          lambda x, y: x + y,
          name="DRA",
          image_uri="image",
          instance_type="ml.m4.xlarge",
      )(1, 2)
  
      delayed_return_f = step(
          lambda x, y: x + y,
          name="DRF",
          image_uri="image",
          instance_type="ml.m4.xlarge",
      )(1, 2)
  
      custom_step_collection = CustomStepCollection(
          name="TestStepCollection", depends_on=[delayed_return_a]
      )
  
      custom_step = CustomStep(
          name="TestStep",
          input_data=custom_step_collection.properties.AlgorithmSpecification.AlgorithmName,
      )
  
      if_step = CustomStepCollection(
          name="IfStepCollection", depends_on=[delayed_return_a]
      )
      else_dr_1 = step(
          lambda x, y: x + y,
          name="ElseDR1",
          image_uri="image",
          instance_type="ml.m4.xlarge",
      )(1, 2)
  
      else_dr_2 = step(
          lambda x, y: x + y,
          name="ElseDR2",
          image_uri="image",
          instance_type="ml.m4.xlarge",
      )(1, 2)
      get_step(else_dr_2).add_depends_on([else_dr_1, delayed_return_f])
  
      param = ParameterInteger(name="MyInt", default_value=2)
      condition_step = ConditionStep(
          name="CondStep",
          conditions=[ConditionEquals(left=param, right=1)],
          depends_on=[custom_step],
          if_steps=[if_step],
          else_steps=[else_dr_2],  # else_dr_1 should be auto fetched
      )
      condition_step.add_depends_on([delayed_return_a])
      # check the extended condition sub step list
      assert len(condition_step.if_steps) == 1
      assert len(condition_step.else_steps) == 1
  
      delayed_return_b = step(
          lambda x, y: x + y,
          name="DRB",
          image_uri="image",
          instance_type="ml.m4.xlarge",
      )(1, 2)
  
      get_step(delayed_return_b).add_depends_on([if_step])
  
      delayed_return_c = step(
          lambda x, y: x + y,
          name="DRC",
          image_uri="image",
          instance_type="ml.m4.xlarge",
      )(1, 2)
      get_step(delayed_return_c).add_depends_on([delayed_return_b])
  
      custom_step_1 = CustomStep(name="TestStep1")
      delayed_return_d = step(
          lambda x, y: x + y,
          name="DRD",
          image_uri="image",
          instance_type="ml.m4.xlarge",
      )(1, 2)
      get_step(delayed_return_d).add_depends_on([custom_step_1])
      custom_step_2 = CustomStep(
          name="TestStep2", depends_on=[delayed_return_d]
      )
  
      delayed_return_e = step(
          lambda x, y: x + y,
          name="DRE",
          image_uri="image",
          instance_type="ml.m4.xlarge",
      )(1, 2)
      get_step(delayed_return_e).add_depends_on([delayed_return_f])
  
      pipeline = Pipeline(
          name="MyPipeline",
          steps=[
              # delayed_return_a, # can be auto fetched
              # custom_step_collection, # this regular step can be auto fetched
              # custom_step, # this regular step is an upstream step, can be auto fetched
              condition_step,
              delayed_return_c,
              # custom_step_1, # this regular step is an upstream step, can be auto fetched
              delayed_return_d,
              custom_step_2,
              delayed_return_e,
          ],
          sagemaker_session=sagemaker_session_mock,
      )
  
      assert len(pipeline.steps) == 5
  
      pipeline_graph = PipelineGraph.from_pipeline(pipeline)
      adjacency_list = pipeline_graph.adjacency_list
      assert ordered(adjacency_list) == ordered(
          {
              "DRA": [
                  "TestStepCollection-0",
                  "CondStep",
                  "IfStepCollection-0",
              ],
              "CondStep": [
                  "IfStepCollection-0",
                  "IfStepCollection-1",
                  "ElseDR2",
              ],
              # ElseDR1 is auto fetched and put outside the condition step
              "ElseDR1": ["ElseDR2"],
              "ElseDR2": [],
              "IfStepCollection-0": ["DRB", "IfStepCollection-1"],
              "IfStepCollection-1": ["DRB"],
              "TestStep": ["CondStep"],
              "TestStepCollection-0": ["TestStepCollection-1"],
              "TestStepCollection-1": ["TestStep"],
              "DRB": ["DRC"],
              "DRC": [],
              "TestStep1": ["DRD"],
              "DRD": ["TestStep2"],
              "TestStep2": [],
              "DRE": [],
              # DRF is auto fetched and put outside the condition step
              "DRF": ["DRE", "ElseDR2"],
          }
      )
      _verify_pipeline_graph_traversal(pipeline_graph)
  
imports: from __future__ import absolute_import
  import pytest
  from sagemaker.remote_function.job import _JobSettings
  from sagemaker.workflow.function_step import step
  from sagemaker.workflow.step_outputs import get_step
  from sagemaker.workflow.pipeline import Pipeline, PipelineGraph
  from sagemaker.workflow.condition_step import ConditionStep
  from sagemaker.workflow.conditions import ConditionEquals, ConditionIn, ConditionNot, ConditionOr, ConditionGreaterThan
  from sagemaker.workflow.execution_variables import ExecutionVariables
  from sagemaker.workflow.parameters import ParameterInteger, ParameterString
  from sagemaker.workflow._steps_compiler import CompiledStep
  from tests.unit.sagemaker.workflow.helpers import ordered, CustomStep, CustomStepCollection
label_desc: Use image node_id 16 for calling a global function with example usage: test_pipeline_graph_with_mixed_step_type_acyclic(sagemaker_session_mock, job_config) without return types