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
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
|
# mostly tests from test_subprocess.py that used to have problems
import sys
import os
import errno
import greentest
import gevent
from gevent import subprocess
import time
if subprocess.mswindows:
SETBINARY = 'import msvcrt; msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY);'
else:
SETBINARY = ''
class Test(greentest.TestCase):
def test_exit(self):
popen = subprocess.Popen([sys.executable, '-c', 'import sys; sys.exit(10)'])
self.assertEqual(popen.wait(), 10)
def test_wait(self):
popen = subprocess.Popen([sys.executable, '-c', 'import sys; sys.exit(11)'])
gevent.wait([popen])
self.assertEqual(popen.poll(), 11)
def test_child_exception(self):
try:
subprocess.Popen(['*']).wait()
except OSError, ex:
assert ex.errno == 2, ex
else:
raise AssertionError('Expected OSError: [Errno 2] No such file or directory')
def test_leak(self):
num_before = greentest.get_number_open_files()
p = subprocess.Popen([sys.executable, "-c", "print()"],
stdout=subprocess.PIPE)
p.wait()
del p
num_after = greentest.get_number_open_files()
self.assertEqual(num_before, num_after)
def test_communicate(self):
p = subprocess.Popen([sys.executable, "-c",
'import sys,os;'
'sys.stderr.write("pineapple");'
'sys.stdout.write(sys.stdin.read())'],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
(stdout, stderr) = p.communicate("banana")
self.assertEqual(stdout, "banana")
if sys.executable.endswith('-dbg'):
assert stderr.startswith('pineapple')
else:
self.assertEqual(stderr, "pineapple")
def test_universal1(self):
p = subprocess.Popen([sys.executable, "-c",
'import sys,os;' + SETBINARY +
'sys.stdout.write("line1\\n");'
'sys.stdout.flush();'
'sys.stdout.write("line2\\r");'
'sys.stdout.flush();'
'sys.stdout.write("line3\\r\\n");'
'sys.stdout.flush();'
'sys.stdout.write("line4\\r");'
'sys.stdout.flush();'
'sys.stdout.write("\\nline5");'
'sys.stdout.flush();'
'sys.stdout.write("\\nline6");'],
stdout=subprocess.PIPE,
universal_newlines=1)
try:
stdout = p.stdout.read()
if hasattr(file, 'newlines'):
# Interpreter with universal newline support
self.assertEqual(stdout,
"line1\nline2\nline3\nline4\nline5\nline6")
else:
# Interpreter without universal newline support
self.assertEqual(stdout,
"line1\nline2\rline3\r\nline4\r\nline5\nline6")
finally:
p.stdout.close()
def test_universal2(self):
p = subprocess.Popen([sys.executable, "-c",
'import sys,os;' + SETBINARY +
'sys.stdout.write("line1\\n");'
'sys.stdout.flush();'
'sys.stdout.write("line2\\r");'
'sys.stdout.flush();'
'sys.stdout.write("line3\\r\\n");'
'sys.stdout.flush();'
'sys.stdout.write("line4\\r\\nline5");'
'sys.stdout.flush();'
'sys.stdout.write("\\nline6");'],
stdout=subprocess.PIPE,
universal_newlines=1)
try:
stdout = p.stdout.read()
if hasattr(file, 'newlines'):
# Interpreter with universal newline support
self.assertEqual(stdout,
"line1\nline2\nline3\nline4\nline5\nline6")
else:
# Interpreter without universal newline support
self.assertEqual(stdout,
"line1\nline2\rline3\r\nline4\r\nline5\nline6")
finally:
p.stdout.close()
if sys.platform != 'win32':
def test_nonblock_removed(self):
# see issue #134
r, w = os.pipe()
p = subprocess.Popen(['grep', 'text'], stdin=subprocess.FileObject(r))
try:
os.close(w)
time.sleep(0.1)
self.assertEqual(p.poll(), None)
finally:
if p.poll() is None:
p.kill()
def test_issue148(self):
for i in range(7):
try:
subprocess.Popen('this_name_must_not_exist')
except OSError:
ex = sys.exc_info()[1]
if ex.errno != errno.ENOENT:
raise
else:
raise AssertionError('must fail with ENOENT')
def test_check_output_keyword_error(self):
try:
subprocess.check_output([sys.executable, '-c', 'import sys; sys.exit(44)'])
except subprocess.CalledProcessError, e:
self.assertEqual(e.returncode, 44)
else:
raise AssertionError('must fail with CalledProcessError')
if __name__ == '__main__':
greentest.main()
|
