twisted/application/test/test_internet.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 """
   5 Tests for (new code in) L{twisted.application.internet}.
   6 """
   7
   8
   9 from zope.interface import implements
  10 from zope.interface.verify import verifyClass
  11
  12 from twisted.internet.protocol import Factory
  13 from twisted.trial.unittest import TestCase
  14 from twisted.application.internet import StreamServerEndpointService
  15 from twisted.internet.interfaces import IStreamServerEndpoint, IListeningPort
  16 from twisted.internet.defer import Deferred, CancelledError
  17
  18 class FakeServer(object):
  19     """
  20     In-memory implementation of L{IStreamServerEndpoint}.
  21
  22     @ivar result: The L{Deferred} resulting from the call to C{listen}, after
  23         C{listen} has been called.
  24
  25     @ivar factory: The factory passed to C{listen}.
  26
  27     @ivar cancelException: The exception to errback C{self.result} when it is
  28         cancelled.
  29
  30     @ivar port: The L{IListeningPort} which C{listen}'s L{Deferred} will fire
  31         with.
  32
  33     @ivar listenAttempts: The number of times C{listen} has been invoked.
  34
  35     @ivar failImmediately: If set, the exception to fail the L{Deferred}
  36         returned from C{listen} before it is returned.
  37     """
  38
  39     implements(IStreamServerEndpoint)
  40
  41     result = None
  42     factory = None
  43     failImmediately = None
  44     cancelException = CancelledError()
  45     listenAttempts = 0
  46
  47     def __init__(self):
  48         self.port = FakePort()
  49
  50
  51     def listen(self, factory):
  52         """
  53         Return a Deferred and store it for future use.  (Implementation of
  54         L{IStreamServerEndpoint}).
  55         """
  56         self.listenAttempts += 1
  57         self.factory = factory
  58         self.result = Deferred(
  59             canceller=lambda d: d.errback(self.cancelException))
  60         if self.failImmediately is not None:
  61             self.result.errback(self.failImmediately)
  62         return self.result
  63
  64
  65     def startedListening(self):
  66         """
  67         Test code should invoke this method after causing C{listen} to be
  68         invoked in order to fire the L{Deferred} previously returned from
  69         C{listen}.
  70         """
  71         self.result.callback(self.port)
  72
  73
  74     def stoppedListening(self):
  75         """
  76         Test code should invoke this method after causing C{stopListening} to
  77         be invoked on the port fired from the L{Deferred} returned from
  78         C{listen} in order to cause the L{Deferred} returned from
  79         C{stopListening} to fire.
  80         """
  81         self.port.deferred.callback(None)
  82
  83 verifyClass(IStreamServerEndpoint, FakeServer)
  84
  85
  86
  87 class FakePort(object):
  88     """
  89     Fake L{IListeningPort} implementation.
  90
  91     @ivar deferred: The L{Deferred} returned by C{stopListening}.
  92     """
  93
  94     implements(IListeningPort)
  95
  96     deferred = None
  97
  98     def stopListening(self):
  99         self.deferred = Deferred()
 100         return self.deferred
 101
 102 verifyClass(IStreamServerEndpoint, FakeServer)
 103
 104
 105
 106 class TestEndpointService(TestCase):
 107     """
 108     Tests for L{twisted.application.internet}.
 109     """
 110
 111     def setUp(self):
 112         """
 113         Construct a stub server, a stub factory, and a
 114         L{StreamServerEndpointService} to test.
 115         """
 116         self.fakeServer = FakeServer()
 117         self.factory = Factory()
 118         self.svc = StreamServerEndpointService(self.fakeServer, self.factory)
 119
 120
 121     def test_privilegedStartService(self):
 122         """
 123         L{StreamServerEndpointService.privilegedStartService} calls its
 124         endpoint's C{listen} method with its factory.
 125         """
 126         self.svc.privilegedStartService()
 127         self.assertIdentical(self.factory, self.fakeServer.factory)
 128
 129
 130     def test_synchronousRaiseRaisesSynchronously(self, thunk=None):
 131         """
 132         L{StreamServerEndpointService.startService} should raise synchronously
 133         if the L{Deferred} returned by its wrapped
 134         L{IStreamServerEndpoint.listen} has already fired with an errback and
 135         the L{StreamServerEndpointService}'s C{_raiseSynchronously} flag has
 136         been set.  This feature is necessary to preserve compatibility with old
 137         behavior of L{twisted.internet.strports.service}, which is to return a
 138         service which synchronously raises an exception from C{startService}
 139         (so that, among other things, twistd will not start running).  However,
 140         since L{IStreamServerEndpoint.listen} may fail asynchronously, it is
 141         a bad idea to rely on this behavior.
 142         """
 143         self.fakeServer.failImmediately = ZeroDivisionError()
 144         self.svc._raiseSynchronously = True
 145         self.assertRaises(ZeroDivisionError, thunk or self.svc.startService)
 146
 147
 148     def test_synchronousRaisePrivileged(self):
 149         """
 150         L{StreamServerEndpointService.privilegedStartService} should behave the
 151         same as C{startService} with respect to
 152         L{TestEndpointService.test_synchronousRaiseRaisesSynchronously}.
 153         """
 154         self.test_synchronousRaiseRaisesSynchronously(
 155             self.svc.privilegedStartService)
 156
 157
 158     def test_failReportsError(self):
 159         """
 160         L{StreamServerEndpointService.startService} and
 161         L{StreamServerEndpointService.privilegedStartService} should both log
 162         an exception when the L{Deferred} returned from their wrapped
 163         L{IStreamServerEndpoint.listen} fails.
 164         """
 165         self.svc.startService()
 166         self.fakeServer.result.errback(ZeroDivisionError())
 167         logged = self.flushLoggedErrors(ZeroDivisionError)
 168         self.assertEqual(len(logged), 1)
 169
 170
 171     def test_synchronousFailReportsError(self):
 172         """
 173         Without the C{_raiseSynchronously} compatibility flag, failing
 174         immediately has the same behavior as failing later; it logs the error.
 175         """
 176         self.fakeServer.failImmediately = ZeroDivisionError()
 177         self.svc.startService()
 178         logged = self.flushLoggedErrors(ZeroDivisionError)
 179         self.assertEqual(len(logged), 1)
 180
 181
 182     def test_startServiceUnstarted(self):
 183         """
 184         L{StreamServerEndpointService.startService} sets the C{running} flag,
 185         and calls its endpoint's C{listen} method with its factory, if it
 186         has not yet been started.
 187         """
 188         self.svc.startService()
 189         self.assertIdentical(self.factory, self.fakeServer.factory)
 190         self.assertEqual(self.svc.running, True)
 191
 192
 193     def test_startServiceStarted(self):
 194         """
 195         L{StreamServerEndpointService.startService} sets the C{running} flag,
 196         but nothing else, if the service has already been started.
 197         """
 198         self.test_privilegedStartService()
 199         self.svc.startService()
 200         self.assertEqual(self.fakeServer.listenAttempts, 1)
 201         self.assertEqual(self.svc.running, True)
 202
 203
 204     def test_stopService(self):
 205         """
 206         L{StreamServerEndpointService.stopService} calls C{stopListening} on
 207         the L{IListeningPort} returned from its endpoint, returns the
 208         C{Deferred} from stopService, and sets C{running} to C{False}.
 209         """
 210         self.svc.privilegedStartService()
 211         self.fakeServer.startedListening()
 212         # Ensure running gets set to true
 213         self.svc.startService()
 214         result = self.svc.stopService()
 215         l = []
 216         result.addCallback(l.append)
 217         self.assertEqual(len(l), 0)
 218         self.fakeServer.stoppedListening()
 219         self.assertEqual(len(l), 1)
 220         self.assertFalse(self.svc.running)
 221
 222
 223     def test_stopServiceBeforeStartFinished(self):
 224         """
 225         L{StreamServerEndpointService.stopService} cancels the L{Deferred}
 226         returned by C{listen} if it has not yet fired.  No error will be logged
 227         about the cancellation of the listen attempt.
 228         """
 229         self.svc.privilegedStartService()
 230         result = self.svc.stopService()
 231         l = []
 232         result.addBoth(l.append)
 233         self.assertEqual(l, [None])
 234         self.assertEqual(self.flushLoggedErrors(CancelledError), [])
 235
 236
 237     def test_stopServiceCancelStartError(self):
 238         """
 239         L{StreamServerEndpointService.stopService} cancels the L{Deferred}
 240         returned by C{listen} if it has not fired yet.  An error will be logged
 241         if the resulting exception is not L{CancelledError}.
 242         """
 243         self.fakeServer.cancelException = ZeroDivisionError()
 244         self.svc.privilegedStartService()
 245         result = self.svc.stopService()
 246         l = []
 247         result.addCallback(l.append)
 248         self.assertEqual(l, [None])
 249         stoppingErrors = self.flushLoggedErrors(ZeroDivisionError)
 250         self.assertEqual(len(stoppingErrors), 1)
 251
 252