code: def test_generate_stream_graph_with_dependencies(self):
      python_file_dir = os.path.join(
          self.tempdir, "python_file_dir_" + str(uuid.uuid4())
      )
      os.mkdir(python_file_dir)
      python_file_path = os.path.join(
          python_file_dir, "test_stream_dependency_manage_lib.py"
      )
      with open(python_file_path, "w") as f:
          f.write("def add_two(a):\n    return a + 2")
      env = self.env
      env.add_python_file(python_file_path)
  
      def plus_two_map(value):
          from test_stream_dependency_manage_lib import add_two
  
          return value[0], add_two(value[1])
  
      def add_from_file(i):
          with open("data/data.txt", "r") as f:
              return i[0], i[1] + int(f.read())
  
      from_collection_source = env.from_collection(
          [("a", 0), ("b", 0), ("c", 1), ("d", 1), ("e", 2)],
          type_info=Types.ROW([Types.STRING(), Types.INT()]),
      )
      from_collection_source.name("From Collection")
      keyed_stream = from_collection_source.key_by(
          lambda x: x[1], key_type=Types.INT()
      )
  
      plus_two_map_stream = (
          keyed_stream.map(plus_two_map)
          .name("Plus Two Map")
          .set_parallelism(3)
      )
  
      add_from_file_map = plus_two_map_stream.map(add_from_file).name(
          "Add From File Map"
      )
  
      test_stream_sink = add_from_file_map.add_sink(
          self.test_sink
      ).name("Test Sink")
      test_stream_sink.set_parallelism(4)
  
      archive_dir_path = os.path.join(
          self.tempdir, "archive_" + str(uuid.uuid4())
      )
      os.mkdir(archive_dir_path)
      with open(os.path.join(archive_dir_path, "data.txt"), "w") as f:
          f.write("3")
      archive_file_path = shutil.make_archive(
          os.path.dirname(archive_dir_path), "zip", archive_dir_path
      )
      env.add_python_archive(archive_file_path, "data")
  
      nodes = eval(env.get_execution_plan())["nodes"]
  
      # The StreamGraph should be as bellow:
      # Source: From Collection -> _stream_key_by_map_operator ->
      # Plus Two Map -> Add From File Map -> Sink: Test Sink.
  
      # Source: From Collection and _stream_key_by_map_operator should have same parallelism.
      self.assertEqual(nodes[0]["parallelism"], nodes[1]["parallelism"])
  
      # The parallelism of Plus Two Map should be 3
      self.assertEqual(nodes[2]["parallelism"], 3)
  
      # The ship_strategy for Source: From Collection and _stream_key_by_map_operator should be
      # FORWARD
      self.assertEqual(
          nodes[1]["predecessors"][0]["ship_strategy"], "FORWARD"
      )
  
      # The ship_strategy for _keyed_stream_values_operator and Plus Two Map should be
      # HASH
      self.assertEqual(
          nodes[2]["predecessors"][0]["ship_strategy"], "HASH"
      )
  
      # The parallelism of Sink: Test Sink should be 4
      self.assertEqual(nodes[4]["parallelism"], 4)
  
      python_dependency_config = dict(
          get_gateway()
          .jvm.org.apache.flink.python.util.PythonDependencyUtils.configurePythonDependencies(
              env._j_stream_execution_environment.getCachedFiles(),
              env._j_stream_execution_environment.getConfiguration(),
          )
          .toMap()
      )
  
      # Make sure that user specified files and archives are correctly added.
      self.assertIsNotNone(
          python_dependency_config["python.internal.files-key-map"]
      )
      self.assertIsNotNone(
          python_dependency_config["python.internal.archives-key-map"]
      )
  
imports: import datetime
  import decimal
  import glob
  import os
  import shutil
  import tempfile
  import time
  import uuid
  from pyflink.common import Configuration, ExecutionConfig, RestartStrategies
  from pyflink.common.typeinfo import Types
  from pyflink.datastream import StreamExecutionEnvironment, CheckpointConfig, CheckpointingMode, MemoryStateBackend, TimeCharacteristic, SlotSharingGroup
  from pyflink.datastream.connectors.kafka import FlinkKafkaConsumer
  from pyflink.datastream.execution_mode import RuntimeExecutionMode
  from pyflink.datastream.formats.json import JsonRowDeserializationSchema
  from pyflink.datastream.functions import SourceFunction
  from pyflink.datastream.slot_sharing_group import MemorySize
  from pyflink.datastream.tests.test_util import DataStreamTestSinkFunction
  from pyflink.find_flink_home import _find_flink_source_root
  from pyflink.java_gateway import get_gateway
  from pyflink.table import DataTypes, StreamTableEnvironment, EnvironmentSettings
  from pyflink.testing.test_case_utils import PyFlinkTestCase, exec_insert_table
  from pyflink.util.java_utils import get_j_env_configuration
label_desc: Use image node_id 36 for calling the StreamExecutionEnvironmentTests obj's underlying member method code with example usage: obj.test_generate_stream_graph_with_dependencies() and returns: value,  add_two, i,  unknown