@@ -90,38 +90,20 @@ def read(self, size):
9090@pytest.mark.skipif(sys.platform == "win32", reason="requires POSIX unbuffered pipe reads")
9191def test_stream_copy_keeps_reading_an_unbuffered_pipe():
9292 read_fd, write_fd = os.pipe()
93- first_copied = threading.Event()
94- errors = []
9593
9694 class Destination(BytesIO):
9795 def write(self, data):
9896 result = super().write(data)
99- if data:
100- first_copied.set()
97+ if data and not writer.closed:
98+ # Make the second chunk available only after the first short read is copied.
99+ writer.write(b"second")
100+ writer.close()
101101 return result
102102
103- def produce():
104- try:
105- os.write(write_fd, b"first-")
106- if not first_copied.wait(5):
107- raise RuntimeError("first pipe chunk was not copied")
108- os.write(write_fd, b"second")
109- except Exception as error:
110- errors.append(error)
111- finally:
112- os.close(write_fd)
113-
114103 destination = Destination()
115- with os.fdopen(read_fd, "rb", buffering=0) as source:
116- thread = threading.Thread(target=produce)
117- thread.start()
118- try:
119- copied = stream_copy(source, destination, chunk_size=1024)
120- finally:
121- first_copied.set()
122- thread.join(5)
123- assert not thread.is_alive()
124- assert not errors
104+ with os.fdopen(read_fd, "rb", buffering=0) as source, os.fdopen(write_fd, "wb", buffering=0) as writer:
105+ writer.write(b"first-")
106+ copied = stream_copy(source, destination, chunk_size=1024)
125107 assert copied == len(b"first-second")
126108 assert destination.getvalue() == b"first-second"
127109
0 commit comments