GitHub CopilotWrite better code with AI | MCP RegistryIntegrate external tools | ActionsAutomate any workflow | CodespacesInstant dev environments | IssuesPlan and track work | Code ReviewManage code changes | Code QualityEnforce quality at merge | Why GitHub | Marketplace | View all features | Enterprises | Small and medium teams | Startups | View all use cases | View all industries | View all solutions | AI | Software Development | DevOps | Security | View all topics | Customer stories | Events & webinars | Ebooks & reports | Business insights | Trust center | Partners | View all resources
@@ -10,6 +10,7 @@
1010importre
1111importsysconfig
1212importwarnings
13+importselect
1314try:
1415importgc
1516exceptImportError:
@@ -964,6 +965,83 @@ def test_bytes_program(self):
964965exitcode=subprocess.call([program, "-c", "pass"], env=envb)
965966self.assertEqual(exitcode, 0)
966967968+deftest_pipe_cloexec(self):
969+sleeper=support.findfile("input_reader.py", subdir="subprocessdata")
970+fd_status=support.findfile("fd_status.py", subdir="subprocessdata")
971+972+p1=subprocess.Popen([sys.executable, sleeper],
973+stdin=subprocess.PIPE, stdout=subprocess.PIPE,
974+stderr=subprocess.PIPE, close_fds=False)
975+976+self.addCleanup(p1.communicate, b'')
977+978+p2=subprocess.Popen([sys.executable, fd_status],
979+stdout=subprocess.PIPE, close_fds=False)
980+981+output, error=p2.communicate()
982+result_fds=set(map(int, output.split(b',')))
983+unwanted_fds=set([p1.stdin.fileno(), p1.stdout.fileno(),
984+p1.stderr.fileno()])
985+986+self.assertFalse(result_fds&unwanted_fds,
987+"Expected no fds from %r to be open in child, "
988+"found %r"%
989+ (unwanted_fds, result_fds&unwanted_fds))
990+991+deftest_pipe_cloexec_real_tools(self):
992+qcat=support.findfile("qcat.py", subdir="subprocessdata")
993+qgrep=support.findfile("qgrep.py", subdir="subprocessdata")
994+995+subdata=b'zxcvbn'
996+data=subdata*4+b'\n'
997+998+p1=subprocess.Popen([sys.executable, qcat],
999+stdin=subprocess.PIPE, stdout=subprocess.PIPE,
1000+close_fds=False)
1001+1002+p2=subprocess.Popen([sys.executable, qgrep, subdata],
1003+stdin=p1.stdout, stdout=subprocess.PIPE,
1004+close_fds=False)
1005+1006+self.addCleanup(p1.wait)
1007+self.addCleanup(p2.wait)
1008+self.addCleanup(p1.terminate)
1009+self.addCleanup(p2.terminate)
1010+1011+p1.stdin.write(data)
1012+p1.stdin.close()
1013+1014+readfiles, ignored1, ignored2=select.select([p2.stdout], [], [], 10)
1015+1016+self.assertTrue(readfiles, "The child hung")
1017+self.assertEqual(p2.stdout.read(), data)
1018+1019+deftest_close_fds(self):
1020+fd_status=support.findfile("fd_status.py", subdir="subprocessdata")
1021+1022+fds=os.pipe()
1023+self.addCleanup(os.close, fds[0])
1024+self.addCleanup(os.close, fds[1])
1025+1026+open_fds=set(fds)
1027+1028+p=subprocess.Popen([sys.executable, fd_status],
1029+stdout=subprocess.PIPE, close_fds=False)
1030+output, ignored=p.communicate()
1031+remaining_fds=set(map(int, output.split(b',')))
1032+1033+self.assertEqual(remaining_fds&open_fds, open_fds,
1034+"Some fds were closed")
1035+1036+p=subprocess.Popen([sys.executable, fd_status],
1037+stdout=subprocess.PIPE, close_fds=True)
1038+output, ignored=p.communicate()
1039+remaining_fds=set(map(int, output.split(b',')))
1040+1041+self.assertFalse(remaining_fds&open_fds,
1042+"Some fds were left open")
1043+self.assertIn(1, remaining_fds, "Subprocess failed")
[email protected](mswindows, "Windows specific tests")
9691047classWin32ProcessTestCase(BaseTestCase):