174
177
self.deferred.errback(reason)
181
class UtilityProcessProtocol(ProcessProtocol):
183
Helper class for launching a Python process and getting a result from it.
185
@ivar program: A string giving a Python program for the child process to
190
def run(cls, reactor, argv, env):
192
Run a Python process connected to a new instance of this protocol
193
class. Return the protocol instance.
195
The Python process is given C{self.program} on the command line to
196
execute, in addition to anything specified by C{argv}. C{env} is
197
the complete environment.
201
reactor.spawnProcess(
202
self, exe, [exe, "-c", self.program] + argv, env=env)
204
run = classmethod(run)
212
def parseChunks(self, bytes):
214
Called with all bytes received on stdout when the process exits.
216
raise NotImplementedError()
221
Return a Deferred which will fire with the result of L{parseChunks}
222
when the child process exits.
225
self.requests.append(d)
229
def _fireResultDeferreds(self, result):
231
Callback all Deferreds returned up until now by L{getResult}
232
with the given result object.
234
requests = self.requests
240
def outReceived(self, bytes):
242
Accumulate output from the child process in a list.
244
self.bytes.append(bytes)
247
def processEnded(self, reason):
249
Handle process termination by parsing all received output and firing
250
any waiting Deferreds.
252
self._fireResultDeferreds(self.parseChunks(self.bytes))
257
class GetArgumentVector(UtilityProcessProtocol):
259
Protocol which will read a serialized argv from a process and
260
expose it to interested parties.
263
"from sys import stdout, argv\n"
264
"stdout.write(chr(0).join(argv))\n"
267
def parseChunks(self, chunks):
269
Parse the output from the process to which this protocol was
270
connected, which is a single unterminated line of \\0-separated
271
strings giving the argv of that process. Return this as a list of
274
return ''.join(chunks).split('\0')
278
class GetEnvironmentDictionary(UtilityProcessProtocol):
280
Protocol which will read a serialized environment dict from a process
281
and expose it to interested parties.
284
"from sys import stdout\n"
285
"from os import environ\n"
286
"items = environ.iteritems()\n"
287
"stdout.write(chr(0).join([k + chr(0) + v for k, v in items]))\n"
290
def parseChunks(self, chunks):
292
Parse the output from the process to which this protocol was
293
connected, which is a single unterminated line of \\0-separated
294
strings giving key value pairs of the environment from that process.
295
Return this as a dictionary.
297
environString = ''.join(chunks)
298
if not environString:
300
environ = iter(environString.split('\0'))
305
except StopIteration:
176
314
class ProcessTestCase(SignalMixin, unittest.TestCase):
177
315
"""Test running a process."""
289
427
return d.addCallback(processEnded)
430
def test_wrongArguments(self):
432
Test invalid arguments to spawnProcess: arguments and environment
433
must only contains string or unicode, and not null bytes.
436
p = protocol.ProcessProtocol()
449
# Sanity check - this will fail for people who have mucked with
450
# their site configuration in a stupid way, but there's nothing we
452
badUnicode = u'\N{SNOWMAN}'
454
badUnicode.encode(sys.getdefaultencoding())
455
except UnicodeEncodeError:
456
# Okay, that unicode doesn't encode, put it in as a bad environment
458
badEnvs.append({badUnicode: 'value for bad unicode key'})
459
badEnvs.append({'key for bad unicode value': badUnicode})
460
badArgs.append([exe, badUnicode])
462
# It _did_ encode. Most likely, Gtk2 is being used and the
463
# default system encoding is UTF-8, which can encode anything.
464
# In any case, if implicit unicode -> str conversion works for
465
# that string, we can't test that TypeError gets raised instead,
466
# so just leave it off.
472
reactor.spawnProcess, p, exe, [exe, "-c", ""], env=env)
477
reactor.spawnProcess, p, exe, args, env=None)
480
# Use upper-case so that the environment key test uses an upper case
481
# name: some versions of Windows only support upper case environment
482
# variable names, and I think Python (as of 2.5) doesn't use the right
483
# syscall for lowercase or mixed case names to work anyway.
484
okayUnicode = u"UNICODE"
485
encodedValue = "UNICODE"
487
def _deprecatedUnicodeSupportTest(self, processProtocolClass, argv=[], env={}):
489
Check that a deprecation warning is emitted when passing unicode to
490
spawnProcess for an argv value or an environment key or value.
491
Check that the warning is of the right type, has the right message,
492
and refers to the correct file. Unfortunately, don't check that the
493
line number is correct, because that is too hard for me to figure
496
@param processProtocolClass: A L{UtilityProcessProtocol} subclass
497
which will be instantiated to communicate with the child process.
499
@param argv: The argv argument to spawnProcess.
501
@param env: The env argument to spawnProcess.
503
@return: A Deferred which fires when the test is complete.
505
# Sanity to check to make sure we can actually encode this unicode
506
# with the default system encoding. This may be excessively
509
self.okayUnicode.encode(sys.getdefaultencoding()),
513
def showwarning(*args):
514
warningsShown.append(args)
516
origshow = warnings.showwarning
517
origregistry = globals().get('__warningregistry__', {})
519
warnings.showwarning = showwarning
520
globals()['__warningregistry__'] = {}
521
p = processProtocolClass.run(reactor, argv, env)
523
warnings.showwarning = origshow
524
globals()['__warningregistry__'] = origregistry
527
self.assertEqual(len(warningsShown), 1, pformat(warningsShown))
528
message, category, filename, lineno = warningsShown[0]
531
("Argument strings and environment keys/values passed to "
532
"reactor.spawnProcess should be str, not unicode.",))
533
self.assertIdentical(category, DeprecationWarning)
535
# Use starts with because of .pyc/.pyo issues.
537
__file__.startswith(filename),
538
'Warning in %r, expected %r' % (filename, __file__))
540
# It would be nice to be able to check the line number as well, but
541
# different configurations actually end up reporting different line
542
# numbers (generally the variation is only 1 line, but that's enough
543
# to fail the test erroneously...).
544
# self.assertEqual(lineno, 202)
548
def test_deprecatedUnicodeArgvSupport(self):
550
Test that a unicode string passed for an argument value is allowed
551
if it can be encoded with the default system encoding, but that a
552
deprecation warning is emitted.
554
d = self._deprecatedUnicodeSupportTest(GetArgumentVector, argv=[self.okayUnicode])
555
def gotArgVector(argv):
556
self.assertEqual(argv, ['-c', self.encodedValue])
557
d.addCallback(gotArgVector)
561
def test_deprecatedUnicodeEnvKeySupport(self):
563
Test that a unicode string passed for the key of the environment
564
dictionary is allowed if it can be encoded with the default system
565
encoding, but that a deprecation warning is emitted.
567
d = self._deprecatedUnicodeSupportTest(
568
GetEnvironmentDictionary, env={self.okayUnicode: self.encodedValue})
569
def gotEnvironment(environ):
570
self.assertEqual(environ[self.encodedValue], self.encodedValue)
571
d.addCallback(gotEnvironment)
575
def test_deprecatedUnicodeEnvValueSupport(self):
577
Test that a unicode string passed for the value of the environment
578
dictionary is allowed if it can be encoded with the default system
579
encoding, but that a deprecation warning is emitted.
581
d = self._deprecatedUnicodeSupportTest(
582
GetEnvironmentDictionary, env={self.encodedValue: self.okayUnicode})
583
def gotEnvironment(environ):
584
# On Windows, the environment contains more things than we
585
# specified, so only make sure that at least the key we wanted
586
# is there, rather than testing the dictionary for exact
588
self.assertEqual(environ[self.encodedValue], self.encodedValue)
589
d.addCallback(gotEnvironment)
292
594
class TwoProcessProtocol(protocol.ProcessProtocol):
701
1003
self.assertRaises(ValueError, reactor.spawnProcess, p, pyExe, pyArgs, childFDs={1:'r'})
703
1005
class UtilTestCase(unittest.TestCase):
704
def setUpClass(klass):
706
foobar = j("foo", "bar")
707
foobaz = j("foo", "baz")
708
bazfoo = j("baz", "foo")
709
barfoo = j("baz", "bar")
711
for d in "foo", foobar, foobaz, "baz", bazfoo, barfoo:
712
if os.path.exists(d):
713
shutil.rmtree(d, True)
716
f = file(j(foobaz, "executable"), "w")
718
os.chmod(j(foobaz, "executable"), 0700)
720
f = file(j("foo", "executable"), "w")
722
os.chmod(j("foo", "executable"), 0700)
724
f = file(j(bazfoo, "executable"), "w")
726
os.chmod(j(bazfoo, "executable"), 0700)
728
f = file(j(bazfoo, "executable.bin"), "w")
730
os.chmod(j(bazfoo, "executable.bin"), 0700)
732
f = file(j(barfoo, "executable"), "w")
735
klass.oldPath = os.environ['PATH']
736
os.environ['PATH'] = os.pathsep.join((foobar, foobaz, bazfoo, barfoo))
738
def tearDownClass(klass):
740
os.environ['PATH'] = klass.oldPath
741
foobar = j("foo", "bar")
742
foobaz = j("foo", "baz")
743
bazfoo = j("baz", "foo")
744
barfoo = j("baz", "bar")
747
os.remove(j(foobaz, "executable"))
748
os.remove(j("foo", "executable"))
749
os.remove(j(bazfoo, "executable"))
750
os.remove(j(bazfoo, "executable.bin"))
751
os.remove(j(barfoo, "executable"))
753
for d in foobar, foobaz, bazfoo, barfoo, "foo", "baz":
1007
Tests for process-related helper functions (currently only
1012
Create several directories and files, some of which are executable
1013
and some of which are not. Save the current PATH setting.
1017
base = self.mktemp()
1019
self.foo = j(base, "foo")
1020
self.baz = j(base, "baz")
1021
self.foobar = j(self.foo, "bar")
1022
self.foobaz = j(self.foo, "baz")
1023
self.bazfoo = j(self.baz, "foo")
1024
self.bazbar = j(self.baz, "bar")
1026
for d in self.foobar, self.foobaz, self.bazfoo, self.bazbar:
1029
for name, mode in [(j(self.foobaz, "executable"), 0700),
1030
(j(self.foo, "executable"), 0700),
1031
(j(self.bazfoo, "executable"), 0700),
1032
(j(self.bazfoo, "executable.bin"), 0700),
1033
(j(self.bazbar, "executable"), 0)]:
1036
os.chmod(name, mode)
1038
self.oldPath = os.environ.get('PATH', None)
1039
os.environ['PATH'] = os.pathsep.join((
1040
self.foobar, self.foobaz, self.bazfoo, self.bazbar))
1045
Restore the saved PATH setting.
1047
if self.oldPath is None:
1049
del os.environ['PATH']
1053
os.environ['PATH'] = self.oldPath
1056
def test_whichWithoutPATH(self):
1058
Test that if C{os.environ} does not have a C{'PATH'} key,
1059
L{procutils.which} returns an empty list.
1061
del os.environ['PATH']
1062
self.assertEqual(procutils.which("executable"), [])
756
1065
def testWhich(self):
757
1066
j = os.path.join
758
1067
paths = procutils.which("executable")
759
expectedPaths = [j("foo", "baz", "executable"),
760
j("baz", "foo", "executable")]
1068
expectedPaths = [j(self.foobaz, "executable"),
1069
j(self.bazfoo, "executable")]
761
1070
if runtime.platform.isWindows():
762
expectedPaths.append(j("baz", "bar", "executable"))
1071
expectedPaths.append(j(self.bazbar, "executable"))
763
1072
self.assertEquals(paths, expectedPaths)
765
1075
def testWhichPathExt(self):
766
1076
j = os.path.join
767
1077
old = os.environ.get('PATHEXT', None)
added
removed
twisted/test/test_process.py
