code: def test_notebook_job_dependencies_and_outputs(sagemaker_session):
      pipeline_name = name_from_base(PIPELINE_NAME, max_length=63)
      notebook_job_name_base = (
          f"{NOTEBOOK_JOB_NAME_Prefix}-case3-outputs"
      )
  
      tags = {
          "sagemaker:user-profile-name": "studio-user",
          "sagemaker:is-scheduling-notebook-job": "true",
      }
  
      input_notebook = (
          "tests/data/workflow/notebook_job_step/simple.ipynb"
      )
      notebook_job_step1 = NotebookJobStep(
          name=f"{STEP_NAME}-1",
          display_name=DISPLAY_NAME,
          description=DESCRIPTION,
          notebook_job_name=f"{notebook_job_name_base}-a",
          input_notebook=input_notebook,
          image_uri=COSMOS_IMAGE_V0_URI,
          kernel_name=KERNEL_NAME,
          role=SAGE_MAKER_ROLE_NAME,
          tags=tags,
      )
  
      # TODO - will be removed when cosmos image is ready
      notebook_job_step1 = (
          _config_manual_dependency_installation_short_term_workaround(
              notebook_job_step1
          )
      )
  
      input_notebook2 = (
          "tests/data/workflow/notebook_job_step/step2_notebook.ipynb"
      )
      notebook_job_step2 = NotebookJobStep(
          name=f"{STEP_NAME}-2",
          display_name=DISPLAY_NAME,
          description=DESCRIPTION,
          notebook_job_name=f"{notebook_job_name_base}-b",
          input_notebook=input_notebook2,
          image_uri=COSMOS_IMAGE_V0_URI,
          kernel_name=KERNEL_NAME,
          role=SAGE_MAKER_ROLE_NAME,
          parameters={
              "step1_JobName": notebook_job_step1.properties.ComputingJobName,
              "step1_JobStatus": notebook_job_step1.properties.ComputingJobStatus,
              "step1_NotebookJobInput": notebook_job_step1.properties.NotebookJobInputLocation,
              "step1_NotebookJobOutput": notebook_job_step1.properties.NotebookJobOutputLocationPrefix,
              "step1_InputNotebookName": notebook_job_step1.properties.InputNotebookName,
              "step1_OutputNotebookName": notebook_job_step1.properties.OutputNotebookName,
          },
          tags=tags,
      )
      notebook_job_step2.add_depends_on([notebook_job_step1])
  
      # TODO - will be removed when cosmos image is ready
      notebook_job_step2 = (
          _config_manual_dependency_installation_short_term_workaround(
              notebook_job_step2
          )
      )
  
      pipeline = Pipeline(
          name=pipeline_name,
          steps=[notebook_job_step1, notebook_job_step2],
          sagemaker_session=sagemaker_session,
      )
      logging.info(
          f"Notebook job step test pipeline definition: {pipeline.definition()}"
      )
  
      try:
          pipeline.create(get_role(sagemaker_session))
          execution_steps = _start_and_verify_execution_with_retry(
              pipeline=pipeline,
              parameters={},
          )
  
          job_description1 = None
          job_description2 = None
          for execution_step in execution_steps:
              if execution_step["StepName"] == f"{STEP_NAME}-1":
                  job_description1 = _get_training_job_details(
                      execution_step, sagemaker_session
                  )
              elif execution_step["StepName"] == f"{STEP_NAME}-2":
                  job_description2 = _get_training_job_details(
                      execution_step, sagemaker_session
                  )
  
          step1_job_name = job_description1["TrainingJobName"]
          step1_job_status = job_description1["TrainingJobStatus"]
  
          step1_job_input = job_description1["InputDataConfig"][0][
              "DataSource"
          ]["S3DataSource"]["S3Uri"]
          step1_job_input_notebook = job_description1["Environment"][
              "SM_INPUT_NOTEBOOK_NAME"
          ]
          step1_job_output = job_description1["OutputDataConfig"][
              "S3OutputPath"
          ]
          step1_job_output_notebook = job_description1["Environment"][
              "SM_OUTPUT_NOTEBOOK_NAME"
          ]
  
          # verify output notebook of step 2
          output_s3_uri = s3_path_join(
              job_description2["OutputDataConfig"]["S3OutputPath"],
              job_description2["TrainingJobName"],
              "output",
              "output.tar.gz",
          )
          output_notebook_name = job_description2["Environment"][
              "SM_OUTPUT_NOTEBOOK_NAME"
          ]
  
          def verify_notebook(cells):
              # logging for better troubleshooting
              logging.info(f"step1_JobName={step1_job_name}")
              logging.info(f"step1_JobStatus={step1_job_status}")
              logging.info(f"step1_NotebookJobInput={step1_job_input}")
              logging.info(
                  f"step1_InputNotebookName={step1_job_input_notebook}"
              )
              logging.info(
                  f"step1_NotebookJobOutput={step1_job_output}"
              )
              logging.info(
                  f"step1_OutputNotebookName={step1_job_output_notebook}"
              )
              logging.info(f"cells: {cells}")
  
              assert NotebookUtils.search_output_of_cells(
                  cells, f"step1_JobName={step1_job_name}"
              )
              assert NotebookUtils.search_output_of_cells(
                  cells, f"step1_JobStatus={step1_job_status}"
              )
              assert NotebookUtils.search_output_of_cells(
                  cells, f"step1_NotebookJobInput={step1_job_input}"
              )
              assert NotebookUtils.search_output_of_cells(
                  cells,
                  f"step1_InputNotebookName={step1_job_input_notebook}",
              )
              assert NotebookUtils.search_output_of_cells(
                  cells, f"step1_NotebookJobOutput={step1_job_output}"
              )
              assert NotebookUtils.search_output_of_cells(
                  cells,
                  f"step1_OutputNotebookName={step1_job_output_notebook}",
              )
  
          _download_and_verify_output_notebook(
              output_s3_uri,
              output_notebook_name,
              verify_notebook,
              sagemaker_session,
          )
      finally:
          try:
              pipeline.delete()
          except Exception as error:
              logging.error(error)
  
imports: from __future__ import absolute_import
  import os.path
  import tarfile
  import logging
  import nbformat
  import pytest
  from sagemaker import get_execution_role
  from sagemaker.s3 import S3Downloader
  from sagemaker.s3_utils import s3_path_join
  from sagemaker.utils import name_from_base
  from sagemaker.workflow import ParameterString
  from sagemaker.workflow.notebook_job_step import NotebookJobStep
  from sagemaker.workflow.parameters import ParameterInteger, ParameterBoolean
  from sagemaker.workflow.pipeline import Pipeline
  from sagemaker.utils import _tmpdir
  from tests.integ.retry import retries
  from tests.integ.sagemaker.workflow.helpers import wait_pipeline_execution
  from tests.integ import kms_utils
  from sagemaker.workflow.function_step import step
label_desc: Use image node_id 3 for calling a global function with example usage: test_notebook_job_dependencies_and_outputs(sagemaker_session) without return types