code: def test_notebook_job_with_more_configuration(sagemaker_session):
      """This test case is for more complex job configuration.
      1. a parent notebook file with %run magic to execute 'subfolder/sub.ipynb' and the
      simple.ipynb
      2. the notebook job has parameters used by parent and sub notebooks
      3. the environment variables are passed from step and init script
      4. pipeline variables are used for a sanity testing purpose
      5. sub notebook is uploaded as additional_dependencies
      6. init script will be used
      7. additional tags are also attached
      8. apply kms key
      """
      pipeline_name = name_from_base(PIPELINE_NAME, max_length=63)
      notebook_job_name = f"{NOTEBOOK_JOB_NAME_Prefix}-case2-configs"
  
      input_notebook = "tests/data/workflow/notebook_job_step/notebook1_happypath.ipynb"
      folder_with_sub_notebook = (
          "tests/data/workflow/notebook_job_step/subfolder"
      )
      simple_notebook_path = (
          "tests/data/workflow/notebook_job_step/simple.ipynb"
      )
      init_script = "tests/data/workflow/notebook_job_step/my-init.sh"
  
      # prepare parameter
      company_parameter = ParameterString(
          name="company", default_value="Amazon_FromParameter"
      )
      notebook_job_parameters = {
          "company": company_parameter,
          "company2": "Amazon2",
      }
  
      # prepare env vars
      env_parameter = ParameterString(
          name="env_key", default_value="EnvValueDefinedByParameter"
      )
      environment_variables = {
          "env_key": env_parameter,
          "env_key2": "env_value_abc2",
      }
  
      # prepare tags
      tags = {
          "company": company_parameter,
          "sagemaker:user-profile-name": "studio-user",
          "sagemaker:is-scheduling-notebook-job": "true",
      }
  
      # ParameterInteger for volume
      volume_size_parameter = ParameterInteger(
          name="volume_size", default_value=40
      )
  
      # ParameterBoolean
      encrypt_container_traffic_parameter = ParameterBoolean(
          name="encrypt_container_traffic", default_value=False
      )
  
      kms_key_id = kms_utils.get_or_create_kms_key(
          sagemaker_session, role_arn=get_role(sagemaker_session)
      )
  
      notebook_job_step = NotebookJobStep(
          name=STEP_NAME,
          display_name=DISPLAY_NAME,
          description=DESCRIPTION,
          notebook_job_name=notebook_job_name,
          image_uri=COSMOS_IMAGE_V0_URI,
          kernel_name=KERNEL_NAME,
          role=SAGE_MAKER_ROLE_NAME,
          input_notebook=input_notebook,
          initialization_script=init_script,
          additional_dependencies=[
              simple_notebook_path,
              folder_with_sub_notebook,
          ],
          parameters=notebook_job_parameters,
          environment_variables=environment_variables,
          tags=tags,
          volume_size=volume_size_parameter,
          encrypt_inter_container_traffic=encrypt_container_traffic_parameter,
          s3_kms_key=kms_key_id,
          volume_kms_key=kms_key_id,
      )
  
      # 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,
          parameters=[
              company_parameter,
              env_parameter,
              volume_size_parameter,
              encrypt_container_traffic_parameter,
          ],
          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.
          assert (
              job_description["AlgorithmSpecification"]["TrainingImage"]
              == COSMOS_IMAGE_V0_URI
          )
          assert SAGE_MAKER_ROLE_NAME in job_description["RoleArn"]
  
          # verify notebook job parameters
          assert (
              job_description["HyperParameters"]["company"]
              == "Amazon_FromParameter"
          )
          assert (
              job_description["HyperParameters"]["company2"]
              == "Amazon2"
          )
  
          # verify custom env variables
          assert (
              job_description["Environment"]["env_key"]
              == "EnvValueDefinedByParameter"
          )
          assert (
              job_description["Environment"]["env_key2"]
              == "env_value_abc2"
          )
  
          # verify passed in volume number
          assert (
              job_description["ResourceConfig"]["VolumeSizeInGB"] == 40
          )
  
          # verify passed in encrypt_container_traffic_parameter
          assert not job_description[
              "EnableInterContainerTrafficEncryption"
          ]
  
          # verify kms key
          assert (
              job_description["OutputDataConfig"]["KmsKeyId"]
              == kms_key_id
          )
          assert (
              job_description["ResourceConfig"]["VolumeKmsKeyId"]
              == kms_key_id
          )
  
          # verify tags
          tags_from_job = sagemaker_session.sagemaker_client.list_tags(
              ResourceArn=job_description["TrainingJobArn"]
          )["Tags"]
          tags_dict = _convert_to_flatten_dict(tags_from_job)
          # tags set by client code
          assert tags_dict["company"] == "Amazon_FromParameter"
          assert (
              tags_dict["sagemaker:user-profile-name"] == "studio-user"
          )
          assert (
              tags_dict["sagemaker:is-scheduling-notebook-job"]
              == "true"
          )
  
          # system tags
          assert tags_dict["sagemaker:is-studio-archived"] == "false"
          assert tags_dict["sagemaker:name"] == notebook_job_name
          assert (
              tags_dict["sagemaker:notebook-job-origin"]
              == "PIPELINE_STEP"
          )
  
          # verify the job output
          assert job_description["TrainingJobStatus"] == "Completed"
  
          output_s3_uri = s3_path_join(
              job_description["OutputDataConfig"]["S3OutputPath"],
              job_description["TrainingJobName"],
              "output",
              "output.tar.gz",
          )
          output_notebook_name = job_description["Environment"][
              "SM_OUTPUT_NOTEBOOK_NAME"
          ]
  
          def verify_notebook(cells):
              # verify the env variable set by the init script
              # logging for better troubleshooting
              logging.info(cells)
              assert NotebookUtils.search_output_of_cells(
                  cells,
                  "ParentNotebook: ENV_VAR_FROM_INIT_SCRIPT=FROM_INIT_SCRIPT",
              )
              assert NotebookUtils.search_output_of_cells(
                  cells,
                  "SubNotebook: ENV_VAR_FROM_INIT_SCRIPT=FROM_INIT_SCRIPT",
              )
  
              # verify the env from the step def
              assert NotebookUtils.search_output_of_cells(
                  cells,
                  "ParentNotebook: env_key=EnvValueDefinedByParameter",
              )
              assert NotebookUtils.search_output_of_cells(
                  cells,
                  "SubNotebook: env_key=EnvValueDefinedByParameter",
              )
  
              # verify parameters
              assert NotebookUtils.search_output_of_cells(
                  cells,
                  "ParentNotebook: company_is_Amazon_FromParameter",
              )
              assert NotebookUtils.search_output_of_cells(
                  cells, "SubNotebook: company_is_Amazon_FromParameter"
              )
  
              # verify the output of %run simple.ipynb
              assert NotebookUtils.search_output_of_cells(
                  cells, "aws-sagemaker"
              )
  
          _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 2 for calling a global function with example usage: test_notebook_job_with_more_configuration(sagemaker_session) without return types