"""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)
IiIiVGVzdCByZWFkYWhlYWQgYnVmZmVyIGluIHBpcGVzIFsxXQoKWzFdOiAgaHR0cDovL2IuLi5jb250ZW50LWF2YWlsYWJsZS10by1hdXRob3Itb25seS4uLm4ub3JnL2lzc3VlMzkwNwoKLSBgZm9yIGxpbmUgaW4gcC5zdGRvdXRgIGZhaWxzIG9uIFB5dGhvbiAyCi0gYGZvciBsaW5lIGluIGl0ZXIocC5zdGRvdXQucmVhZGxpbmUsIGInJylgIHN1Y2NlZWRzIG9uIFB5dGhvbiAyCi0gYm90aCB3b3JrIG9uIFB5dGhvbiAzCiIiIgppbXBvcnQgc3lzCmltcG9ydCB0aW1lCmZyb20gc3VicHJvY2VzcyBpbXBvcnQgUG9wZW4sIFBJUEUKCmRlZiB0ZXN0X3Byb2ModXNlX2l0ZXI9RmFsc2UsIHRpbWVvdXQ9MSk6CiAgICBwID0gUG9wZW4oW3N5cy5leGVjdXRhYmxlLCAiLWMiLCAiIiJpbXBvcnQgc3lzLCB0aW1lCmZvciBpIGluIHJhbmdlKDIpOgogICAgcHJpbnQoaSkKICAgIHN5cy5zdGRvdXQuZmx1c2goKQogICAgdGltZS5zbGVlcCglZCkKICAgICIiIiAlIHRpbWVvdXRdLCBzdGRvdXQ9UElQRSwgYnVmc2l6ZT0xKQoKICAgIGlmIHVzZV9pdGVyOgogICAgICAgIGl0ID0gaXRlcihwLnN0ZG91dC5yZWFkbGluZSwgYicnKQogICAgZWxzZToKICAgICAgICBpdCA9IHAuc3Rkb3V0CgogICAgZWxhcHNlZCA9IFtdCiAgICBzdGFydCA9IHRpbWUudGltZSgpCiAgICBmb3IgbGluZSBpbiBpdDoKICAgICAgICBlbGFwc2VkLmFwcGVuZCh0aW1lLnRpbWUoKSAtIHN0YXJ0KQogICAgICAgIHByaW50KCIlcyAlLjJmIiAlIChsaW5lLmRlY29kZSgpLnJzdHJpcCgpLCBlbGFwc2VkWy0xXSkpCiAgICBhc3NlcnQgbWluKGVsYXBzZWQpIDwgdGltZW91dAogICAgYXNzZXJ0IG1heChlbGFwc2VkKSA8IDIqdGltZW91dAoKdGVzdF9wcm9jKHVzZV9pdGVyPVRydWUpCnRlc3RfcHJvYyh1c2VfaXRlcj1GYWxzZSkK