1
# Copyright (c) 2007-2009 Twisted Matrix Laboratories.
2
# See LICENSE for details.
5
Tests for L{twisted.application.app} and L{twisted.scripts.twistd}.
8
import signal, inspect, errno
10
import os, sys, cPickle, StringIO
16
from zope.interface import implements
18
from twisted.trial import unittest
20
from twisted.application import service, app
21
from twisted.scripts import twistd
22
from twisted.python import log
23
from twisted.python.usage import UsageError
24
from twisted.python.log import ILogObserver
25
from twisted.python.versions import Version
26
from twisted.python.components import Componentized
27
from twisted.internet.defer import Deferred
28
from twisted.python.fakepwd import UserDatabase
31
from twisted.python import syslog
36
from twisted.scripts import _twistd_unix
40
from twisted.scripts._twistd_unix import UnixApplicationRunner
41
from twisted.scripts._twistd_unix import UnixAppLogger
51
except (ImportError, SystemExit):
52
# For some reasons, hotshot.stats seems to raise SystemExit on some
53
# distributions, probably when considered non-free. See the import of
54
# this module in twisted.application.app for more details.
65
def patchUserDatabase(patch, user, uid, group, gid):
67
Patch L{pwd.getpwnam} so that it behaves as though only one user exists
68
and patch L{grp.getgrnam} so that it behaves as though only one group
71
@param patch: A function like L{TestCase.patch} which will be used to
72
install the fake implementations.
75
@param user: The name of the single user which will exist.
78
@param uid: The UID of the single user which will exist.
81
@param group: The name of the single user which will exist.
84
@param gid: The GID of the single group which will exist.
86
# Try not to be an unverified fake, but try not to depend on quirks of
87
# the system either (eg, run as a process with a uid and gid which
88
# equal each other, and so doesn't reliably test that uid is used where
89
# uid should be used and gid is used where gid should be used). -exarkun
90
pwent = pwd.getpwuid(os.getuid())
91
grent = grp.getgrgid(os.getgid())
93
database = UserDatabase()
95
user, pwent.pw_passwd, uid, pwent.pw_gid,
96
pwent.pw_gecos, pwent.pw_dir, pwent.pw_shell)
100
result[result.index(grent.gr_name)] = group
101
result[result.index(grent.gr_gid)] = gid
102
result = tuple(result)
103
return {group: result}[name]
105
patch(pwd, "getpwnam", database.getpwnam)
106
patch(grp, "getgrnam", getgrnam)
110
class MockServiceMaker(object):
112
A non-implementation of L{twisted.application.service.IServiceMaker}.
116
def makeService(self, options):
118
Take a L{usage.Options} instance and return a
119
L{service.IService} provider.
121
self.options = options
122
self.service = service.Service()
127
class CrippledAppLogger(app.AppLogger):
129
@see: CrippledApplicationRunner.
132
def start(self, application):
137
class CrippledApplicationRunner(twistd._SomeApplicationRunner):
139
An application runner that cripples the platform-specific runner and
140
nasty side-effect-having code so that we can use it without actually
141
running any environment-affecting code.
143
loggerFactory = CrippledAppLogger
145
def preApplication(self):
149
def postApplication(self):
154
class ServerOptionsTest(unittest.TestCase):
156
Non-platform-specific tests for the pltaform-specific ServerOptions class.
159
def test_postOptionsSubCommandCausesNoSave(self):
161
postOptions should set no_save to True when a subcommand is used.
163
config = twistd.ServerOptions()
164
config.subCommand = 'ueoa'
166
self.assertEquals(config['no_save'], True)
169
def test_postOptionsNoSubCommandSavesAsUsual(self):
171
If no sub command is used, postOptions should not touch no_save.
173
config = twistd.ServerOptions()
175
self.assertEquals(config['no_save'], False)
178
def test_reportProfileDeprecation(self):
180
Check that the --report-profile option prints a C{DeprecationWarning}.
182
config = twistd.ServerOptions()
184
DeprecationWarning, "--report-profile option is deprecated and "
185
"a no-op since Twisted 8.0.", app.__file__,
186
config.parseOptions, ["--report-profile", "foo"])
189
def test_listAllProfilers(self):
191
All the profilers that can be used in L{app.AppProfiler} are listed in
194
config = twistd.ServerOptions()
195
helpOutput = str(config)
196
for profiler in app.AppProfiler.profilers:
197
self.assertIn(profiler, helpOutput)
200
def test_defaultUmask(self):
202
The default value for the C{umask} option is C{None}.
204
config = twistd.ServerOptions()
205
self.assertEqual(config['umask'], None)
208
def test_umask(self):
210
The value given for the C{umask} option is parsed as an octal integer
213
config = twistd.ServerOptions()
214
config.parseOptions(['--umask', '123'])
215
self.assertEqual(config['umask'], 83)
216
config.parseOptions(['--umask', '0123'])
217
self.assertEqual(config['umask'], 83)
220
def test_invalidUmask(self):
222
If a value is given for the C{umask} option which cannot be parsed as
223
an integer, L{UsageError} is raised by L{ServerOptions.parseOptions}.
225
config = twistd.ServerOptions()
226
self.assertRaises(UsageError, config.parseOptions, ['--umask', 'abcdef'])
228
if _twistd_unix is None:
229
msg = "twistd unix not available"
230
test_defaultUmask.skip = test_umask.skip = test_invalidUmask.skip = msg
234
class TapFileTest(unittest.TestCase):
236
Test twistd-related functionality that requires a tap file on disk.
241
Create a trivial Application and put it in a tap file on disk.
243
self.tapfile = self.mktemp()
244
f = file(self.tapfile, 'wb')
245
cPickle.dump(service.Application("Hi!"), f)
249
def test_createOrGetApplicationWithTapFile(self):
251
Ensure that the createOrGetApplication call that 'twistd -f foo.tap'
252
makes will load the Application out of foo.tap.
254
config = twistd.ServerOptions()
255
config.parseOptions(['-f', self.tapfile])
256
application = CrippledApplicationRunner(config).createOrGetApplication()
257
self.assertEquals(service.IService(application).name, 'Hi!')
261
class TestLoggerFactory(object):
263
A logger factory for L{TestApplicationRunner}.
266
def __init__(self, runner):
270
def start(self, application):
272
Save the logging start on the C{runner} instance.
274
self.runner.order.append("log")
275
self.runner.hadApplicationLogObserver = hasattr(self.runner,
286
class TestApplicationRunner(app.ApplicationRunner):
288
An ApplicationRunner which tracks the environment in which its methods are
292
def __init__(self, options):
293
app.ApplicationRunner.__init__(self, options)
295
self.logger = TestLoggerFactory(self)
298
def preApplication(self):
299
self.order.append("pre")
300
self.hadApplicationPreApplication = hasattr(self, 'application')
303
def postApplication(self):
304
self.order.append("post")
305
self.hadApplicationPostApplication = hasattr(self, 'application')
309
class ApplicationRunnerTest(unittest.TestCase):
311
Non-platform-specific tests for the platform-specific ApplicationRunner.
314
config = twistd.ServerOptions()
315
self.serviceMaker = MockServiceMaker()
316
# Set up a config object like it's been parsed with a subcommand
317
config.loadedPlugins = {'test_command': self.serviceMaker}
318
config.subOptions = object()
319
config.subCommand = 'test_command'
323
def test_applicationRunnerGetsCorrectApplication(self):
325
Ensure that a twistd plugin gets used in appropriate ways: it
326
is passed its Options instance, and the service it returns is
327
added to the application.
329
arunner = CrippledApplicationRunner(self.config)
332
self.assertIdentical(
333
self.serviceMaker.options, self.config.subOptions,
334
"ServiceMaker.makeService needs to be passed the correct "
335
"sub Command object.")
336
self.assertIdentical(
337
self.serviceMaker.service,
338
service.IService(arunner.application).services[0],
339
"ServiceMaker.makeService's result needs to be set as a child "
340
"of the Application.")
343
def test_preAndPostApplication(self):
345
Test thet preApplication and postApplication methods are
346
called by ApplicationRunner.run() when appropriate.
348
s = TestApplicationRunner(self.config)
350
self.assertFalse(s.hadApplicationPreApplication)
351
self.assertTrue(s.hadApplicationPostApplication)
352
self.assertTrue(s.hadApplicationLogObserver)
353
self.assertEquals(s.order, ["pre", "log", "post"])
356
def _applicationStartsWithConfiguredID(self, argv, uid, gid):
358
Assert that given a particular command line, an application is started
359
as a particular UID/GID.
361
@param argv: A list of strings giving the options to parse.
362
@param uid: An integer giving the expected UID.
363
@param gid: An integer giving the expected GID.
365
self.config.parseOptions(argv)
368
class FakeUnixApplicationRunner(twistd._SomeApplicationRunner):
369
def setupEnvironment(self, chroot, rundir, nodaemon, umask,
371
events.append('environment')
373
def shedPrivileges(self, euid, uid, gid):
374
events.append(('privileges', euid, uid, gid))
376
def startReactor(self, reactor, oldstdout, oldstderr):
377
events.append('reactor')
379
def removePID(self, pidfile):
383
class FakeService(object):
384
implements(service.IService, service.IProcess)
388
def privilegedStartService(self):
389
events.append('privilegedStartService')
391
def startService(self):
392
events.append('startService')
394
def stopService(self):
397
runner = FakeUnixApplicationRunner(self.config)
398
runner.preApplication()
399
runner.application = FakeService()
400
runner.postApplication()
404
['environment', 'privilegedStartService',
405
('privileges', False, uid, gid), 'startService', 'reactor'])
408
def test_applicationStartsWithConfiguredNumericIDs(self):
410
L{postApplication} should change the UID and GID to the values
411
specified as numeric strings by the configuration after running
412
L{service.IService.privilegedStartService} and before running
413
L{service.IService.startService}.
417
self._applicationStartsWithConfiguredID(
418
["--uid", str(uid), "--gid", str(gid)], uid, gid)
421
def test_applicationStartsWithConfiguredNameIDs(self):
423
L{postApplication} should change the UID and GID to the values
424
specified as user and group names by the configuration after running
425
L{service.IService.privilegedStartService} and before running
426
L{service.IService.startService}.
432
patchUserDatabase(self.patch, user, uid, group, gid)
433
self._applicationStartsWithConfiguredID(
434
["--uid", user, "--gid", group], uid, gid)
436
if getattr(os, 'setuid', None) is None:
437
msg = "Platform does not support --uid/--gid twistd options."
438
test_applicationStartsWithConfiguredNameIDs.skip = msg
439
test_applicationStartsWithConfiguredNumericIDs.skip = msg
443
def test_startReactorRunsTheReactor(self):
445
L{startReactor} calls L{reactor.run}.
447
reactor = DummyReactor()
448
runner = app.ApplicationRunner({
450
"profiler": "profile",
452
runner.startReactor(reactor, None, None)
454
reactor.called, "startReactor did not call reactor.run()")
457
def test_legacyApplicationRunnerGetLogObserver(self):
459
L{app.ApplicationRunner} subclasses can have a getLogObserver that used
460
to return a log observer. This test is there to ensure that it's
461
supported but it raises a warning when used.
464
self.addCleanup(log.removeObserver, observer.append)
465
class GetLogObserverRunner(app.ApplicationRunner):
466
def getLogObserver(self):
467
return observer.append
469
def startLogging(self, observer):
471
Override C{startLogging} to call L{log.addObserver} instead of
472
L{log.startLoggingWithObserver}.
474
log.addObserver(observer)
475
self.logger._initialLog()
477
def preApplication(self):
480
def postApplication(self):
483
def createOrGetApplication(self):
486
conf = twistd.ServerOptions()
487
runner = GetLogObserverRunner(conf)
488
self.assertWarns(DeprecationWarning,
489
"Specifying a log observer with getLogObserver is "
490
"deprecated. Please use a loggerFactory instead.",
491
app.__file__, runner.run)
492
self.assertEquals(len(observer), 3)
496
class UnixApplicationRunnerSetupEnvironmentTests(unittest.TestCase):
498
Tests for L{UnixApplicationRunner.setupEnvironment}.
500
@ivar root: The root of the filesystem, or C{unset} if none has been
501
specified with a call to L{os.chroot} (patched for this TestCase with
502
L{UnixApplicationRunnerSetupEnvironmentTests.chroot ).
504
@ivar cwd: The current working directory of the process, or C{unset} if
505
none has been specified with a call to L{os.chdir} (patched for this
506
TestCase with L{UnixApplicationRunnerSetupEnvironmentTests.chdir).
508
@ivar mask: The current file creation mask of the process, or C{unset} if
509
none has been specified with a call to L{os.umask} (patched for this
510
TestCase with L{UnixApplicationRunnerSetupEnvironmentTests.umask).
512
@ivar daemon: A boolean indicating whether daemonization has been performed
513
by a call to L{_twistd_unix.daemonize} (patched for this TestCase with
514
L{UnixApplicationRunnerSetupEnvironmentTests.
516
if _twistd_unix is None:
517
skip = "twistd unix not available"
522
self.root = self.unset
523
self.cwd = self.unset
524
self.mask = self.unset
526
self.pid = os.getpid()
527
self.patch(os, 'chroot', lambda path: setattr(self, 'root', path))
528
self.patch(os, 'chdir', lambda path: setattr(self, 'cwd', path))
529
self.patch(os, 'umask', lambda mask: setattr(self, 'mask', mask))
530
self.patch(_twistd_unix, "daemonize", self.daemonize)
531
self.runner = UnixApplicationRunner({})
536
Indicate that daemonization has happened and change the PID so that the
537
value written to the pidfile can be tested in the daemonization case.
540
self.patch(os, 'getpid', lambda: self.pid + 1)
543
def test_chroot(self):
545
L{UnixApplicationRunner.setupEnvironment} changes the root of the
546
filesystem if passed a non-C{None} value for the C{chroot} parameter.
548
self.runner.setupEnvironment("/foo/bar", ".", True, None, None)
549
self.assertEqual(self.root, "/foo/bar")
552
def test_noChroot(self):
554
L{UnixApplicationRunner.setupEnvironment} does not change the root of
555
the filesystem if passed C{None} for the C{chroot} parameter.
557
self.runner.setupEnvironment(None, ".", True, None, None)
558
self.assertIdentical(self.root, self.unset)
561
def test_changeWorkingDirectory(self):
563
L{UnixApplicationRunner.setupEnvironment} changes the working directory
564
of the process to the path given for the C{rundir} parameter.
566
self.runner.setupEnvironment(None, "/foo/bar", True, None, None)
567
self.assertEqual(self.cwd, "/foo/bar")
570
def test_daemonize(self):
572
L{UnixApplicationRunner.setupEnvironment} daemonizes the process if
573
C{False} is passed for the C{nodaemon} parameter.
575
self.runner.setupEnvironment(None, ".", False, None, None)
576
self.assertTrue(self.daemon)
579
def test_noDaemonize(self):
581
L{UnixApplicationRunner.setupEnvironment} does not daemonize the
582
process if C{True} is passed for the C{nodaemon} parameter.
584
self.runner.setupEnvironment(None, ".", True, None, None)
585
self.assertFalse(self.daemon)
588
def test_nonDaemonPIDFile(self):
590
L{UnixApplicationRunner.setupEnvironment} writes the process's PID to
591
the file specified by the C{pidfile} parameter.
593
pidfile = self.mktemp()
594
self.runner.setupEnvironment(None, ".", True, None, pidfile)
596
pid = int(fObj.read())
598
self.assertEqual(pid, self.pid)
601
def test_daemonPIDFile(self):
603
L{UnixApplicationRunner.setupEnvironment} writes the daemonized
604
process's PID to the file specified by the C{pidfile} parameter if
605
C{nodaemon} is C{False}.
607
pidfile = self.mktemp()
608
self.runner.setupEnvironment(None, ".", False, None, pidfile)
610
pid = int(fObj.read())
612
self.assertEqual(pid, self.pid + 1)
615
def test_umask(self):
617
L{UnixApplicationRunner.setupEnvironment} changes the process umask to
618
the value specified by the C{umask} parameter.
620
self.runner.setupEnvironment(None, ".", False, 123, None)
621
self.assertEqual(self.mask, 123)
624
def test_noDaemonizeNoUmask(self):
626
L{UnixApplicationRunner.setupEnvironment} doesn't change the process
627
umask if C{None} is passed for the C{umask} parameter and C{True} is
628
passed for the C{nodaemon} parameter.
630
self.runner.setupEnvironment(None, ".", True, None, None)
631
self.assertIdentical(self.mask, self.unset)
634
def test_daemonizedNoUmask(self):
636
L{UnixApplicationRunner.setupEnvironment} changes the process umask to
637
C{0077} if C{None} is passed for the C{umask} parameter and C{False} is
638
passed for the C{nodaemon} parameter.
640
self.runner.setupEnvironment(None, ".", False, None, None)
641
self.assertEqual(self.mask, 0077)
645
class UnixApplicationRunnerStartApplicationTests(unittest.TestCase):
647
Tests for L{UnixApplicationRunner.startApplication}.
649
if _twistd_unix is None:
650
skip = "twistd unix not available"
652
def test_setupEnvironment(self):
654
L{UnixApplicationRunner.startApplication} calls
655
L{UnixApplicationRunner.setupEnvironment} with the chroot, rundir,
656
nodaemon, umask, and pidfile parameters from the configuration it is
659
options = twistd.ServerOptions()
660
options.parseOptions([
663
'--chroot', '/foo/chroot',
664
'--rundir', '/foo/rundir',
665
'--pidfile', '/foo/pidfile'])
666
application = service.Application("test_setupEnvironment")
667
self.runner = UnixApplicationRunner(options)
670
def fakeSetupEnvironment(self, chroot, rundir, nodaemon, umask, pidfile):
671
args.extend((chroot, rundir, nodaemon, umask, pidfile))
675
inspect.getargspec(self.runner.setupEnvironment),
676
inspect.getargspec(fakeSetupEnvironment))
678
self.patch(UnixApplicationRunner, 'setupEnvironment', fakeSetupEnvironment)
679
self.patch(UnixApplicationRunner, 'shedPrivileges', lambda *a, **kw: None)
680
self.patch(app, 'startApplication', lambda *a, **kw: None)
681
self.runner.startApplication(application)
685
['/foo/chroot', '/foo/rundir', True, 56, '/foo/pidfile'])
689
class UnixApplicationRunnerRemovePID(unittest.TestCase):
691
Tests for L{UnixApplicationRunner.removePID}.
693
if _twistd_unix is None:
694
skip = "twistd unix not available"
697
def test_removePID(self):
699
L{UnixApplicationRunner.removePID} deletes the file the name of
700
which is passed to it.
702
runner = UnixApplicationRunner({})
705
pidfile = os.path.join(path, "foo.pid")
706
file(pidfile, "w").close()
707
runner.removePID(pidfile)
708
self.assertFalse(os.path.exists(pidfile))
711
def test_removePIDErrors(self):
713
Calling L{UnixApplicationRunner.removePID} with a non-existent filename logs
716
runner = UnixApplicationRunner({})
717
runner.removePID("fakepid")
718
errors = self.flushLoggedErrors(OSError)
719
self.assertEquals(len(errors), 1)
720
self.assertEquals(errors[0].value.errno, errno.ENOENT)
724
class DummyReactor(object):
726
A dummy reactor, only providing a C{run} method and checking that it
729
@ivar called: if C{run} has been called or not.
730
@type called: C{bool}
736
A fake run method, checking that it's been called one and only time.
739
raise RuntimeError("Already called")
744
class AppProfilingTestCase(unittest.TestCase):
746
Tests for L{app.AppProfiler}.
749
def test_profile(self):
751
L{app.ProfileRunner.run} should call the C{run} method of the reactor
752
and save profile data in the specified file.
754
config = twistd.ServerOptions()
755
config["profile"] = self.mktemp()
756
config["profiler"] = "profile"
757
profiler = app.AppProfiler(config)
758
reactor = DummyReactor()
760
profiler.run(reactor)
762
self.assertTrue(reactor.called)
763
data = file(config["profile"]).read()
764
self.assertIn("DummyReactor.run", data)
765
self.assertIn("function calls", data)
768
test_profile.skip = "profile module not available"
771
def _testStats(self, statsClass, profile):
772
out = StringIO.StringIO()
774
# Patch before creating the pstats, because pstats binds self.stream to
775
# sys.stdout early in 2.5 and newer.
776
stdout = self.patch(sys, 'stdout', out)
778
# If pstats.Stats can load the data and then reformat it, then the
779
# right thing probably happened.
780
stats = statsClass(profile)
784
data = out.getvalue()
785
self.assertIn("function calls", data)
786
self.assertIn("(run)", data)
789
def test_profileSaveStats(self):
791
With the C{savestats} option specified, L{app.ProfileRunner.run}
792
should save the raw stats object instead of a summary output.
794
config = twistd.ServerOptions()
795
config["profile"] = self.mktemp()
796
config["profiler"] = "profile"
797
config["savestats"] = True
798
profiler = app.AppProfiler(config)
799
reactor = DummyReactor()
801
profiler.run(reactor)
803
self.assertTrue(reactor.called)
804
self._testStats(pstats.Stats, config['profile'])
807
test_profileSaveStats.skip = "profile module not available"
810
def test_withoutProfile(self):
812
When the C{profile} module is not present, L{app.ProfilerRunner.run}
813
should raise a C{SystemExit} exception.
815
savedModules = sys.modules.copy()
817
config = twistd.ServerOptions()
818
config["profiler"] = "profile"
819
profiler = app.AppProfiler(config)
821
sys.modules["profile"] = None
823
self.assertRaises(SystemExit, profiler.run, None)
826
sys.modules.update(savedModules)
829
def test_profilePrintStatsError(self):
831
When an error happens during the print of the stats, C{sys.stdout}
832
should be restored to its initial value.
834
class ErroneousProfile(profile.Profile):
835
def print_stats(self):
836
raise RuntimeError("Boom")
837
self.patch(profile, "Profile", ErroneousProfile)
839
config = twistd.ServerOptions()
840
config["profile"] = self.mktemp()
841
config["profiler"] = "profile"
842
profiler = app.AppProfiler(config)
843
reactor = DummyReactor()
845
oldStdout = sys.stdout
846
self.assertRaises(RuntimeError, profiler.run, reactor)
847
self.assertIdentical(sys.stdout, oldStdout)
850
test_profilePrintStatsError.skip = "profile module not available"
853
def test_hotshot(self):
855
L{app.HotshotRunner.run} should call the C{run} method of the reactor
856
and save profile data in the specified file.
858
config = twistd.ServerOptions()
859
config["profile"] = self.mktemp()
860
config["profiler"] = "hotshot"
861
profiler = app.AppProfiler(config)
862
reactor = DummyReactor()
864
profiler.run(reactor)
866
self.assertTrue(reactor.called)
867
data = file(config["profile"]).read()
868
self.assertIn("run", data)
869
self.assertIn("function calls", data)
872
test_hotshot.skip = "hotshot module not available"
875
def test_hotshotSaveStats(self):
877
With the C{savestats} option specified, L{app.HotshotRunner.run} should
878
save the raw stats object instead of a summary output.
880
config = twistd.ServerOptions()
881
config["profile"] = self.mktemp()
882
config["profiler"] = "hotshot"
883
config["savestats"] = True
884
profiler = app.AppProfiler(config)
885
reactor = DummyReactor()
887
profiler.run(reactor)
889
self.assertTrue(reactor.called)
890
self._testStats(hotshot.stats.load, config['profile'])
893
test_hotshotSaveStats.skip = "hotshot module not available"
896
def test_withoutHotshot(self):
898
When the C{hotshot} module is not present, L{app.HotshotRunner.run}
899
should raise a C{SystemExit} exception and log the C{ImportError}.
901
savedModules = sys.modules.copy()
902
sys.modules["hotshot"] = None
904
config = twistd.ServerOptions()
905
config["profiler"] = "hotshot"
906
profiler = app.AppProfiler(config)
908
self.assertRaises(SystemExit, profiler.run, None)
911
sys.modules.update(savedModules)
914
def test_nothotshotDeprecation(self):
916
Check that switching on the C{nothotshot} option produces a warning and
917
sets the profiler to B{profile}.
919
config = twistd.ServerOptions()
920
config['nothotshot'] = True
921
profiler = self.assertWarns(DeprecationWarning,
922
"The --nothotshot option is deprecated. Please specify the "
923
"profiler name using the --profiler option",
924
app.__file__, app.AppProfiler, config)
925
self.assertEquals(profiler.profiler, "profile")
928
def test_hotshotPrintStatsError(self):
930
When an error happens while printing the stats, C{sys.stdout}
931
should be restored to its initial value.
933
class ErroneousStats(pstats.Stats):
934
def print_stats(self):
935
raise RuntimeError("Boom")
936
self.patch(pstats, "Stats", ErroneousStats)
938
config = twistd.ServerOptions()
939
config["profile"] = self.mktemp()
940
config["profiler"] = "hotshot"
941
profiler = app.AppProfiler(config)
942
reactor = DummyReactor()
944
oldStdout = sys.stdout
945
self.assertRaises(RuntimeError, profiler.run, reactor)
946
self.assertIdentical(sys.stdout, oldStdout)
949
test_hotshotPrintStatsError.skip = "hotshot module not available"
952
def test_cProfile(self):
954
L{app.CProfileRunner.run} should call the C{run} method of the
955
reactor and save profile data in the specified file.
957
config = twistd.ServerOptions()
958
config["profile"] = self.mktemp()
959
config["profiler"] = "cProfile"
960
profiler = app.AppProfiler(config)
961
reactor = DummyReactor()
963
profiler.run(reactor)
965
self.assertTrue(reactor.called)
966
data = file(config["profile"]).read()
967
self.assertIn("run", data)
968
self.assertIn("function calls", data)
971
test_cProfile.skip = "cProfile module not available"
974
def test_cProfileSaveStats(self):
976
With the C{savestats} option specified,
977
L{app.CProfileRunner.run} should save the raw stats object
978
instead of a summary output.
980
config = twistd.ServerOptions()
981
config["profile"] = self.mktemp()
982
config["profiler"] = "cProfile"
983
config["savestats"] = True
984
profiler = app.AppProfiler(config)
985
reactor = DummyReactor()
987
profiler.run(reactor)
989
self.assertTrue(reactor.called)
990
self._testStats(pstats.Stats, config['profile'])
993
test_cProfileSaveStats.skip = "cProfile module not available"
996
def test_withoutCProfile(self):
998
When the C{cProfile} module is not present,
999
L{app.CProfileRunner.run} should raise a C{SystemExit}
1000
exception and log the C{ImportError}.
1002
savedModules = sys.modules.copy()
1003
sys.modules["cProfile"] = None
1005
config = twistd.ServerOptions()
1006
config["profiler"] = "cProfile"
1007
profiler = app.AppProfiler(config)
1009
self.assertRaises(SystemExit, profiler.run, None)
1012
sys.modules.update(savedModules)
1015
def test_unknownProfiler(self):
1017
Check that L{app.AppProfiler} raises L{SystemExit} when given an
1018
unknown profiler name.
1020
config = twistd.ServerOptions()
1021
config["profile"] = self.mktemp()
1022
config["profiler"] = "foobar"
1024
error = self.assertRaises(SystemExit, app.AppProfiler, config)
1025
self.assertEquals(str(error), "Unsupported profiler name: foobar")
1028
def test_defaultProfiler(self):
1030
L{app.Profiler} defaults to the hotshot profiler if not specified.
1032
profiler = app.AppProfiler({})
1033
self.assertEquals(profiler.profiler, "hotshot")
1036
def test_profilerNameCaseInsentive(self):
1038
The case of the profiler name passed to L{app.AppProfiler} is not
1041
profiler = app.AppProfiler({"profiler": "HotShot"})
1042
self.assertEquals(profiler.profiler, "hotshot")
1045
def test_oldRunWithProfiler(self):
1047
L{app.runWithProfiler} should print a C{DeprecationWarning} pointing
1050
class DummyProfiler(object):
1052
def run(self, reactor):
1054
profiler = DummyProfiler()
1055
self.patch(app, "AppProfiler", lambda conf: profiler)
1057
def runWithProfiler():
1058
return app.runWithProfiler(DummyReactor(), {})
1060
self.assertWarns(DeprecationWarning,
1061
"runWithProfiler is deprecated since Twisted 8.0. "
1062
"Use ProfileRunner instead.", __file__,
1064
self.assertTrue(profiler.called)
1067
def test_oldRunWithHotshot(self):
1069
L{app.runWithHotshot} should print a C{DeprecationWarning} pointing
1072
class DummyProfiler(object):
1074
def run(self, reactor):
1076
profiler = DummyProfiler()
1077
self.patch(app, "AppProfiler", lambda conf: profiler)
1079
def runWithHotshot():
1080
return app.runWithHotshot(DummyReactor(), {})
1082
self.assertWarns(DeprecationWarning,
1083
"runWithHotshot is deprecated since Twisted 8.0. "
1084
"Use HotshotRunner instead.", __file__,
1086
self.assertTrue(profiler.called)
1090
def _patchFileLogObserver(patch):
1092
Patch L{log.FileLogObserver} to record every call and keep a reference to
1093
the passed log file for tests.
1095
@param patch: a callback for patching (usually L{unittest.TestCase.patch}).
1097
@return: the list that keeps track of the log files.
1101
oldFileLobObserver = log.FileLogObserver
1102
def FileLogObserver(logFile):
1103
logFiles.append(logFile)
1104
return oldFileLobObserver(logFile)
1105
patch(log, 'FileLogObserver', FileLogObserver)
1110
class AppLoggerTestCase(unittest.TestCase):
1112
Tests for L{app.AppLogger}.
1114
@ivar observers: list of observers installed during the tests.
1115
@type observers: C{list}
1120
Override L{log.addObserver} so that we can trace the observers
1121
installed in C{self.observers}.
1124
def startLoggingWithObserver(observer):
1125
self.observers.append(observer)
1126
log.addObserver(observer)
1127
self.patch(log, 'startLoggingWithObserver', startLoggingWithObserver)
1132
Remove all installed observers.
1134
for observer in self.observers:
1135
log.removeObserver(observer)
1138
def _checkObserver(self, logs):
1140
Ensure that initial C{twistd} logs are written to the given list.
1143
@param logs: The list whose C{append} method was specified as the
1144
initial log observer.
1146
self.assertEquals(self.observers, [logs.append])
1147
self.assertIn("starting up", logs[0]["message"][0])
1148
self.assertIn("reactor class", logs[1]["message"][0])
1151
def test_start(self):
1153
L{app.AppLogger.start} calls L{log.addObserver}, and then writes some
1154
messages about twistd and the reactor.
1156
logger = app.AppLogger({})
1158
logger._getLogObserver = lambda: observer.append
1159
logger.start(Componentized())
1160
self._checkObserver(observer)
1163
def test_startUsesApplicationLogObserver(self):
1165
When the L{ILogObserver} component is available on the application,
1166
that object will be used as the log observer instead of constructing a
1169
application = Componentized()
1171
application.setComponent(ILogObserver, logs.append)
1172
logger = app.AppLogger({})
1173
logger.start(application)
1174
self._checkObserver(logs)
1177
def test_getLogObserverStdout(self):
1179
When logfile is empty or set to C{-}, L{app.AppLogger._getLogObserver}
1180
returns a log observer pointing at C{sys.stdout}.
1182
logger = app.AppLogger({"logfile": "-"})
1183
logFiles = _patchFileLogObserver(self.patch)
1185
observer = logger._getLogObserver()
1187
self.assertEquals(len(logFiles), 1)
1188
self.assertIdentical(logFiles[0], sys.stdout)
1190
logger = app.AppLogger({"logfile": ""})
1191
observer = logger._getLogObserver()
1193
self.assertEquals(len(logFiles), 2)
1194
self.assertIdentical(logFiles[1], sys.stdout)
1197
def test_getLogObserverFile(self):
1199
When passing the C{logfile} option, L{app.AppLogger._getLogObserver}
1200
returns a log observer pointing at the specified path.
1202
logFiles = _patchFileLogObserver(self.patch)
1203
filename = self.mktemp()
1204
logger = app.AppLogger({"logfile": filename})
1206
observer = logger._getLogObserver()
1208
self.assertEquals(len(logFiles), 1)
1209
self.assertEquals(logFiles[0].path,
1210
os.path.abspath(filename))
1213
def test_stop(self):
1215
L{app.AppLogger.stop} removes the observer created in C{start}, and
1216
reinitialize its C{_observer} so that if C{stop} is called several
1217
times it doesn't break.
1221
def remove(observer):
1222
removed.append(observer)
1223
self.patch(log, 'removeObserver', remove)
1224
logger = app.AppLogger({})
1225
logger._observer = observer
1227
self.assertEquals(removed, [observer])
1229
self.assertEquals(removed, [observer])
1230
self.assertIdentical(logger._observer, None)
1234
class UnixAppLoggerTestCase(unittest.TestCase):
1236
Tests for L{UnixAppLogger}.
1238
@ivar signals: list of signal handlers installed.
1239
@type signals: C{list}
1241
if _twistd_unix is None:
1242
skip = "twistd unix not available"
1246
Fake C{signal.signal} for not installing the handlers but saving them
1250
def fakeSignal(sig, f):
1251
self.signals.append((sig, f))
1252
self.patch(signal, "signal", fakeSignal)
1255
def test_getLogObserverStdout(self):
1257
When non-daemonized and C{logfile} is empty or set to C{-},
1258
L{UnixAppLogger._getLogObserver} returns a log observer pointing at
1261
logFiles = _patchFileLogObserver(self.patch)
1263
logger = UnixAppLogger({"logfile": "-", "nodaemon": True})
1264
observer = logger._getLogObserver()
1265
self.assertEquals(len(logFiles), 1)
1266
self.assertIdentical(logFiles[0], sys.stdout)
1268
logger = UnixAppLogger({"logfile": "", "nodaemon": True})
1269
observer = logger._getLogObserver()
1270
self.assertEquals(len(logFiles), 2)
1271
self.assertIdentical(logFiles[1], sys.stdout)
1274
def test_getLogObserverStdoutDaemon(self):
1276
When daemonized and C{logfile} is set to C{-},
1277
L{UnixAppLogger._getLogObserver} raises C{SystemExit}.
1279
logger = UnixAppLogger({"logfile": "-", "nodaemon": False})
1280
error = self.assertRaises(SystemExit, logger._getLogObserver)
1281
self.assertEquals(str(error), "Daemons cannot log to stdout, exiting!")
1284
def test_getLogObserverFile(self):
1286
When C{logfile} contains a file name, L{app.AppLogger._getLogObserver}
1287
returns a log observer pointing at the specified path, and a signal
1288
handler rotating the log is installed.
1290
logFiles = _patchFileLogObserver(self.patch)
1291
filename = self.mktemp()
1292
logger = UnixAppLogger({"logfile": filename})
1293
observer = logger._getLogObserver()
1295
self.assertEquals(len(logFiles), 1)
1296
self.assertEquals(logFiles[0].path,
1297
os.path.abspath(filename))
1299
self.assertEquals(len(self.signals), 1)
1300
self.assertEquals(self.signals[0][0], signal.SIGUSR1)
1305
logFiles[0].rotate = rotate
1307
rotateLog = self.signals[0][1]
1308
rotateLog(None, None)
1312
def test_getLogObserverDontOverrideSignalHandler(self):
1314
If a signal handler is already installed,
1315
L{UnixAppLogger._getLogObserver} doesn't override it.
1317
def fakeGetSignal(sig):
1318
self.assertEquals(sig, signal.SIGUSR1)
1320
self.patch(signal, "getsignal", fakeGetSignal)
1321
filename = self.mktemp()
1322
logger = UnixAppLogger({"logfile": filename})
1323
observer = logger._getLogObserver()
1325
self.assertEquals(self.signals, [])
1328
def test_getLogObserverDefaultFile(self):
1330
When daemonized and C{logfile} is empty, the observer returned by
1331
L{UnixAppLogger._getLogObserver} points at C{twistd.log} in the current
1334
logFiles = _patchFileLogObserver(self.patch)
1335
logger = UnixAppLogger({"logfile": "", "nodaemon": False})
1336
observer = logger._getLogObserver()
1338
self.assertEquals(len(logFiles), 1)
1339
self.assertEquals(logFiles[0].path,
1340
os.path.abspath("twistd.log"))
1343
def test_getLogObserverSyslog(self):
1345
If C{syslog} is set to C{True}, L{UnixAppLogger._getLogObserver} starts
1346
a L{syslog.SyslogObserver} with given C{prefix}.
1348
class fakesyslogobserver(object):
1349
def __init__(self, prefix):
1350
fakesyslogobserver.prefix = prefix
1351
def emit(self, eventDict):
1353
self.patch(syslog, "SyslogObserver", fakesyslogobserver)
1354
logger = UnixAppLogger({"syslog": True, "prefix": "test-prefix"})
1355
observer = logger._getLogObserver()
1356
self.assertEquals(fakesyslogobserver.prefix, "test-prefix")
1359
test_getLogObserverSyslog.skip = "Syslog not available"
1363
class DeprecationTests(unittest.TestCase):
1365
Tests for deprecated features.
1368
def test_initialLog(self):
1370
L{app.initialLog} is deprecated.
1373
log.addObserver(logs.append)
1374
self.addCleanup(log.removeObserver, logs.append)
1375
self.callDeprecated(Version("Twisted", 8, 2, 0), app.initialLog)
1376
self.assertEquals(len(logs), 2)
1377
self.assertIn("starting up", logs[0]["message"][0])
1378
self.assertIn("reactor class", logs[1]["message"][0])
added
removed
vendor/Twisted-10.0.0/twisted/test/test_twistd.py
