| 722 | @unittest.skipUnless(fcntl and hasattr(fcntl, 'F_GETPIPE_SZ'), |
| 723 | 'fcntl.F_GETPIPE_SZ required for test.') |
| 724 | def test_pipesizes(self): |
| 725 | test_pipe_r, test_pipe_w = os.pipe() |
| 726 | try: |
| 727 | # Get the default pipesize with F_GETPIPE_SZ |
| 728 | pipesize_default = fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ) |
| 729 | finally: |
| 730 | os.close(test_pipe_r) |
| 731 | os.close(test_pipe_w) |
| 732 | pipesize = pipesize_default // 2 |
| 733 | pagesize_default = support.get_pagesize() |
| 734 | if pipesize < pagesize_default: # the POSIX minimum |
| 735 | raise unittest.SkipTest( |
| 736 | 'default pipesize too small to perform test.') |
| 737 | p = subprocess.Popen( |
| 738 | [sys.executable, "-c", |
| 739 | 'import sys; sys.stdin.read(); sys.stdout.write("out"); ' |
| 740 | 'sys.stderr.write("error!")'], |
| 741 | stdin=subprocess.PIPE, stdout=subprocess.PIPE, |
| 742 | stderr=subprocess.PIPE, pipesize=pipesize) |
| 743 | try: |
| 744 | for fifo in [p.stdin, p.stdout, p.stderr]: |
| 745 | self.assertEqual( |
| 746 | fcntl.fcntl(fifo.fileno(), fcntl.F_GETPIPE_SZ), |
| 747 | pipesize) |
| 748 | # Windows pipe size can be acquired via GetNamedPipeInfoFunction |
| 749 | # https://docs.microsoft.com/en-us/windows/win32/api/namedpipeapi/nf-namedpipeapi-getnamedpipeinfo |
| 750 | # However, this function is not yet in _winapi. |
| 751 | p.stdin.write(b"pear") |
| 752 | p.stdin.close() |
| 753 | p.stdout.close() |
| 754 | p.stderr.close() |
| 755 | finally: |
| 756 | p.kill() |
| 757 | p.wait() |
| 758 | |
| 759 | @unittest.skipUnless(fcntl and hasattr(fcntl, 'F_GETPIPE_SZ'), |
| 760 | 'fcntl.F_GETPIPE_SZ required for test.') |