code: def test_environment_variables_complex(shutdown_only):
      ray.init(
          runtime_env={
              "env_vars": {
                  "a": "job_a",
                  "b": "job_b",
                  "z": "job_z",
              }
          }
      )
  
      @ray.remote
      def get_env(key):
          return os.environ.get(key)
  
      @ray.remote
      class NestedEnvGetter:
          def get(self, key):
              return os.environ.get(key)
  
          def get_task(self, key):
              return ray.get(get_env.remote(key))
  
      @ray.remote
      class EnvGetter:
          def get(self, key):
              return os.environ.get(key)
  
          def get_task(self, key):
              return ray.get(get_env.remote(key))
  
          def nested_get(self, key):
              aa = NestedEnvGetter.options(
                  runtime_env={
                      "env_vars": {
                          "c": "e",
                          "d": "dd",
                      }
                  }
              ).remote()
              return ray.get(aa.get.remote(key))
  
      a = EnvGetter.options(
          runtime_env={
              "env_vars": {
                  "a": "b",
                  "c": "d",
              }
          }
      ).remote()
  
      assert ray.get(a.get.remote("a")) == "b"
      assert ray.get(a.get_task.remote("a")) == "b"
      assert ray.get(a.nested_get.remote("a")) == "b"
      assert ray.get(a.nested_get.remote("c")) == "e"
      assert ray.get(a.nested_get.remote("d")) == "dd"
      assert (
          ray.get(
              get_env.options(
                  runtime_env={
                      "env_vars": {
                          "a": "b",
                      }
                  }
              ).remote("a")
          )
          == "b"
      )
  
      assert ray.get(a.get.remote("z")) == "job_z"
      assert ray.get(a.get_task.remote("z")) == "job_z"
      assert ray.get(a.nested_get.remote("z")) == "job_z"
      assert (
          ray.get(
              get_env.options(
                  runtime_env={
                      "env_vars": {
                          "a": "b",
                      }
                  }
              ).remote("z")
          )
          == "job_z"
      )
  
file_path: ray/python/ray/tests/test_runtime_env_env_vars.py
imports: import os
  import sys
  import pytest
  import ray
label_desc: Use image node_id 5 for calling a global function with example usage: test_environment_variables_complex(shutdown_only) and returns: os, os, ray, os, ray, ray