[Runtime env] Add a test to make sure resource deadlock message is not printed when waiting for workers (#19870)

* ip

* Add a runtime env resource deadlock msg test

* Fix a bug

* Skip on windows

python/ray/tests/test_runtime_env.py

@@ -7,7 +7,7 @@ from pathlib import Path
 
 import ray
 from ray.exceptions import RuntimeEnvSetupError
-from ray._private.test_utils import wait_for_condition
+from ray._private.test_utils import wait_for_condition, get_error_message
 from ray._private.utils import (get_wheel_filename, get_master_wheel_url,
                                 get_release_wheel_url)
 
@@ -240,6 +240,29 @@ def test_no_spurious_worker_startup(shutdown_only):
     assert got_num_workers, "failed to read num workers for 10 seconds"
 
 
+@pytest.fixture
+def runtime_env_local_dev_env_var():
+    os.environ["RAY_RUNTIME_ENV_LOCAL_DEV_MODE"] = "1"
+    yield
+    del os.environ["RAY_RUNTIME_ENV_LOCAL_DEV_MODE"]
+
+
+@pytest.mark.skipif(
+    sys.platform == "win32", reason="runtime_env unsupported on Windows.")
+def test_runtime_env_no_spurious_resource_deadlock_msg(
+        runtime_env_local_dev_env_var, ray_start_regular, error_pubsub):
+    p = error_pubsub
+
+    @ray.remote(runtime_env={"pip": ["tensorflow", "torch"]})
+    def f():
+        pass
+
+    # Check no warning printed.
+    ray.get(f.remote())
+    errors = get_error_message(p, 5, ray.ray_constants.RESOURCE_DEADLOCK_ERROR)
+    assert len(errors) == 0
+
+
 @pytest.fixture
 def set_agent_failure_env_var():
     os.environ["_RAY_AGENT_FAILING"] = "1"