fork download
  1. """Test readahead buffer in pipes [1]
  2.  
  3. [1]: http://bugs.python.org/issue3907
  4.  
  5. - `for line in p.stdout` fails on Python 2
  6. - `for line in iter(p.stdout.readline, b'')` succeeds on Python 2
  7. - both work on Python 3
  8. """
  9. import sys
  10. import time
  11. from subprocess import Popen, PIPE
  12.  
  13. def test_proc(use_iter=False, timeout=1):
  14. p = Popen([sys.executable, "-c", """import sys, time
  15. for i in range(2):
  16. print(i)
  17. sys.stdout.flush()
  18. time.sleep(%d)
  19. """ % timeout], stdout=PIPE, bufsize=1)
  20.  
  21. if use_iter:
  22. it = iter(p.stdout.readline, b'')
  23. else:
  24. it = p.stdout
  25.  
  26. elapsed = []
  27. start = time.time()
  28. for line in it:
  29. elapsed.append(time.time() - start)
  30. print("%s %.2f" % (line.decode().rstrip(), elapsed[-1]))
  31. assert min(elapsed) < timeout
  32. assert max(elapsed) < 2*timeout
  33.  
  34. test_proc(use_iter=True)
  35. test_proc(use_iter=False)
  36.  
Success #stdin #stdout 0.48s 11280KB
stdin
Standard input is empty
stdout
0 0.15
1 1.16
0 0.16
1 1.16