r46876 - re: #4735 review feedback

Author: glyph
Date: Mon Feb 29 15:09:00 2016
New Revision: 46876

Modified:
   branches/persistent-client-service-4735-5/twisted/application/internet.py
   branches/persistent-client-service-4735-5/twisted/application/test/test_internet.py

Log:
re: #4735 review feedback

it turns out that not using SynchronousTestCase is a bug magnet: I
should not have been using `assertFailure`, since I always forget you
have to ''return'' those.  Switching to `failureResultOf` demonstrated a
couple of implementation bugs which are now fixed.

Modified: branches/persistent-client-service-4735-5/twisted/application/internet.py
==============================================================================
--- branches/persistent-client-service-4735-5/twisted/application/internet.py	(original)
+++ branches/persistent-client-service-4735-5/twisted/application/internet.py	Mon Feb 29 15:09:00 2016
@@ -41,17 +41,15 @@
 
 from random import random as _goodEnoughRandom
 
-from zope.interface import directlyProvides
-
 from twisted.python import log
 from twisted.logger import Logger
 
 from twisted.application import service
 from twisted.internet import task
+from twisted.python.failure import Failure
 from twisted.internet.defer import (
-    CancelledError, gatherResults, Deferred, succeed
+    CancelledError, gatherResults, Deferred, succeed, fail
 )
-from twisted.internet import interfaces
 
 
 
@@ -566,6 +564,7 @@
 
         self._endpoint = endpoint
         self._failedAttempts = 0
+        self._stopped = False
         self._factory = factory
         self._timeoutForAttempt = retryPolicy
         self._clock = clock
@@ -590,12 +589,26 @@
         """
         if self._currentConnection is not None:
             return succeed(self._currentConnection)
+        elif self._stopped:
+            return fail(CancelledError())
         else:
             result = Deferred()
             self._awaitingConnected.append(result)
             return result
 
 
+    def _unawait(self, value):
+        """
+        Fire all outstanding L{ClientService.whenConnected} L{Deferred}s.
+
+        @param value: the value to fire the L{Deferred}s with.
+        """
+        self._awaitingConnected, waiting = [], self._awaitingConnected
+        for w in waiting:
+            w.callback(value)
+
+
+
     def startService(self):
         """
         Start this L{ClientService}, initiating the connection retry loop.
@@ -611,9 +624,7 @@
             self._loseConnection = protocol.transport.loseConnection
             self._lostDeferred = Deferred()
             self._currentConnection = protocol._protocol
-            self._awaitingConnected, waiting = [], self._awaitingConnected
-            for w in waiting:
-                w.callback(self._currentConnection)
+            self._unawait(self._currentConnection)
 
         def clientDisconnect(reason):
             self._currentConnection = None
@@ -650,10 +661,12 @@
             closed and all in-progress connection attempts halted.
         """
         super(ClientService, self).stopService()
+        self._stopped = True
         self._stopRetry()
         self._stopRetry = _noop
         self._connectionInProgress.cancel()
         self._loseConnection()
+        self._unawait(Failure(CancelledError()))
         return gatherResults([self._connectionInProgress, self._lostDeferred])
 
 

Modified: branches/persistent-client-service-4735-5/twisted/application/test/test_internet.py
==============================================================================
--- branches/persistent-client-service-4735-5/twisted/application/test/test_internet.py	(original)
+++ branches/persistent-client-service-4735-5/twisted/application/test/test_internet.py	Mon Feb 29 15:09:00 2016
@@ -17,7 +17,7 @@
 
 from twisted.internet.protocol import Factory, Protocol
 from twisted.internet.task import Clock
-from twisted.trial.unittest import TestCase
+from twisted.trial.unittest import TestCase, SynchronousTestCase
 from twisted.application import internet
 from twisted.application.internet import (
     StreamServerEndpointService, TimerService, ClientService)
@@ -492,7 +492,7 @@
 
 AT_LEAST_ONE_ATTEMPT = 100.
 
-class ClientServiceTests(TestCase):
+class ClientServiceTests(SynchronousTestCase):
     """
     Tests for L{ClientService}.
     """
@@ -632,7 +632,11 @@
                                            clock=clock)
         self.assertEqual(len(cq.connectQueue), 1)
         cq.connectQueue[0].errback(Failure(Exception()))
-        self.assertNoResult(service.whenConnected())
+        whenConnected = service.whenConnected()
+        self.assertNoResult(whenConnected)
+        # Don't fail during test tear-down when service shutdown causes all
+        # waiting connections to fail.
+        whenConnected.addErrback(lambda ignored: ignored.trap(CancelledError))
         clock.advance(AT_LEAST_ONE_ATTEMPT)
         self.assertEqual(len(cq.connectQueue), 2)
 
@@ -663,10 +667,11 @@
         protocol stopping deferred is called and the reference to the protocol
         is removed.
         """
-        cq, service = self.makeReconnector()
+        clock = Clock()
+        cq, service = self.makeReconnector(clock=clock)
         d = service.stopService()
         cq.constructedProtocols[0].connectionLost(Failure(IndentationError()))
-        self.assertFailure(service.whenConnected(), CancelledError)
+        self.failureResultOf(service.whenConnected(), CancelledError)
         self.assertTrue(d.called)
 
 
@@ -717,6 +722,7 @@
         self.assertNoResult(a)
         self.assertNoResult(b)
         service.stopService()
-        self.assertFailure(a, CancelledError)
-        self.assertFailure(b, CancelledError)
+        clock.advance(AT_LEAST_ONE_ATTEMPT)
+        self.failureResultOf(a, CancelledError)
+        self.failureResultOf(b, CancelledError)