1
# Copyright (c) 2008-2009 Twisted Matrix Laboratories.
2
# See LICENSE for details.
5
Tests for implementations of L{IReactorProcess}.
10
import os, sys, signal, threading
12
from twisted.trial.unittest import TestCase
13
from twisted.internet.test.reactormixins import ReactorBuilder
14
from twisted.python.compat import set
15
from twisted.python.log import msg, err
16
from twisted.python.runtime import platform
17
from twisted.python.filepath import FilePath
18
from twisted.internet import utils
19
from twisted.internet.interfaces import IReactorProcess
20
from twisted.internet.defer import Deferred, succeed
21
from twisted.internet.protocol import ProcessProtocol
22
from twisted.internet.error import ProcessDone, ProcessTerminated
25
class _ShutdownCallbackProcessProtocol(ProcessProtocol):
27
An L{IProcessProtocol} which fires a Deferred when the process it is
30
def __init__(self, whenFinished):
31
self.whenFinished = whenFinished
34
def processEnded(self, reason):
35
self.whenFinished.callback(None)
39
class ProcessTestsBuilderBase(ReactorBuilder):
41
Base class for L{IReactorProcess} tests which defines some tests which
42
can be applied to PTY or non-PTY uses of C{spawnProcess}.
44
Subclasses are expected to set the C{usePTY} attribute to C{True} or
47
requiredInterfaces = [IReactorProcess]
49
def test_spawnProcessEarlyIsReaped(self):
51
If, before the reactor is started with L{IReactorCore.run}, a
52
process is started with L{IReactorProcess.spawnProcess} and
53
terminates, the process is reaped once the reactor is started.
55
reactor = self.buildReactor()
57
# Create the process with no shared file descriptors, so that there
58
# are no other events for the reactor to notice and "cheat" with.
59
# We want to be sure it's really dealing with the process exiting,
60
# not some associated event.
66
# Arrange to notice the SIGCHLD.
67
signaled = threading.Event()
70
signal.signal(signal.SIGCHLD, handler)
72
# Start a process - before starting the reactor!
75
_ShutdownCallbackProcessProtocol(ended), sys.executable,
76
[sys.executable, "-c", ""], usePTY=self.usePTY, childFDs=childFDs)
78
# Wait for the SIGCHLD (which might have been delivered before we got
79
# here, but that's okay because the signal handler was installed above,
80
# before we could have gotten it).
82
if not signaled.isSet():
83
self.fail("Timed out waiting for child process to exit.")
85
# Capture the processEnded callback.
87
ended.addCallback(result.append)
90
# The synchronous path through spawnProcess / Process.__init__ /
91
# registerReapProcessHandler was encountered. There's no reason to
92
# start the reactor, because everything is done already.
95
# Otherwise, though, start the reactor so it can tell us the process
97
ended.addCallback(lambda ignored: reactor.stop())
98
self.runReactor(reactor)
100
# Make sure the reactor stopped because the Deferred fired.
101
self.assertTrue(result)
103
if getattr(signal, 'SIGCHLD', None) is None:
104
test_spawnProcessEarlyIsReaped.skip = (
105
"Platform lacks SIGCHLD, early-spawnProcess test can't work.")
108
def test_processExitedWithSignal(self):
110
The C{reason} argument passed to L{IProcessProtocol.processExited} is a
111
L{ProcessTerminated} instance if the child process exits with a signal.
114
sigNum = getattr(signal, 'SIG' + sigName)
118
# Talk so the parent process knows the process is running. This is
119
# necessary because ProcessProtocol.makeConnection may be called
120
# before this process is exec'd. It would be unfortunate if we
121
# SIGTERM'd the Twisted process while it was on its way to doing
123
"sys.stdout.write('x')\n"
124
"sys.stdout.flush()\n"
125
"sys.stdin.read()\n")
127
class Exiter(ProcessProtocol):
128
def childDataReceived(self, fd, data):
129
msg('childDataReceived(%d, %r)' % (fd, data))
130
self.transport.signalProcess(sigName)
132
def childConnectionLost(self, fd):
133
msg('childConnectionLost(%d)' % (fd,))
135
def processExited(self, reason):
136
msg('processExited(%r)' % (reason,))
137
# Protect the Deferred from the failure so that it follows
138
# the callback chain. This doesn't use the errback chain
139
# because it wants to make sure reason is a Failure. An
140
# Exception would also make an errback-based test pass, and
141
# that would be wrong.
142
exited.callback([reason])
144
def processEnded(self, reason):
145
msg('processEnded(%r)' % (reason,))
147
reactor = self.buildReactor()
148
reactor.callWhenRunning(
149
reactor.spawnProcess, Exiter(), sys.executable,
150
[sys.executable, "-c", source], usePTY=self.usePTY)
152
def cbExited((failure,)):
153
# Trapping implicitly verifies that it's a Failure (rather than
154
# an exception) and explicitly makes sure it's the right type.
155
failure.trap(ProcessTerminated)
157
if platform.isWindows():
158
# Windows can't really /have/ signals, so it certainly can't
159
# report them as the reason for termination. Maybe there's
160
# something better we could be doing here, anyway? Hard to
161
# say. Anyway, this inconsistency between different platforms
162
# is extremely unfortunate and I would remove it if I
164
self.assertIdentical(err.signal, None)
165
self.assertEqual(err.exitCode, 1)
167
self.assertEqual(err.signal, sigNum)
168
self.assertIdentical(err.exitCode, None)
170
exited.addCallback(cbExited)
171
exited.addErrback(err)
172
exited.addCallback(lambda ign: reactor.stop())
174
self.runReactor(reactor)
178
class ProcessTestsBuilder(ProcessTestsBuilderBase):
180
Builder defining tests relating to L{IReactorProcess} for child processes
181
which do not have a PTY.
185
keepStdioOpenProgram = FilePath(__file__).sibling('process_helper.py').path
186
if platform.isWindows():
187
keepStdioOpenArg = "windows"
189
# Just a value that doesn't equal "windows"
190
keepStdioOpenArg = ""
192
# Define this test here because PTY-using processes only have stdin and
193
# stdout and the test would need to be different for that to work.
194
def test_childConnectionLost(self):
196
L{IProcessProtocol.childConnectionLost} is called each time a file
197
descriptor associated with a child process is closed.
199
connected = Deferred()
200
lost = {0: Deferred(), 1: Deferred(), 2: Deferred()}
202
class Closer(ProcessProtocol):
203
def makeConnection(self, transport):
204
connected.callback(transport)
206
def childConnectionLost(self, childFD):
207
lost[childFD].callback(None)
212
" line = sys.stdin.readline().strip()\n"
215
" os.close(int(line))\n")
217
reactor = self.buildReactor()
218
reactor.callWhenRunning(
219
reactor.spawnProcess, Closer(), sys.executable,
220
[sys.executable, "-c", source], usePTY=self.usePTY)
222
def cbConnected(transport):
223
transport.write('2\n')
224
return lost[2].addCallback(lambda ign: transport)
225
connected.addCallback(cbConnected)
227
def lostSecond(transport):
228
transport.write('1\n')
229
return lost[1].addCallback(lambda ign: transport)
230
connected.addCallback(lostSecond)
232
def lostFirst(transport):
233
transport.write('\n')
234
connected.addCallback(lostFirst)
235
connected.addErrback(err)
237
def cbEnded(ignored):
239
connected.addCallback(cbEnded)
241
self.runReactor(reactor)
244
# This test is here because PTYProcess never delivers childConnectionLost.
245
def test_processEnded(self):
247
L{IProcessProtocol.processEnded} is called after the child process
248
exits and L{IProcessProtocol.childConnectionLost} is called for each of
249
its file descriptors.
254
class Ender(ProcessProtocol):
255
def childDataReceived(self, fd, data):
256
msg('childDataReceived(%d, %r)' % (fd, data))
257
self.transport.loseConnection()
259
def childConnectionLost(self, childFD):
260
msg('childConnectionLost(%d)' % (childFD,))
263
def processExited(self, reason):
264
msg('processExited(%r)' % (reason,))
266
def processEnded(self, reason):
267
msg('processEnded(%r)' % (reason,))
268
ended.callback([reason])
270
reactor = self.buildReactor()
271
reactor.callWhenRunning(
272
reactor.spawnProcess, Ender(), sys.executable,
273
[sys.executable, self.keepStdioOpenProgram, "child",
274
self.keepStdioOpenArg],
277
def cbEnded((failure,)):
278
failure.trap(ProcessDone)
279
self.assertEqual(set(lost), set([0, 1, 2]))
280
ended.addCallback(cbEnded)
282
ended.addErrback(err)
283
ended.addCallback(lambda ign: reactor.stop())
285
self.runReactor(reactor)
288
# This test is here because PTYProcess.loseConnection does not actually
289
# close the file descriptors to the child process. This test needs to be
290
# written fairly differently for PTYProcess.
291
def test_processExited(self):
293
L{IProcessProtocol.processExited} is called when the child process
294
exits, even if file descriptors associated with the child are still
301
class Waiter(ProcessProtocol):
302
def childDataReceived(self, fd, data):
303
msg('childDataReceived(%d, %r)' % (fd, data))
305
def childConnectionLost(self, childFD):
306
msg('childConnectionLost(%d)' % (childFD,))
309
allLost.callback(None)
311
def processExited(self, reason):
312
msg('processExited(%r)' % (reason,))
313
# See test_processExitedWithSignal
314
exited.callback([reason])
315
self.transport.loseConnection()
317
reactor = self.buildReactor()
318
reactor.callWhenRunning(
319
reactor.spawnProcess, Waiter(), sys.executable,
320
[sys.executable, self.keepStdioOpenProgram, "child",
321
self.keepStdioOpenArg],
324
def cbExited((failure,)):
325
failure.trap(ProcessDone)
326
msg('cbExited; lost = %s' % (lost,))
327
self.assertEqual(lost, [])
329
exited.addCallback(cbExited)
331
def cbAllLost(ignored):
332
self.assertEqual(set(lost), set([0, 1, 2]))
333
exited.addCallback(cbAllLost)
335
exited.addErrback(err)
336
exited.addCallback(lambda ign: reactor.stop())
338
self.runReactor(reactor)
341
def makeSourceFile(self, sourceLines):
343
Write the given list of lines to a text file and return the absolute
346
script = self.mktemp()
347
scriptFile = file(script, 'wt')
348
scriptFile.write(os.linesep.join(sourceLines) + os.linesep)
350
return os.path.abspath(script)
353
def test_shebang(self):
355
Spawning a process with an executable which is a script starting
356
with an interpreter definition line (#!) uses that interpreter to
359
SHEBANG_OUTPUT = 'this is the shebang output'
361
scriptFile = self.makeSourceFile([
362
"#!%s" % (sys.executable,),
364
"sys.stdout.write('%s')" % (SHEBANG_OUTPUT,),
365
"sys.stdout.flush()"])
366
os.chmod(scriptFile, 0700)
368
reactor = self.buildReactor()
370
def cbProcessExited((out, err, code)):
371
msg("cbProcessExited((%r, %r, %d))" % (out, err, code))
372
self.assertEqual(out, SHEBANG_OUTPUT)
373
self.assertEqual(err, "")
374
self.assertEqual(code, 0)
376
def shutdown(passthrough):
381
d = utils.getProcessOutputAndValue(scriptFile, reactor=reactor)
383
d.addCallback(cbProcessExited)
386
reactor.callWhenRunning(start)
387
self.runReactor(reactor)
390
def test_processCommandLineArguments(self):
392
Arguments given to spawnProcess are passed to the child process as
396
# On Windows, stdout is not opened in binary mode by default,
397
# so newline characters are munged on writing, interfering with
402
' msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)\n'
403
'except ImportError:\n'
405
'for arg in sys.argv[1:]:\n'
406
' sys.stdout.write(arg + chr(0))\n'
407
' sys.stdout.flush()')
409
args = ['hello', '"', ' \t|<>^&', r'"\\"hello\\"', r'"foo\ bar baz\""']
410
# Ensure that all non-NUL characters can be passed too.
411
args.append(''.join(map(chr, xrange(1, 256))))
413
reactor = self.buildReactor()
415
def processFinished(output):
416
output = output.split('\0')
417
# Drop the trailing \0.
419
self.assertEquals(args, output)
421
def shutdown(result):
427
d.addCallback(lambda dummy: utils.getProcessOutput(
428
sys.executable, ['-c', source] + args, reactor=reactor))
429
d.addCallback(processFinished)
432
reactor.callWhenRunning(spawnChild)
433
self.runReactor(reactor)
434
globals().update(ProcessTestsBuilder.makeTestCaseClasses())
438
class PTYProcessTestsBuilder(ProcessTestsBuilderBase):
440
Builder defining tests relating to L{IReactorProcess} for child processes
445
if platform.isWindows():
446
skip = "PTYs are not supported on Windows."
447
elif platform.isMacOSX():
449
"twisted.internet.pollreactor.PollReactor":
450
"OS X's poll() does not support PTYs"}
451
globals().update(PTYProcessTestsBuilder.makeTestCaseClasses())
455
class PotentialZombieWarningTests(TestCase):
457
Tests for L{twisted.internet.error.PotentialZombieWarning}.
459
def test_deprecated(self):
461
Accessing L{PotentialZombieWarning} via the
462
I{PotentialZombieWarning} attribute of L{twisted.internet.error}
463
results in a deprecation warning being emitted.
465
from twisted.internet import error
466
error.PotentialZombieWarning
468
warnings = self.flushWarnings([self.test_deprecated])
469
self.assertEquals(warnings[0]['category'], DeprecationWarning)
471
warnings[0]['message'],
472
"twisted.internet.error.PotentialZombieWarning was deprecated in "
473
"Twisted 10.0.0: There is no longer any potential for zombie "
475
self.assertEquals(len(warnings), 1)