Rietveld Code Review Tool
Help | Bug tracker | Discussion group | Source code | Sign in
(27)

Side by Side Diff: Lib/test/test_subprocess.py

Issue 10639: reindent.py converts newlines to platform default
Patch Set: Created 8 years, 8 months ago
Left:
Right:
Use n/p to move between diff chunks; N/P to move between comments. Please Sign in to add in-line comments.
Jump to:
View unified diff | Download patch
« no previous file with comments | « Lib/test/test_ssl.py ('k') | Lib/test/test_support.py » ('j') | no next file with comments »
Toggle Intra-line Diffs ('i') | Expand Comments ('e') | Collapse Comments ('c') | Show Comments Hide Comments ('s')
OLDNEW
1 import unittest 1 import unittest
2 from test import support 2 from test import support
3 import subprocess 3 import subprocess
4 import sys 4 import sys
5 import signal 5 import signal
6 import io 6 import io
7 import os 7 import os
8 import errno 8 import errno
9 import tempfile 9 import tempfile
10 import time 10 import time
(...skipping 372 matching lines...) Expand 10 before | Expand all | Expand 10 after
383 newenv = os.environ.copy() 383 newenv = os.environ.copy()
384 newenv["FRUIT"] = "orange" 384 newenv["FRUIT"] = "orange"
385 with subprocess.Popen([sys.executable, "-c", 385 with subprocess.Popen([sys.executable, "-c",
386 'import sys,os;' 386 'import sys,os;'
387 'sys.stdout.write(os.getenv("FRUIT"))'], 387 'sys.stdout.write(os.getenv("FRUIT"))'],
388 stdout=subprocess.PIPE, 388 stdout=subprocess.PIPE,
389 env=newenv) as p: 389 env=newenv) as p:
390 stdout, stderr = p.communicate() 390 stdout, stderr = p.communicate()
391 self.assertEqual(stdout, b"orange") 391 self.assertEqual(stdout, b"orange")
392 392
393 # Windows requires at least the SYSTEMROOT environment variable to start 393 @unittest.skipIf(sysconfig.get_config_var('Py_ENABLE_SHARED') == 1,
394 # Python
395 @unittest.skipIf(sys.platform == 'win32',
396 'cannot test an empty env on Windows')
397 @unittest.skipIf(sysconfig.get_config_var('Py_ENABLE_SHARED') is not None,
398 'the python library cannot be loaded ' 394 'the python library cannot be loaded '
399 'with an empty environment') 395 'with an empty environment')
400 def test_empty_env(self): 396 def test_empty_env(self):
401 with subprocess.Popen([sys.executable, "-c", 397 with subprocess.Popen([sys.executable, "-c",
402 'import os; ' 398 'import os; '
403 'print(list(os.environ.keys()))'], 399 'print(list(os.environ.keys()))'],
404 stdout=subprocess.PIPE, 400 stdout=subprocess.PIPE,
405 env={}) as p: 401 env={}) as p:
406 stdout, stderr = p.communicate() 402 stdout, stderr = p.communicate()
407 self.assertIn(stdout.strip(), 403 self.assertEqual(stdout.strip(), b"[]")
408 (b"[]",
409 # Mac OS X adds __CF_USER_TEXT_ENCODING variable to an empty
410 # environment
411 b"['__CF_USER_TEXT_ENCODING']"))
412 404
413 def test_communicate_stdin(self): 405 def test_communicate_stdin(self):
414 p = subprocess.Popen([sys.executable, "-c", 406 p = subprocess.Popen([sys.executable, "-c",
415 'import sys;' 407 'import sys;'
416 'sys.exit(sys.stdin.read() == "pear")'], 408 'sys.exit(sys.stdin.read() == "pear")'],
417 stdin=subprocess.PIPE) 409 stdin=subprocess.PIPE)
418 p.communicate(b"pear") 410 p.communicate(b"pear")
419 self.assertEqual(p.returncode, 1) 411 self.assertEqual(p.returncode, 1)
420 412
421 def test_communicate_stdout(self): 413 def test_communicate_stdout(self):
(...skipping 128 matching lines...) Expand 10 before | Expand all | Expand 10 after
550 self.addCleanup(p.stderr.close) 542 self.addCleanup(p.stderr.close)
551 self.addCleanup(p.stdin.close) 543 self.addCleanup(p.stdin.close)
552 p.stdin.write(b"banana") 544 p.stdin.write(b"banana")
553 (stdout, stderr) = p.communicate(b"split") 545 (stdout, stderr) = p.communicate(b"split")
554 self.assertEqual(stdout, b"bananasplit") 546 self.assertEqual(stdout, b"bananasplit")
555 self.assertStderrEqual(stderr, b"") 547 self.assertStderrEqual(stderr, b"")
556 548
557 def test_universal_newlines(self): 549 def test_universal_newlines(self):
558 p = subprocess.Popen([sys.executable, "-c", 550 p = subprocess.Popen([sys.executable, "-c",
559 'import sys,os;' + SETBINARY + 551 'import sys,os;' + SETBINARY +
560 'sys.stdout.write(sys.stdin.readline());' 552 'sys.stdout.write("line1\\n");'
561 'sys.stdout.flush();' 553 'sys.stdout.flush();'
562 'sys.stdout.write("line2\\n");' 554 'sys.stdout.write("line2\\n");'
563 'sys.stdout.flush();' 555 'sys.stdout.flush();'
564 'sys.stdout.write(sys.stdin.read());' 556 'sys.stdout.write("line3\\r\\n");'
565 'sys.stdout.flush();' 557 'sys.stdout.flush();'
566 'sys.stdout.write("line4\\n");' 558 'sys.stdout.write("line4\\r");'
567 'sys.stdout.flush();' 559 'sys.stdout.flush();'
568 'sys.stdout.write("line5\\r\\n");' 560 'sys.stdout.write("\\nline5");'
569 'sys.stdout.flush();' 561 'sys.stdout.flush();'
570 'sys.stdout.write("line6\\r");' 562 'sys.stdout.write("\\nline6");'],
571 'sys.stdout.flush();'
572 'sys.stdout.write("\\nline7");'
573 'sys.stdout.flush();'
574 'sys.stdout.write("\\nline8");'],
575 stdin=subprocess.PIPE,
576 stdout=subprocess.PIPE, 563 stdout=subprocess.PIPE,
577 universal_newlines=1) 564 universal_newlines=1)
578 p.stdin.write("line1\n")
579 self.assertEqual(p.stdout.readline(), "line1\n")
580 p.stdin.write("line3\n")
581 p.stdin.close()
582 self.addCleanup(p.stdout.close) 565 self.addCleanup(p.stdout.close)
583 self.assertEqual(p.stdout.readline(), 566 stdout = p.stdout.read()
584 "line2\n") 567 self.assertEqual(stdout, "line1\nline2\nline3\nline4\nline5\nline6")
585 self.assertEqual(p.stdout.read(6),
586 "line3\n")
587 self.assertEqual(p.stdout.read(),
588 "line4\nline5\nline6\nline7\nline8")
589 568
590 def test_universal_newlines_communicate(self): 569 def test_universal_newlines_communicate(self):
591 # universal newlines through communicate() 570 # universal newlines through communicate()
592 p = subprocess.Popen([sys.executable, "-c", 571 p = subprocess.Popen([sys.executable, "-c",
593 'import sys,os;' + SETBINARY + 572 'import sys,os;' + SETBINARY +
573 'sys.stdout.write("line1\\n");'
574 'sys.stdout.flush();'
594 'sys.stdout.write("line2\\n");' 575 'sys.stdout.write("line2\\n");'
595 'sys.stdout.flush();' 576 'sys.stdout.flush();'
596 'sys.stdout.write("line4\\n");' 577 'sys.stdout.write("line3\\r\\n");'
597 'sys.stdout.flush();' 578 'sys.stdout.flush();'
598 'sys.stdout.write("line5\\r\\n");' 579 'sys.stdout.write("line4\\r");'
599 'sys.stdout.flush();' 580 'sys.stdout.flush();'
600 'sys.stdout.write("line6\\r");' 581 'sys.stdout.write("\\nline5");'
601 'sys.stdout.flush();' 582 'sys.stdout.flush();'
602 'sys.stdout.write("\\nline7");' 583 'sys.stdout.write("\\nline6");'],
603 'sys.stdout.flush();' 584 stdout=subprocess.PIPE, stderr=subprocess.PIPE,
604 'sys.stdout.write("\\nline8");'],
605 stderr=subprocess.PIPE,
606 stdout=subprocess.PIPE,
607 universal_newlines=1) 585 universal_newlines=1)
608 self.addCleanup(p.stdout.close) 586 self.addCleanup(p.stdout.close)
609 self.addCleanup(p.stderr.close) 587 self.addCleanup(p.stderr.close)
610 # BUG: can't give a non-empty stdin because it breaks both the
611 # select- and poll-based communicate() implementations.
612 (stdout, stderr) = p.communicate() 588 (stdout, stderr) = p.communicate()
613 self.assertEqual(stdout, 589 self.assertEqual(stdout, "line1\nline2\nline3\nline4\nline5\nline6")
614 "line2\nline4\nline5\nline6\nline7\nline8")
615
616 def test_universal_newlines_communicate_stdin(self):
617 # universal newlines through communicate(), with only stdin
618 p = subprocess.Popen([sys.executable, "-c",
619 'import sys,os;' + SETBINARY + '''\nif True:
620 s = sys.stdin.readline()
621 assert s == "line1\\n", repr(s)
622 s = sys.stdin.read()
623 assert s == "line3\\n", repr(s)
624 '''],
625 stdin=subprocess.PIPE,
626 universal_newlines=1)
627 (stdout, stderr) = p.communicate("line1\nline3\n")
628 self.assertEqual(p.returncode, 0)
629 590
630 def test_no_leaking(self): 591 def test_no_leaking(self):
631 # Make sure we leak no resources 592 # Make sure we leak no resources
632 if not mswindows: 593 if not mswindows:
633 max_handles = 1026 # too much for most UNIX systems 594 max_handles = 1026 # too much for most UNIX systems
634 else: 595 else:
635 max_handles = 2050 # too much for (at least some) Windows setups 596 max_handles = 2050 # too much for (at least some) Windows setups
636 handles = [] 597 handles = []
637 tmpdir = tempfile.mkdtemp() 598 tmpdir = tempfile.mkdtemp()
638 try: 599 try:
(...skipping 151 matching lines...) Expand 10 before | Expand all | Expand 10 after
790 self.addCleanup(p.stdin.close) 751 self.addCleanup(p.stdin.close)
791 p.communicate(b"x" * 2**20) 752 p.communicate(b"x" * 2**20)
792 753
793 def test_communicate_epipe_only_stdin(self): 754 def test_communicate_epipe_only_stdin(self):
794 # Issue 10963: communicate() should hide EPIPE 755 # Issue 10963: communicate() should hide EPIPE
795 p = subprocess.Popen([sys.executable, "-c", 'pass'], 756 p = subprocess.Popen([sys.executable, "-c", 'pass'],
796 stdin=subprocess.PIPE) 757 stdin=subprocess.PIPE)
797 self.addCleanup(p.stdin.close) 758 self.addCleanup(p.stdin.close)
798 time.sleep(2) 759 time.sleep(2)
799 p.communicate(b"x" * 2**20) 760 p.communicate(b"x" * 2**20)
800
801 @unittest.skipUnless(hasattr(signal, 'SIGALRM'),
802 "Requires signal.SIGALRM")
803 def test_communicate_eintr(self):
804 # Issue #12493: communicate() should handle EINTR
805 def handler(signum, frame):
806 pass
807 old_handler = signal.signal(signal.SIGALRM, handler)
808 self.addCleanup(signal.signal, signal.SIGALRM, old_handler)
809
810 # the process is running for 2 seconds
811 args = [sys.executable, "-c", 'import time; time.sleep(2)']
812 for stream in ('stdout', 'stderr'):
813 kw = {stream: subprocess.PIPE}
814 with subprocess.Popen(args, **kw) as process:
815 signal.alarm(1)
816 # communicate() will be interrupted by SIGALRM
817 process.communicate()
818 761
819 762
820 # context manager 763 # context manager
821 class _SuppressCoreFiles(object): 764 class _SuppressCoreFiles(object):
822 """Try to prevent core files from being created.""" 765 """Try to prevent core files from being created."""
823 old_limit = None 766 old_limit = None
824 767
825 def __enter__(self): 768 def __enter__(self):
826 """Try to save previous ulimit, then set it to (0, 0).""" 769 """Try to save previous ulimit, then set it to (0, 0)."""
827 try: 770 try:
(...skipping 854 matching lines...) Expand 10 before | Expand all | Expand 10 after
1682 1625
1683 1626
1684 def test_main(): 1627 def test_main():
1685 unit_tests = (ProcessTestCase, 1628 unit_tests = (ProcessTestCase,
1686 POSIXProcessTestCase, 1629 POSIXProcessTestCase,
1687 Win32ProcessTestCase, 1630 Win32ProcessTestCase,
1688 CommandTests, 1631 CommandTests,
1689 ProcessTestCaseNoPoll, 1632 ProcessTestCaseNoPoll,
1690 HelperFunctionTests, 1633 HelperFunctionTests,
1691 CommandsWithSpaces, 1634 CommandsWithSpaces,
1692 ContextManagerTests, 1635 ContextManagerTests)
1693 )
1694 1636
1695 support.run_unittest(*unit_tests) 1637 support.run_unittest(*unit_tests)
1696 support.reap_children() 1638 support.reap_children()
1697 1639
1698 if __name__ == "__main__": 1640 if __name__ == "__main__":
1699 unittest.main() 1641 unittest.main()
OLDNEW
« no previous file with comments | « Lib/test/test_ssl.py ('k') | Lib/test/test_support.py » ('j') | no next file with comments »

RSS Feeds Recent Issues | This issue
This is Rietveld 894c83f36cb7+