| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
1 parent c5fcdb4 commit e1384cf
1 file changed
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -22,6 +22,7 @@ | |||
| 22 | 22 | import sysconfig | |
| 23 | 23 | import select | |
| 24 | 24 | import shutil | |
| 25 | + import socket | ||
| 25 | 26 | import threading | |
| 26 | 27 | import gc | |
| 27 | 28 | import textwrap | |
@@ -1044,19 +1045,49 @@ def test_communicate_timeout_large_input(self): | |||
| 1044 | 1045 | # On Windows, stdin writing must also honor the timeout rather than | |
| 1045 | 1046 | # blocking indefinitely when the pipe buffer fills. | |
| 1046 | 1047 | ||
| 1047 | - # Input larger than typical pipe buffer (4-64KB on Windows) | ||
| 1048 | - input_data = b"x" * (128 * 1024) | ||
| 1048 | + input_data = b"x" * (128 * 1024) # > typical pipe buffer | ||
| 1049 | + | ||
| 1050 | + # Cross-platform wake mechanism: the slow reader connects to a | ||
| 1051 | + # loopback TCP socket and blocks in select() on it (capped at 9s | ||
| 1052 | + # as a safety net we don't expect to hit). After phase 1 raises | ||
| 1053 | + # TimeoutExpired, the parent sends a byte to release the child so | ||
| 1054 | + # it drains stdin. A socket (rather than a raw pipe) is required | ||
| 1055 | + # because Windows select() only supports sockets, not arbitrary | ||
| 1056 | + # file descriptors. | ||
| 1057 | + server = socket.create_server(('127.0.0.1', 0), backlog=1) | ||
| 1058 | + server.settimeout(10) # bound the accept() if the child fails to start | ||
| 1059 | + port = server.getsockname()[1] | ||
| 1060 | + # The child sends one byte (low byte of its PID) first so the parent | ||
| 1061 | + # can detect the rare case of an unrelated process on the same host | ||
| 1062 | + # connecting to our ephemeral port before our child does. A single | ||
| 1063 | + # byte gives 1/256 collision odds, which is plenty for flake-prevention. | ||
| 1064 | + slow_reader = ( | ||
| 1065 | + "import os, socket, sys, select; " | ||
| 1066 | + f"s = socket.create_connection(('127.0.0.1', {port}), timeout=9); " | ||
| 1067 | + "s.sendall(bytes([os.getpid() & 0xff])); " | ||
| 1068 | + "select.select([s], [], [], 9); " | ||
| 1069 | + "sys.stdout.buffer.write(sys.stdin.buffer.read())" | ||
| 1070 | + ) | ||
| 1049 | 1071 | ||
| 1050 | 1072 | p = subprocess.Popen( | |
| 1051 | - [sys.executable, "-c", | ||
| 1052 | - "import sys, time; " | ||
| 1053 | - "time.sleep(30); " # Don't read stdin for a long time | ||
| 1054 | - "sys.stdout.buffer.write(sys.stdin.buffer.read())"], | ||
| 1073 | + [sys.executable, "-c", slow_reader], | ||
| 1055 | 1074 | stdin=subprocess.PIPE, | |
| 1056 | 1075 | stdout=subprocess.PIPE, | |
| 1057 | 1076 | stderr=subprocess.PIPE) | |
| 1058 | 1077 | ||
| 1078 | + conn = None | ||
| 1059 | 1079 | try: | |
| 1080 | + conn, _ = server.accept() | ||
| 1081 | + server.close() | ||
| 1082 | + server = None | ||
| 1083 | + | ||
| 1084 | + conn.settimeout(5) | ||
| 1085 | + peer_byte = conn.recv(1) | ||
| 1086 | + conn.settimeout(None) | ||
| 1087 | + self.assertEqual(peer_byte, bytes([p.pid & 0xff]), | ||
| 1088 | + f"loopback handshake byte {peer_byte!r} != " | ||
| 1089 | + f"low byte of child PID {p.pid} ({p.pid & 0xff:#x})") | ||
| 1090 | + | ||
| 1060 | 1091 | timeout = 0.2 | |
| 1061 | 1092 | start = time.monotonic() | |
| 1062 | 1093 | try: | |
@@ -1065,19 +1096,24 @@ def test_communicate_timeout_large_input(self): | |||
| 1065 | 1096 | elapsed = time.monotonic() - start | |
| 1066 | 1097 | self.fail( | |
| 1067 | 1098 | f"TimeoutExpired not raised. communicate() completed in " | |
| 1068 | - f"{elapsed:.2f}s, but subprocess sleeps for 30s. " | ||
| 1099 | + f"{elapsed:.2f}s, but slow reader stalls for up to 9s. " | ||
| 1069 | 1100 | "Stdin writing blocked without enforcing timeout.") | |
| 1070 | 1101 | except subprocess.TimeoutExpired: | |
| 1071 | 1102 | elapsed = time.monotonic() - start | |
| 1072 | 1103 | ||
| 1073 | 1104 | # Timeout should occur close to the specified timeout value, | |
| 1074 | 1105 | # not after waiting for the subprocess to finish sleeping. | |
| 1075 | 1106 | # Allow generous margin for slow CI, but must be well under | |
| 1076 | - # the subprocess sleep time. | ||
| 1107 | + # the slow-reader's stall cap. | ||
| 1077 | 1108 | self.assertLess(elapsed, 5.0, | |
| 1078 | 1109 | f"TimeoutExpired raised after {elapsed:.2f}s; expected ~{timeout}s. " | |
| 1079 | 1110 | "Stdin writing blocked without checking timeout.") | |
| 1080 | 1111 | ||
| 1112 | + # Release the slow reader so it stops blocking and drains stdin. | ||
| 1113 | + conn.sendall(b'go') | ||
| 1114 | + conn.close() | ||
| 1115 | + conn = None | ||
| 1116 | + | ||
| 1081 | 1117 | # After timeout, continue communication. The remaining input | |
| 1082 | 1118 | # should be sent and we should receive all data back. | |
| 1083 | 1119 | stdout, stderr = p.communicate() | |
@@ -1087,6 +1123,10 @@ def test_communicate_timeout_large_input(self): | |||
| 1087 | 1123 | f"Expected {len(input_data)} bytes output but got {len(stdout)}") | |
| 1088 | 1124 | self.assertEqual(stdout, input_data) | |
| 1089 | 1125 | finally: | |
| 1126 | + if conn is not None: | ||
| 1127 | + conn.close() | ||
| 1128 | + if server is not None: | ||
| 1129 | + server.close() | ||
| 1090 | 1130 | p.kill() | |
| 1091 | 1131 | p.wait() | |
| 1092 | 1132 | ||
| Back | FazBrowse Home | New Git URL |
0 commit comments