code: def test_happycase_minimum_input(sagemaker_session):
      pipeline_name = name_from_base(PIPELINE_NAME, max_length=63)
      input_notebook = (
          "tests/data/workflow/notebook_job_step/simple.ipynb"
      )
  
      notebook_job_step = NotebookJobStep(
          input_notebook=input_notebook,
          image_uri=COSMOS_IMAGE_V0_URI,
          kernel_name=KERNEL_NAME,
          role=SAGE_MAKER_ROLE_NAME,
      )
  
      # TODO - will be removed when cosmos image is ready
      notebook_job_step = (
          _config_manual_dependency_installation_short_term_workaround(
              notebook_job_step
          )
      )
  
      pipeline = Pipeline(
          name=pipeline_name,
          steps=[notebook_job_step],
          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_description = _get_training_job_details(
              execution_steps[0], sagemaker_session
          )
  
          # verify the training job is created consistent with input.
          training_job_name = job_description["TrainingJobName"]
          assert training_job_name.startswith("simple-ipynb-")
          assert (
              job_description["AlgorithmSpecification"]["TrainingImage"]
              == COSMOS_IMAGE_V0_URI
          )
          assert SAGE_MAKER_ROLE_NAME in job_description["RoleArn"]
  
          assert job_description["Environment"][
              "SM_INPUT_NOTEBOOK_NAME"
          ] == os.path.basename(input_notebook)
          assert (
              job_description["Environment"]["SM_KERNEL_NAME"]
              == KERNEL_NAME
          )
          assert job_description["TrainingJobStatus"] == "Completed"
  
          # verify the output notebook
          output_s3_uri = s3_path_join(
              job_description["OutputDataConfig"]["S3OutputPath"],
              training_job_name,
              "output",
              "output.tar.gz",
          )
          output_notebook_name = job_description["Environment"][
              "SM_OUTPUT_NOTEBOOK_NAME"
          ]
  
          def verify_notebook_for_happy_case(cells):
              # happy case is using tests/data/workflow/notebook_job_step/simple.ipynb which print
              # aws-sagemaker.
              assert NotebookUtils.search_output_of_cells(
                  cells, "aws-sagemaker"
              )
  
          _download_and_verify_output_notebook(
              output_s3_uri,
              output_notebook_name,
              verify_notebook_for_happy_case,
              sagemaker_session,
          )
  
          # upsert and run a second time
          pipeline.upsert(get_role(sagemaker_session))
          execution_steps = _start_and_verify_execution_with_retry(
              pipeline=pipeline,
              parameters={},
          )
  
          job_description = _get_training_job_details(
              execution_steps[0], sagemaker_session
          )
          assert job_description["TrainingJobStatus"] == "Completed"
      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 1 for calling a global function with example usage: test_happycase_minimum_input(sagemaker_session) without return types