[Core] Fix rpdb.py for windows and encoding failures, adjust tests (#…

…48117)

Signed-off-by: Matti Picus <matti.picus@gmail.com>

python/ray/tests/test_ray_debugger.py

@@ -54,7 +54,6 @@ def f():
     ray.get(result)
 
 
-@pytest.mark.skipif(platform.system() == "Windows", reason="Failing on Windows.")
 def test_ray_debugger_commands(shutdown_only):
     ray.init(num_cpus=2, runtime_env={"env_vars": {"RAY_DEBUG": "legacy"}})
 
@@ -77,7 +76,12 @@ def f():
 
     # Make sure that calling "continue" in the debugger
     # gives back control to the debugger loop:
-    p = pexpect.spawn("ray debug")
+    if sys.platform == "win32":
+        from pexpect.popen_spawn import PopenSpawn
+
+        p = PopenSpawn("ray debug", encoding="utf-8")
+    else:
+        p = pexpect.spawn("ray debug")
     p.expect("Enter breakpoint index or press enter to refresh: ")
     p.sendline("0")
     p.expect("-> ray.util.pdb.set_trace()")
@@ -94,7 +98,6 @@ def f():
     ray.get([result1, result2])
 
 
-@pytest.mark.skipif(platform.system() == "Windows", reason="Failing on Windows.")
 def test_ray_debugger_stepping(shutdown_only):
     os.environ["RAY_DEBUG"] = "legacy"
     ray.init(num_cpus=1, runtime_env={"env_vars": {"RAY_DEBUG": "legacy"}})
@@ -120,7 +123,12 @@ def f():
         > 0
     )
 
-    p = pexpect.spawn("ray debug")
+    if sys.platform == "win32":
+        from pexpect.popen_spawn import PopenSpawn
+
+        p = PopenSpawn("ray debug", encoding="utf-8")
+    else:
+        p = pexpect.spawn("ray debug")
     p.expect("Enter breakpoint index or press enter to refresh: ")
     p.sendline("0")
     p.expect("-> x = g.remote()")
@@ -134,7 +142,6 @@ def f():
     ray.get(result)
 
 
-@pytest.mark.skipif(platform.system() == "Windows", reason="Failing on Windows.")
 def test_ray_debugger_recursive(shutdown_only):
     os.environ["RAY_DEBUG"] = "legacy"
     ray.init(num_cpus=1, runtime_env={"env_vars": {"RAY_DEBUG": "legacy"}})
@@ -159,6 +166,12 @@ def fact(n):
     )
 
     p = pexpect.spawn("ray debug")
+    if sys.platform == "win32":
+        from pexpect.popen_spawn import PopenSpawn
+
+        p = PopenSpawn("ray debug", encoding="utf-8")
+    else:
+        p = pexpect.spawn("ray debug")
     p.expect("Enter breakpoint index or press enter to refresh: ")
     p.sendline("0")
     p.expect("(Pdb)")
@@ -177,7 +190,6 @@ def fact(n):
     ray.get(result)
 
 
-@pytest.mark.skipif(platform.system() == "Windows", reason="Failing on Windows.")
 def test_job_exit_cleanup(ray_start_regular):
     address = ray_start_regular["address"]
 
@@ -217,7 +229,12 @@ def one_active_session():
 
     # Start the debugger. This should clean up any existing sessions that
     # belong to dead jobs.
-    p = pexpect.spawn("ray debug")  # noqa:F841
+    if sys.platform == "win32":
+        from pexpect.popen_spawn import PopenSpawn
+
+        p = PopenSpawn("ray debug", encoding="utf-8")  # noqa: F841
+    else:
+        p = pexpect.spawn("ray debug")  # noqa:F841
 
     def no_active_sessions():
         return not len(
@@ -229,7 +246,9 @@ def no_active_sessions():
     wait_for_condition(no_active_sessions)
 
 
-@pytest.mark.skipif(platform.system() == "Windows", reason="Failing on Windows.")
+@pytest.mark.skipif(
+    platform.system() == "Windows", reason="Windows does not print '--address' on init"
+)
 @pytest.mark.parametrize("ray_debugger_external", [False, True])
 def test_ray_debugger_public(shutdown_only, call_ray_stop_only, ray_debugger_external):
     redis_substring_prefix = "--address='"

python/ray/util/rpdb.py

@@ -43,7 +43,9 @@ def __init__(self, connection):
         self.flush = fh.flush
         self.fileno = fh.fileno
         if hasattr(fh, "encoding"):
-            self._send = lambda data: connection.sendall(data.encode(fh.encoding))
+            self._send = lambda data: connection.sendall(
+                data.encode(fh.encoding, errors="replace")
+            )
         else:
             self._send = connection.sendall
 
@@ -117,7 +119,7 @@ def listen(self):
         self._listen_socket.listen(1)
         connection, address = self._listen_socket.accept()
         if not self._quiet:
-            _cry("RemotePdb accepted connection from %s." % repr(address))
+            _cry(f"RemotePdb accepted connection from {address}")
         self.handle = _LF2CRLF_FileWrapper(connection)
         Pdb.__init__(
             self,
@@ -342,16 +344,28 @@ def _post_mortem():
 
 
 def _connect_pdb_client(host, port):
+    if sys.platform == "win32":
+        import msvcrt
     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     s.connect((host, port))
 
     while True:
         # Get the list of sockets which are readable.
-        read_sockets, write_sockets, error_sockets = select.select(
-            [sys.stdin, s], [], []
-        )
+        if sys.platform == "win32":
+            ready_to_read = select.select([s], [], [], 1)[0]
+            if msvcrt.kbhit():
+                ready_to_read.append(sys.stdin)
+            if not ready_to_read and not sys.stdin.isatty():
+                # in tests, when using pexpect, the pipe makes
+                # the msvcrt.kbhit() trick fail. Assume we are waiting
+                # for stdin, since this will block waiting for input
+                ready_to_read.append(sys.stdin)
+        else:
+            ready_to_read, write_sockets, error_sockets = select.select(
+                [sys.stdin, s], [], []
+            )
 
-        for sock in read_sockets:
+        for sock in ready_to_read:
             if sock == s:
                 # Incoming message from remote debugger.
                 data = sock.recv(4096)