twisted/runner/test/test_procmon.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 """
   5 Tests for L{twisted.runner.procmon}.
   6 """
   7
   8 from twisted.trial import unittest
   9 from twisted.runner.procmon import LoggingProtocol, ProcessMonitor
  10 from twisted.internet.error import (ProcessDone, ProcessTerminated,
  11                                     ProcessExitedAlready)
  12 from twisted.internet.task import Clock
  13 from twisted.python.failure import Failure
  14 from twisted.test.proto_helpers import MemoryReactor
  15
  16
  17
  18 class DummyProcess(object):
  19     """
  20     An incomplete and fake L{IProcessTransport} implementation for testing how
  21     L{ProcessMonitor} behaves when its monitored processes exit.
  22
  23     @ivar _terminationDelay: the delay in seconds after which the DummyProcess
  24         will appear to exit when it receives a TERM signal
  25     """
  26
  27     pid = 1
  28     proto = None
  29
  30     _terminationDelay = 1
  31
  32     def __init__(self, reactor, executable, args, environment, path,
  33                  proto, uid=None, gid=None, usePTY=0, childFDs=None):
  34
  35         self.proto = proto
  36
  37         self._reactor = reactor
  38         self._executable = executable
  39         self._args = args
  40         self._environment = environment
  41         self._path = path
  42         self._uid = uid
  43         self._gid = gid
  44         self._usePTY = usePTY
  45         self._childFDs = childFDs
  46
  47
  48     def signalProcess(self, signalID):
  49         """
  50         A partial implementation of signalProcess which can only handle TERM and
  51         KILL signals.
  52          - When a TERM signal is given, the dummy process will appear to exit
  53            after L{DummyProcess._terminationDelay} seconds with exit code 0
  54          - When a KILL signal is given, the dummy process will appear to exit
  55            immediately with exit code 1.
  56
  57         @param signalID: The signal name or number to be issued to the process.
  58         @type signalID: C{str}
  59         """
  60         params = {
  61             "TERM": (self._terminationDelay, 0),
  62             "KILL": (0, 1)
  63         }
  64
  65         if self.pid is None:
  66             raise ProcessExitedAlready()
  67
  68         if signalID in params:
  69             delay, status = params[signalID]
  70             self._signalHandler = self._reactor.callLater(
  71                 delay, self.processEnded, status)
  72
  73
  74     def processEnded(self, status):
  75         """
  76         Deliver the process ended event to C{self.proto}.
  77         """
  78         self.pid = None
  79         statusMap = {
  80             0: ProcessDone,
  81             1: ProcessTerminated,
  82         }
  83         self.proto.processEnded(Failure(statusMap[status](status)))
  84
  85
  86
  87 class DummyProcessReactor(MemoryReactor, Clock):
  88     """
  89     @ivar spawnedProcesses: a list that keeps track of the fake process
  90         instances built by C{spawnProcess}.
  91     @type spawnedProcesses: C{list}
  92     """
  93     def __init__(self):
  94         MemoryReactor.__init__(self)
  95         Clock.__init__(self)
  96
  97         self.spawnedProcesses = []
  98
  99
 100     def spawnProcess(self, processProtocol, executable, args=(), env={},
 101                      path=None, uid=None, gid=None, usePTY=0,
 102                      childFDs=None):
 103         """
 104         Fake L{reactor.spawnProcess}, that logs all the process
 105         arguments and returns a L{DummyProcess}.
 106         """
 107
 108         proc = DummyProcess(self, executable, args, env, path,
 109                             processProtocol, uid, gid, usePTY, childFDs)
 110         processProtocol.makeConnection(proc)
 111         self.spawnedProcesses.append(proc)
 112         return proc
 113
 114
 115
 116 class ProcmonTests(unittest.TestCase):
 117     """
 118     Tests for L{ProcessMonitor}.
 119     """
 120
 121     def setUp(self):
 122         """
 123         Create an L{ProcessMonitor} wrapped around a fake reactor.
 124         """
 125         self.reactor = DummyProcessReactor()
 126         self.pm = ProcessMonitor(reactor=self.reactor)
 127         self.pm.minRestartDelay = 2
 128         self.pm.maxRestartDelay = 10
 129         self.pm.threshold = 10
 130
 131
 132     def test_getStateIncludesProcesses(self):
 133         """
 134         The list of monitored processes must be included in the pickle state.
 135         """
 136         self.pm.addProcess("foo", ["arg1", "arg2"],
 137                            uid=1, gid=2, env={})
 138         self.assertEqual(self.pm.__getstate__()['processes'],
 139                           {'foo': (['arg1', 'arg2'], 1, 2, {})})
 140
 141
 142     def test_getStateExcludesReactor(self):
 143         """
 144         The private L{ProcessMonitor._reactor} instance variable should not be
 145         included in the pickle state.
 146         """
 147         self.assertNotIn('_reactor', self.pm.__getstate__())
 148
 149
 150     def test_addProcess(self):
 151         """
 152         L{ProcessMonitor.addProcess} only starts the named program if
 153         L{ProcessMonitor.startService} has been called.
 154         """
 155         self.pm.addProcess("foo", ["arg1", "arg2"],
 156                            uid=1, gid=2, env={})
 157         self.assertEqual(self.pm.protocols, {})
 158         self.assertEqual(self.pm.processes,
 159                           {"foo": (["arg1", "arg2"], 1, 2, {})})
 160         self.pm.startService()
 161         self.reactor.advance(0)
 162         self.assertEqual(self.pm.protocols.keys(), ["foo"])
 163
 164
 165     def test_addProcessDuplicateKeyError(self):
 166         """
 167         L{ProcessMonitor.addProcess} raises a C{KeyError} if a process with the
 168         given name already exists.
 169         """
 170         self.pm.addProcess("foo", ["arg1", "arg2"],
 171                            uid=1, gid=2, env={})
 172         self.assertRaises(KeyError, self.pm.addProcess,
 173                           "foo", ["arg1", "arg2"], uid=1, gid=2, env={})
 174
 175
 176     def test_addProcessEnv(self):
 177         """
 178         L{ProcessMonitor.addProcess} takes an C{env} parameter that is passed to
 179         L{IReactorProcess.spawnProcess}.
 180         """
 181         fakeEnv = {"KEY": "value"}
 182         self.pm.startService()
 183         self.pm.addProcess("foo", ["foo"], uid=1, gid=2, env=fakeEnv)
 184         self.reactor.advance(0)
 185         self.assertEqual(
 186             self.reactor.spawnedProcesses[0]._environment, fakeEnv)
 187
 188
 189     def test_removeProcess(self):
 190         """
 191         L{ProcessMonitor.removeProcess} removes the process from the public
 192         processes list.
 193         """
 194         self.pm.startService()
 195         self.pm.addProcess("foo", ["foo"])
 196         self.assertEqual(len(self.pm.processes), 1)
 197         self.pm.removeProcess("foo")
 198         self.assertEqual(len(self.pm.processes), 0)
 199
 200
 201     def test_removeProcessUnknownKeyError(self):
 202         """
 203         L{ProcessMonitor.removeProcess} raises a C{KeyError} if the given
 204         process name isn't recognised.
 205         """
 206         self.pm.startService()
 207         self.assertRaises(KeyError, self.pm.removeProcess, "foo")
 208
 209
 210     def test_startProcess(self):
 211         """
 212         When a process has been started, an instance of L{LoggingProtocol} will
 213         be added to the L{ProcessMonitor.protocols} dict and the start time of
 214         the process will be recorded in the L{ProcessMonitor.timeStarted}
 215         dictionary.
 216         """
 217         self.pm.addProcess("foo", ["foo"])
 218         self.pm.startProcess("foo")
 219         self.assertIsInstance(self.pm.protocols["foo"], LoggingProtocol)
 220         self.assertIn("foo", self.pm.timeStarted.keys())
 221
 222
 223     def test_startProcessAlreadyStarted(self):
 224         """
 225         L{ProcessMonitor.startProcess} silently returns if the named process is
 226         already started.
 227         """
 228         self.pm.addProcess("foo", ["foo"])
 229         self.pm.startProcess("foo")
 230         self.assertIdentical(None, self.pm.startProcess("foo"))
 231
 232
 233     def test_startProcessUnknownKeyError(self):
 234         """
 235         L{ProcessMonitor.startProcess} raises a C{KeyError} if the given
 236         process name isn't recognised.
 237         """
 238         self.assertRaises(KeyError, self.pm.startProcess, "foo")
 239
 240
 241     def test_stopProcessNaturalTermination(self):
 242         """
 243         L{ProcessMonitor.stopProcess} immediately sends a TERM signal to the
 244         named process.
 245         """
 246         self.pm.startService()
 247         self.pm.addProcess("foo", ["foo"])
 248         self.assertIn("foo", self.pm.protocols)
 249
 250         # Configure fake process to die 1 second after receiving term signal
 251         timeToDie = self.pm.protocols["foo"].transport._terminationDelay = 1
 252
 253         # Advance the reactor to just before the short lived process threshold
 254         # and leave enough time for the process to die
 255         self.reactor.advance(self.pm.threshold)
 256         # Then signal the process to stop
 257         self.pm.stopProcess("foo")
 258
 259         # Advance the reactor just enough to give the process time to die and
 260         # verify that the process restarts
 261         self.reactor.advance(timeToDie)
 262
 263         # We expect it to be restarted immediately
 264         self.assertEqual(self.reactor.seconds(),
 265                          self.pm.timeStarted["foo"])
 266
 267
 268     def test_stopProcessForcedKill(self):
 269         """
 270         L{ProcessMonitor.stopProcess} kills a process which fails to terminate
 271         naturally within L{ProcessMonitor.killTime} seconds.
 272         """
 273         self.pm.startService()
 274         self.pm.addProcess("foo", ["foo"])
 275         self.assertIn("foo", self.pm.protocols)
 276         self.reactor.advance(self.pm.threshold)
 277         proc = self.pm.protocols["foo"].transport
 278         # Arrange for the fake process to live longer than the killTime
 279         proc._terminationDelay = self.pm.killTime + 1
 280         self.pm.stopProcess("foo")
 281         # If process doesn't die before the killTime, procmon should
 282         # terminate it
 283         self.reactor.advance(self.pm.killTime - 1)
 284         self.assertEqual(0.0, self.pm.timeStarted["foo"])
 285
 286         self.reactor.advance(1)
 287         # We expect it to be immediately restarted
 288         self.assertEqual(self.reactor.seconds(), self.pm.timeStarted["foo"])
 289
 290
 291     def test_stopProcessUnknownKeyError(self):
 292         """
 293         L{ProcessMonitor.stopProcess} raises a C{KeyError} if the given process
 294         name isn't recognised.
 295         """
 296         self.assertRaises(KeyError, self.pm.stopProcess, "foo")
 297
 298
 299     def test_stopProcessAlreadyStopped(self):
 300         """
 301         L{ProcessMonitor.stopProcess} silently returns if the named process
 302         is already stopped. eg Process has crashed and a restart has been
 303         rescheduled, but in the meantime, the service is stopped.
 304         """
 305         self.pm.addProcess("foo", ["foo"])
 306         self.assertIdentical(None, self.pm.stopProcess("foo"))
 307
 308
 309     def test_connectionLostLongLivedProcess(self):
 310         """
 311         L{ProcessMonitor.connectionLost} should immediately restart a process
 312         if it has been running longer than L{ProcessMonitor.threshold} seconds.
 313         """
 314         self.pm.addProcess("foo", ["foo"])
 315         # Schedule the process to start
 316         self.pm.startService()
 317         # advance the reactor to start the process
 318         self.reactor.advance(0)
 319         self.assertIn("foo", self.pm.protocols)
 320         # Long time passes
 321         self.reactor.advance(self.pm.threshold)
 322         # Process dies after threshold
 323         self.pm.protocols["foo"].processEnded(Failure(ProcessDone(0)))
 324         self.assertNotIn("foo", self.pm.protocols)
 325         # Process should be restarted immediately
 326         self.reactor.advance(0)
 327         self.assertIn("foo", self.pm.protocols)
 328
 329
 330     def test_connectionLostMurderCancel(self):
 331         """
 332         L{ProcessMonitor.connectionLost} cancels a scheduled process killer and
 333         deletes the DelayedCall from the L{ProcessMonitor.murder} list.
 334         """
 335         self.pm.addProcess("foo", ["foo"])
 336         # Schedule the process to start
 337         self.pm.startService()
 338         # Advance 1s to start the process then ask ProcMon to stop it
 339         self.reactor.advance(1)
 340         self.pm.stopProcess("foo")
 341         # A process killer has been scheduled, delayedCall is active
 342         self.assertIn("foo", self.pm.murder)
 343         delayedCall = self.pm.murder["foo"]
 344         self.assertTrue(delayedCall.active())
 345         # Advance to the point at which the dummy process exits
 346         self.reactor.advance(
 347             self.pm.protocols["foo"].transport._terminationDelay)
 348         # Now the delayedCall has been cancelled and deleted
 349         self.assertFalse(delayedCall.active())
 350         self.assertNotIn("foo", self.pm.murder)
 351
 352
 353     def test_connectionLostProtocolDeletion(self):
 354         """
 355         L{ProcessMonitor.connectionLost} removes the corresponding
 356         ProcessProtocol instance from the L{ProcessMonitor.protocols} list.
 357         """
 358         self.pm.startService()
 359         self.pm.addProcess("foo", ["foo"])
 360         self.assertIn("foo", self.pm.protocols)
 361         self.pm.protocols["foo"].transport.signalProcess("KILL")
 362         self.reactor.advance(
 363             self.pm.protocols["foo"].transport._terminationDelay)
 364         self.assertNotIn("foo", self.pm.protocols)
 365
 366
 367     def test_connectionLostMinMaxRestartDelay(self):
 368         """
 369         L{ProcessMonitor.connectionLost} will wait at least minRestartDelay s
 370         and at most maxRestartDelay s
 371         """
 372         self.pm.minRestartDelay = 2
 373         self.pm.maxRestartDelay = 3
 374
 375         self.pm.startService()
 376         self.pm.addProcess("foo", ["foo"])
 377
 378         self.assertEqual(self.pm.delay["foo"], self.pm.minRestartDelay)
 379         self.reactor.advance(self.pm.threshold - 1)
 380         self.pm.protocols["foo"].processEnded(Failure(ProcessDone(0)))
 381         self.assertEqual(self.pm.delay["foo"], self.pm.maxRestartDelay)
 382
 383
 384     def test_connectionLostBackoffDelayDoubles(self):
 385         """
 386         L{ProcessMonitor.connectionLost} doubles the restart delay each time
 387         the process dies too quickly.
 388         """
 389         self.pm.startService()
 390         self.pm.addProcess("foo", ["foo"])
 391         self.reactor.advance(self.pm.threshold - 1) #9s
 392         self.assertIn("foo", self.pm.protocols)
 393         self.assertEqual(self.pm.delay["foo"], self.pm.minRestartDelay)
 394         # process dies within the threshold and should not restart immediately
 395         self.pm.protocols["foo"].processEnded(Failure(ProcessDone(0)))
 396         self.assertEqual(self.pm.delay["foo"], self.pm.minRestartDelay * 2)
 397
 398
 399     def test_startService(self):
 400         """
 401         L{ProcessMonitor.startService} starts all monitored processes.
 402         """
 403         self.pm.addProcess("foo", ["foo"])
 404         # Schedule the process to start
 405         self.pm.startService()
 406         # advance the reactor to start the process
 407         self.reactor.advance(0)
 408         self.assertTrue("foo" in self.pm.protocols)
 409
 410
 411     def test_stopService(self):
 412         """
 413         L{ProcessMonitor.stopService} should stop all monitored processes.
 414         """
 415         self.pm.addProcess("foo", ["foo"])
 416         self.pm.addProcess("bar", ["bar"])
 417         # Schedule the process to start
 418         self.pm.startService()
 419         # advance the reactor to start the processes
 420         self.reactor.advance(self.pm.threshold)
 421         self.assertIn("foo", self.pm.protocols)
 422         self.assertIn("bar", self.pm.protocols)
 423
 424         self.reactor.advance(1)
 425
 426         self.pm.stopService()
 427         # Advance to beyond the killTime - all monitored processes
 428         # should have exited
 429         self.reactor.advance(self.pm.killTime + 1)
 430         # The processes shouldn't be restarted
 431         self.assertEqual({}, self.pm.protocols)
 432
 433
 434     def test_stopServiceCancelRestarts(self):
 435         """
 436         L{ProcessMonitor.stopService} should cancel any scheduled process
 437         restarts.
 438         """
 439         self.pm.addProcess("foo", ["foo"])
 440         # Schedule the process to start
 441         self.pm.startService()
 442         # advance the reactor to start the processes
 443         self.reactor.advance(self.pm.threshold)
 444         self.assertIn("foo", self.pm.protocols)
 445
 446         self.reactor.advance(1)
 447         # Kill the process early
 448         self.pm.protocols["foo"].processEnded(Failure(ProcessDone(0)))
 449         self.assertTrue(self.pm.restart['foo'].active())
 450         self.pm.stopService()
 451         # Scheduled restart should have been cancelled
 452         self.assertFalse(self.pm.restart['foo'].active())
 453
 454
 455     def test_stopServiceCleanupScheduledRestarts(self):
 456         """
 457         L{ProcessMonitor.stopService} should cancel all scheduled process
 458         restarts.
 459         """
 460         self.pm.threshold = 5
 461         self.pm.minRestartDelay = 5
 462         # Start service and add a process (started immediately)
 463         self.pm.startService()
 464         self.pm.addProcess("foo", ["foo"])
 465         # Stop the process after 1s
 466         self.reactor.advance(1)
 467         self.pm.stopProcess("foo")
 468         # Wait 1s for it to exit it will be scheduled to restart 5s later
 469         self.reactor.advance(1)
 470         # Meanwhile stop the service
 471         self.pm.stopService()
 472         # Advance to beyond the process restart time
 473         self.reactor.advance(6)
 474         # The process shouldn't have restarted because stopService has cancelled
 475         # all pending process restarts.
 476         self.assertEqual(self.pm.protocols, {})
 477