15
16
from twisted.python.runtime import platform
16
17
from twisted.python.filepath import FilePath
17
18
from twisted.python.failure import Failure
18
from twisted.internet.defer import Deferred
19
from twisted.internet import utils
20
from twisted.internet.defer import Deferred, succeed
19
21
from twisted.internet.protocol import ProcessProtocol
20
22
from twisted.internet.error import ProcessDone, PotentialZombieWarning
21
23
from twisted.internet.error import ProcessTerminated
25
skipWindowsNopywin32 = None
26
if platform.isWindows():
30
skipWindowsNopywin32 = ("On windows, spawnProcess is not available "
31
"in the absence of win32process.")
33
class _ShutdownCallbackProcessProtocol(ProcessProtocol):
35
An L{IProcessProtocol} which fires a Deferred when the process it is
38
def __init__(self, whenFinished):
39
self.whenFinished = whenFinished
42
def processEnded(self, reason):
43
self.whenFinished.callback(None)
24
47
class ProcessTestsBuilderBase(ReactorBuilder):
25
48
def spawnProcess(self, reactor):
27
50
Call C{reactor.spawnProcess} with some simple arguments. Do this here
28
51
so that code object referenced by the stack frame has a C{co_filename}
29
52
attribute set to this file so that L{TestCase.assertWarns} can be used.
54
Return a L{Deferred} which fires when the process has exited.
56
onConnection = Deferred()
31
57
reactor.spawnProcess(
32
ProcessProtocol(), sys.executable, [sys.executable, "-c", ""],
58
_ShutdownCallbackProcessProtocol(onConnection), sys.executable,
59
[sys.executable, "-c", ""], usePTY=self.usePTY)
36
63
def test_spawnProcessTooEarlyWarns(self):
43
70
warning and this test.
45
72
reactor = self.buildReactor()
73
whenFinished = self.assertWarns(
47
74
PotentialZombieWarning,
48
75
PotentialZombieWarning.MESSAGE, __file__,
49
76
self.spawnProcess, reactor)
79
# Handle the exact problem the warning we're testing for is about.
80
# If the SIGCHLD arrives before we install our SIGCHLD handler,
81
# we'll never see the process exit. So after the SIGCHLD handler
82
# is installed, try to reap children, just in case.
83
from twisted.internet.process import reapAllProcesses
85
reactor.callWhenRunning(check)
87
# Now wait for the process to exit before finishing the test. If we
88
# don't do this, tearDown might close the PTY before the child is done
90
whenFinished.addCallback(
91
lambda ignored: reactor.callWhenRunning(reactor.stop))
93
self.runReactor(reactor)
52
96
def test_callWhenRunningSpawnProcessWarningFree(self):
54
98
L{PotentialZombieWarning} is not emitted when the reactor is run after
55
99
C{reactor.callWhenRunning(reactor.spawnProcess, ...)} has been called.
58
self.patch(warnings, 'warn', lambda *a, **kw: events.append(a))
59
101
reactor = self.buildReactor()
60
reactor.callWhenRunning(self.spawnProcess, reactor)
61
reactor.callWhenRunning(reactor.stop)
103
whenFinished = self.spawnProcess(reactor)
104
# Wait for the process to exit before finishing the test. If we
105
# don't do this, tearDown might close the PTY before the child is
107
whenFinished.addCallback(lambda ign: reactor.stop())
108
reactor.callWhenRunning(spawnProcess)
62
109
self.runReactor(reactor)
63
self.assertFalse(events)
110
self.assertEqual(self.flushWarnings(), [])
66
113
if getattr(signal, 'SIGCHLD', None) is None:
302
349
self.runReactor(reactor)
352
def makeSourceFile(self, sourceLines):
354
Write the given list of lines to a text file and return the absolute
357
script = self.mktemp()
358
scriptFile = file(script, 'wt')
359
scriptFile.write(os.linesep.join(sourceLines) + os.linesep)
361
return os.path.abspath(script)
364
def test_shebang(self):
366
Spawning a process with an executable which is a script starting
367
with an interpreter definition line (#!) uses that interpreter to
370
SHEBANG_OUTPUT = 'this is the shebang output'
372
scriptFile = self.makeSourceFile([
373
"#!%s" % (sys.executable,),
375
"sys.stdout.write('%s')" % (SHEBANG_OUTPUT,),
376
"sys.stdout.flush()"])
377
os.chmod(scriptFile, 0700)
379
reactor = self.buildReactor()
381
def cbProcessExited((out, err, code)):
382
msg("cbProcessExited((%r, %r, %d))" % (out, err, code))
383
self.assertEqual(out, SHEBANG_OUTPUT)
384
self.assertEqual(err, "")
385
self.assertEqual(code, 0)
387
def shutdown(passthrough):
392
d = utils.getProcessOutputAndValue(scriptFile, reactor=reactor)
394
d.addCallback(cbProcessExited)
397
reactor.callWhenRunning(start)
398
self.runReactor(reactor)
401
def test_processCommandLineArguments(self):
403
Arguments given to spawnProcess are passed to the child process as
407
# On Windows, stdout is not opened in binary mode by default,
408
# so newline characters are munged on writing, interfering with
413
' msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY)\n'
414
'except ImportError:\n'
416
'for arg in sys.argv[1:]:\n'
417
' sys.stdout.write(arg + chr(0))\n'
418
' sys.stdout.flush()')
420
args = ['hello', '"', ' \t|<>^&', r'"\\"hello\\"', r'"foo\ bar baz\""']
421
# Ensure that all non-NUL characters can be passed too.
422
args.append(''.join(map(chr, xrange(1, 256))))
424
reactor = self.buildReactor()
426
def processFinished(output):
427
output = output.split('\0')
428
# Drop the trailing \0.
430
self.assertEquals(args, output)
432
def shutdown(result):
438
d.addCallback(lambda dummy: utils.getProcessOutput(
439
sys.executable, ['-c', source] + args, reactor=reactor))
440
d.addCallback(processFinished)
443
reactor.callWhenRunning(spawnChild)
444
self.runReactor(reactor)
448
ProcessTestsBuilder.skip = skipWindowsNopywin32
305
449
globals().update(ProcessTestsBuilder.makeTestCaseClasses())
added
removed
twisted/internet/test/test_process.py
