fork download
"""Test readahead buffer in pipes [1]

[1]:  http://b...content-available-to-author-only...n.org/issue3907

- `for line in p.stdout` fails on Python 2
- `for line in iter(p.stdout.readline, b'')` succeeds on Python 2
- both work on Python 3
"""
import sys
import time
from subprocess import Popen, PIPE

def test_proc(use_iter=False, timeout=1):
    p = Popen([sys.executable, "-c", """import sys, time
for i in range(2):
    print(i)
    sys.stdout.flush()
    time.sleep(%d)
    """ % timeout], stdout=PIPE, bufsize=1)

    if use_iter:
        it = iter(p.stdout.readline, b'')
    else:
        it = p.stdout

    elapsed = []
    start = time.time()
    for line in it:
        elapsed.append(time.time() - start)
        print("%s %.2f" % (line.decode().rstrip(), elapsed[-1]))
    assert min(elapsed) < timeout
    assert max(elapsed) < 2*timeout

test_proc(use_iter=True)
test_proc(use_iter=False)
Runtime error #stdin #stdout 0.25s 9344KB
stdin
Standard input is empty
stdout
0 0.08
1 1.08
0 2.09
1 2.09