code: def test_source_dirs(tmpdir, sagemaker_local_session):
      source_dir = os.path.join(DATA_DIR, "pytorch_source_dirs")
      lib = os.path.join(str(tmpdir), "alexa.py")
  
      with open(lib, "w") as f:
          f.write("def question(to_anything): return 42")
  
      # TODO: fails on newer versions of pytorch in call to np.load(BytesIO(stream.read()))
      # "ValueError: Cannot load file containing pickled data when allow_pickle=False"
      estimator = PyTorch(
          entry_point="train.py",
          role="SageMakerRole",
          source_dir=source_dir,
          dependencies=[lib],
          framework_version="0.4",  # hard-code to last known good pytorch for now (see TODO above)
          py_version="py3",
          instance_count=1,
          instance_type="local",
          sagemaker_session=sagemaker_local_session,
      )
      estimator.fit()
  
      # endpoint tests all use the same port, so we use this lock to prevent concurrent execution
      with lock.lock():
          try:
              predictor = estimator.deploy(
                  initial_instance_count=1, instance_type="local"
              )
              predict_response = predictor.predict([7])
              assert predict_response == [49]
          finally:
              predictor.delete_endpoint()
  
imports: from __future__ import absolute_import
  import os
  import pytest
  import tests.integ.lock
  from sagemaker.pytorch.estimator import PyTorch
  from tests.integ import DATA_DIR
label_desc: Use image node_id 1 for calling a global function with example usage: test_source_dirs(tmpdir, sagemaker_local_session) without return types