Make test_utils.py use pipes to avoid file access conflicts on Windows (#9072)

Co-authored-by: Mehrdad <noreply@github.com>

python/ray/serve/tests/test_persistence.py

@@ -1,8 +1,5 @@
-import os
-import subprocess
-import tempfile
-
 import ray
+import ray.test_utils
 from ray import serve
 
 
@@ -20,20 +17,11 @@ def driver(flask_request):
 serve.create_backend("driver", driver)
 serve.create_endpoint("driver", backend="driver", route="/driver")
 """.format(ray.worker._global_node._redis_address)
-
-    with tempfile.NamedTemporaryFile(mode="w", delete=False) as f:
-        path = f.name
-        f.write(script)
-
-    proc = subprocess.Popen(["python", path])
-    return_code = proc.wait(timeout=10)
-    assert return_code == 0
+    ray.test_utils.run_string_as_driver(script)
 
     handle = serve.get_handle("driver")
     assert ray.get(handle.remote()) == "OK!"
 
-    os.remove(path)
-
 
 if __name__ == "__main__":
     import sys

python/ray/test_utils.py

@@ -5,7 +5,6 @@ import fnmatch
 import os
 import subprocess
 import sys
-import tempfile
 import time
 import socket
 
@@ -104,13 +103,17 @@ def run_string_as_driver(driver_script):
     Returns:
         The script's output.
     """
-    # Save the driver script as a file so we can call it using subprocess.
-    with tempfile.NamedTemporaryFile() as f:
-        f.write(driver_script.encode("ascii"))
-        f.flush()
-        out = ray.utils.decode(
-            subprocess.check_output(
-                [sys.executable, f.name], stderr=subprocess.STDOUT))
+    proc = subprocess.Popen(
+        [sys.executable, "-"],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT)
+    with proc:
+        output = proc.communicate(driver_script.encode("ascii"))[0]
+        if proc.returncode:
+            raise subprocess.CalledProcessError(proc.returncode, proc.args,
+                                                output, proc.stderr)
+        out = ray.utils.decode(output)
     return out
 
 
@@ -123,16 +126,21 @@ def run_string_as_driver_nonblocking(driver_script):
     Returns:
         A handle to the driver process.
     """
-    # Save the driver script as a file so we can call it using subprocess. We
-    # do not delete this file because if we do then it may get removed before
-    # the Python process tries to run it.
-    with tempfile.NamedTemporaryFile(delete=False) as f:
-        f.write(driver_script.encode("ascii"))
-        f.flush()
-        return subprocess.Popen(
-            [sys.executable, f.name],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE)
+    script = "; ".join([
+        "import sys",
+        "script = sys.stdin.read()",
+        "sys.stdin.close()",
+        "del sys",
+        "exec(\"del script\\n\" + script)",
+    ])
+    proc = subprocess.Popen(
+        [sys.executable, "-c", script],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE)
+    proc.stdin.write(driver_script.encode("ascii"))
+    proc.stdin.close()
+    return proc
 
 
 def flat_errors():

python/ray/tune/tests/test_progress_reporter.py

@@ -1,11 +1,10 @@
 import pytest
 import collections
-import subprocess
-import tempfile
 import os
 import unittest
 from unittest.mock import MagicMock, Mock
 
+from ray.test_utils import run_string_as_driver
 from ray.tune.trial import Trial
 from ray.tune.progress_reporter import (CLIReporter, _fair_filter_trials,
                                         trial_progress_str)
@@ -284,19 +283,15 @@ class ProgressReporterTest(unittest.TestCase):
     def testEndToEndReporting(self):
         try:
             os.environ["_TEST_TUNE_TRIAL_UUID"] = "xxxxx"
-            with tempfile.NamedTemporaryFile(suffix=".py") as f:
-                f.write(END_TO_END_COMMAND.encode("utf-8"))
-                f.flush()
-                output = subprocess.check_output(["python3", f.name])
-                output = output.decode("utf-8")
-                try:
-                    assert EXPECTED_END_TO_END_START in output
-                    assert EXPECTED_END_TO_END_END in output
-                except Exception:
-                    print("*** BEGIN OUTPUT ***")
-                    print(output)
-                    print("*** END OUTPUT ***")
-                    raise
+            output = run_string_as_driver(END_TO_END_COMMAND)
+            try:
+                assert EXPECTED_END_TO_END_START in output
+                assert EXPECTED_END_TO_END_END in output
+            except Exception:
+                print("*** BEGIN OUTPUT ***")
+                print(output)
+                print("*** END OUTPUT ***")
+                raise
         finally:
             del os.environ["_TEST_TUNE_TRIAL_UUID"]