1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
|
from os.path import dirname
import py, sys
if sys.platform == 'win32':
py.test.skip("not used on win32")
class AppTestSubprocess:
spaceconfig = dict(usemodules=('_posixsubprocess', 'signal',
'fcntl', 'select', 'time', 'struct'))
# XXX write more tests
def setup_class(cls):
cls.w_dir = cls.space.wrap(dirname(__file__))
def test_cloexec_pipe(self):
import _posixsubprocess, os
fd1, fd2 = _posixsubprocess.cloexec_pipe()
# Sanity checks
assert 0 <= fd1 < 4096
assert 0 <= fd2 < 4096
os.close(fd1)
os.close(fd2)
def test_close_fds_true(self):
import traceback # Work around a recursion limit
import subprocess
import os.path
import os
fds = os.pipe()
#self.addCleanup(os.close, fds[0])
#self.addCleanup(os.close, fds[1])
open_fds = set(fds)
# add a bunch more fds
for _ in range(9):
fd = os.open("/dev/null", os.O_RDONLY)
#self.addCleanup(os.close, fd)
open_fds.add(fd)
p = subprocess.Popen(['/usr/bin/env', 'python', os.path.join(self.dir, 'fd_status.py')], stdout=subprocess.PIPE, close_fds=True)
output, ignored = p.communicate()
remaining_fds = set(map(int, output.split(b',')))
assert not (remaining_fds & open_fds), "Some fds were left open"
assert 1 in remaining_fds, "Subprocess failed"
def test_start_new_session(self):
# For code coverage of calling setsid(). We don't care if we get an
# EPERM error from it depending on the test execution environment, that
# still indicates that it was called.
import traceback # Work around a recursion limit
import subprocess
import os
try:
output = subprocess.check_output(
['/usr/bin/env', 'python', "-c",
"import os; print(os.getpgid(os.getpid()))"],
start_new_session=True)
except OSError as e:
if e.errno != errno.EPERM:
raise
else:
parent_pgid = os.getpgid(os.getpid())
child_pgid = int(output)
assert parent_pgid != child_pgid
def test_cpython_issue15736(self):
import _posixsubprocess
import sys
n = 0
class Z(object):
def __len__(self):
return sys.maxsize + n
def __getitem__(self, i):
return b'x'
raises(MemoryError, _posixsubprocess.fork_exec,
1,Z(),3,[1, 2],5,6,7,8,9,10,11,12,13,14,15,16,17)
n = 1
raises(OverflowError, _posixsubprocess.fork_exec,
1,Z(),3,[1, 2],5,6,7,8,9,10,11,12,13,14,15,16,17)
def test_pass_fds_make_inheritable(self):
import subprocess, posix
fd1, fd2 = posix.pipe()
assert posix.get_inheritable(fd1) is False
assert posix.get_inheritable(fd2) is False
subprocess.check_call(['/usr/bin/env', 'python', '-c',
'import os;os.write(%d,b"K")' % fd2],
close_fds=True, pass_fds=[fd2])
res = posix.read(fd1, 1)
assert res == b"K"
posix.close(fd1)
posix.close(fd2)
|
