diff --git a/python/ray/tests/test_runtime_env_working_dir.py b/python/ray/tests/test_runtime_env_working_dir.py index 889cb958f..93e203c26 100644 --- a/python/ray/tests/test_runtime_env_working_dir.py +++ b/python/ray/tests/test_runtime_env_working_dir.py @@ -72,17 +72,24 @@ def test_lazy_reads(start_cluster, tmp_working_dir, option: str): """ cluster, address = start_cluster - if option == "failure": - # Don't pass the files at all, so it should fail! - ray.init(address) - elif option == "working_dir": - ray.init(address, runtime_env={"working_dir": tmp_working_dir}) - elif option == "py_modules": - ray.init( - address, - runtime_env={ - "py_modules": [str(Path(tmp_working_dir) / "test_module")] - }) + def call_ray_init(): + if option == "failure": + # Don't pass the files at all, so it should fail! + ray.init(address) + elif option == "working_dir": + ray.init(address, runtime_env={"working_dir": tmp_working_dir}) + elif option == "py_modules": + ray.init( + address, + runtime_env={ + "py_modules": [str(Path(tmp_working_dir) / "test_module")] + }) + + call_ray_init() + + def reinit(): + ray.shutdown() + call_ray_init() @ray.remote def test_import(): @@ -95,6 +102,8 @@ def test_lazy_reads(start_cluster, tmp_working_dir, option: str): else: assert ray.get(test_import.remote()) == 1 + reinit() + @ray.remote def test_read(): return open("hello").read() @@ -105,6 +114,8 @@ def test_lazy_reads(start_cluster, tmp_working_dir, option: str): elif option == "working_dir": assert ray.get(test_read.remote()) == "world" + reinit() + @ray.remote class Actor: def test_import(self): @@ -135,17 +146,26 @@ def test_captured_import(start_cluster, tmp_working_dir, option: str): """ cluster, address = start_cluster - if option == "failure": - # Don't pass the files at all, so it should fail! - ray.init(address) - elif option == "working_dir": - ray.init(address, runtime_env={"working_dir": tmp_working_dir}) - elif option == "py_modules": - ray.init( - address, - runtime_env={ - "py_modules": [os.path.join(tmp_working_dir, "test_module")] - }) + def call_ray_init(): + if option == "failure": + # Don't pass the files at all, so it should fail! + ray.init(address) + elif option == "working_dir": + ray.init(address, runtime_env={"working_dir": tmp_working_dir}) + elif option == "py_modules": + ray.init( + address, + runtime_env={ + "py_modules": [ + os.path.join(tmp_working_dir, "test_module") + ] + }) + + call_ray_init() + + def reinit(): + ray.shutdown() + call_ray_init() # Import in the driver. sys.path.insert(0, tmp_working_dir) @@ -161,6 +181,8 @@ def test_captured_import(start_cluster, tmp_working_dir, option: str): else: assert ray.get(test_import.remote()) == 1 + reinit() + @ray.remote class Actor: def test_import(self): @@ -197,6 +219,10 @@ def test_empty_working_dir(start_cluster): a = A.remote() assert len(ray.get(a.listdir.remote())) == 0 + # Test that we can reconnect with no errors + ray.shutdown() + ray.init(address, runtime_env={"working_dir": working_dir}) + @pytest.mark.parametrize("option", ["working_dir", "py_modules"]) @pytest.mark.skipif(sys.platform == "win32", reason="Fail to create temp dir.") diff --git a/release/.buildkite/build_pipeline.py b/release/.buildkite/build_pipeline.py index c7d51bbcb..5faa766d8 100644 --- a/release/.buildkite/build_pipeline.py +++ b/release/.buildkite/build_pipeline.py @@ -189,8 +189,7 @@ NIGHTLY_TESTS = { "serve_cluster_fault_tolerance", ], "~/ray/release/runtime_env_tests/runtime_env_tests.yaml": [ - "rte_many_tasks_actors", - "wheel_urls", + "rte_many_tasks_actors", "wheel_urls", "rte_ray_client" ], } diff --git a/release/runtime_env_tests/workloads/rte_ray_client.py b/release/runtime_env_tests/workloads/rte_ray_client.py index 3f4831262..d32556e72 100644 --- a/release/runtime_env_tests/workloads/rte_ray_client.py +++ b/release/runtime_env_tests/workloads/rte_ray_client.py @@ -69,7 +69,8 @@ if __name__ == "__main__": addr = os.environ.get("RAY_ADDRESS") job_name = os.environ.get("RAY_JOB_NAME", "rte_ray_client") - for use_working_dir in [True, False]: + # Test reconnecting to the same cluster multiple times. + for use_working_dir in [True, True, False, False]: with tempfile.TemporaryDirectory() as tmpdir: runtime_env = {"working_dir": tmpdir} if use_working_dir else None print("Testing with use_working_dir=" + str(use_working_dir))