def test_pipe(self):
r, w = os.pipe()
rf = greenio.GreenPipe(r, 'rb')
wf = greenio.GreenPipe(w, 'wb', 0)
def sender(f, content):
for ch in map(six.int2byte, six.iterbytes(content)):
eventlet.sleep(0.0001)
f.write(ch)
f.close()
one_line = b"12345\n"
eventlet.spawn(sender, wf, one_line * 5)
for i in range(5):
line = rf.readline()
eventlet.sleep(0.01)
self.assertEqual(line, one_line)
self.assertEqual(rf.readline(), b'')
评论列表
文章目录