Repository: twisted/txacme

Issue:
stopService() leaves pooled connections open
With the default client, a persistent `HTTPConnectionPool` is created for connections to the ACME directory. This pool isn't tracked, however, and when `stopService()` is called on the txacme service a pooled connection remains waiting on the reactor. This breaks my integration tests.

Discussion:

Patch:
diff --git a/.travis.yml b/.travis.yml
index 894777e..2e063c5 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -7,54 +7,33 @@ stages:
     if: tag IS present
 matrix:
   include:
-  - python: 2.7
-    env: TOXENV=py27-twlatest-alldeps
-  - python: 3.5
-    env: TOXENV=py35-twlatest-alldeps
+  - python: 3.6
+    env: TOXENV=py36-twlatest
   - python: 3.6
     env: TOXENV=py36-twlatest-alldeps
   - python: 3.7
     env: TOXENV=py37-twlatest-alldeps
     dist: xenial
-  - python: pypy2.7-7.2.0
-    dist: bionic
-    env: TOXENV=pypy-twlatest-alldeps
   - python: pypy3.6-7.2.0
     dist: bionic
     env: TOXENV=pypy3-twlatest-alldeps
-  - python: 2.7
-    env: TOXENV=py27-twtrunk-acmaster-alldeps
-  - python: 3.5
-    env: TOXENV=py35-twtrunk-acmaster-alldeps
   - python: 3.6
     env: TOXENV=py36-twtrunk-acmaster-alldeps
   - python: 3.7
     env: TOXENV=py37-twtrunk-acmaster-alldeps
     dist: xenial
-  - python: pypy2.7-7.2.0
-    dist: bionic
-    env: TOXENV=pypy-twtrunk-acmaster-alldeps
   - python: pypy3.6-7.2.0
     dist: bionic
     env: TOXENV=pypy3-twtrunk-acmaster-alldeps
-  - python: 2.7
-    env: TOXENV=py27-twlowest-alldeps
-  - python: 3.5
-    env: TOXENV=py35-twlowest-alldeps
   - python: 3.6
     env: TOXENV=py36-twlowest-alldeps
   - python: 3.7
     env: TOXENV=py37-twlowest-alldeps
     dist: xenial
-  - python: pypy2.7-7.2.0
-    dist: bionic
-    env: TOXENV=pypy-twlowest-alldeps
   - python: pypy3.6-7.2.0
     dist: bionic
     env: TOXENV=pypy3-twlowest-alldeps
-  - python: 2.7
-    env: TOXENV=py27-twlatest
-  - python: 2.7
+  - python: 3.7
     env: TOXENV=docs
     addons:
       apt:
@@ -63,11 +42,8 @@ matrix:
   - python: 3.6
     env: TOXENV=flake8
   allow_failures:
-  - env: TOXENV=py27-twtrunk-acmaster-alldeps
-  - env: TOXENV=py35-twtrunk-acmaster-alldeps
   - env: TOXENV=py36-twtrunk-acmaster-alldeps
   - env: TOXENV=py37-twtrunk-acmaster-alldeps
-  - env: TOXENV=pypy-twtrunk-acmaster-alldeps
   - env: TOXENV=pypy3-twtrunk-acmaster-alldeps
 install:
   # Upgrade packaging tools separately, so that other installations are
diff --git a/README.rst b/README.rst
index 1c3cf44..6c7585e 100644
--- a/README.rst
+++ b/README.rst
@@ -30,7 +30,7 @@ interested in implementing or have need of the server side, please get in
 touch!
 
 ``txacme``\ ’s documentation lives at `Read the Docs`_, the code on `GitHub`_.
-It’s rigorously tested on Python 2.7, 3.4+, and PyPy.
+It’s rigorously tested on Python 3.6+, and PyPy3.
 
 .. _ACME: https://github.com/ietf-wg-acme/acme/blob/master/draft-ietf-acme-acme.md
 
diff --git a/docs/conf.py b/docs/conf.py
index fe8a1bd..3ad8af5 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -364,9 +364,9 @@ def run_apidoc(_):
         cur_dir = os.path.abspath(os.path.dirname(__file__))
         output_path = os.path.join(cur_dir, 'api')
         cmd_path = 'sphinx-apidoc'
-        if hasattr(sys, 'real_prefix'):  # Check to see if we are in a virtualenv
-            # If we are, assemble the path manually
-            cmd_path = os.path.abspath(os.path.join(sys.prefix, 'bin', 'sphinx-apidoc'))
+        cmd_path = os.path.abspath(
+            os.path.join(sys.prefix, 'bin', 'sphinx-apidoc')
+        )
         subprocess.check_call(
             [cmd_path, '-e', '-o', output_path, module, '--force'],
             env={'SPHINX_APIDOC_OPTIONS': 'members'})
diff --git a/src/txacme/client.py b/src/txacme/client.py
index 1ecd10b..d0a2795 100644
--- a/src/txacme/client.py
+++ b/src/txacme/client.py
@@ -79,7 +79,7 @@ def _default_client(jws_client, reactor, key, alg):
     if jws_client is None:
         pool = HTTPConnectionPool(reactor)
         agent = Agent(reactor, pool=pool)
-        jws_client = JWSClient(HTTPClient(agent=agent), key, alg)
+        jws_client = JWSClient(agent, key, alg)
     return jws_client
 
 
@@ -116,6 +116,8 @@ class Client(object):
         """
         Construct a client from an ACME directory at a given URL.
 
+        At construct time, it validates the ACME directory.
+
         :param url: The ``twisted.python.url.URL`` to fetch the directory from.
             See `txacme.urls` for constants for various well-known public
             directories.
@@ -146,6 +148,16 @@ class Client(object):
                 .addCallback(cls, reactor, key, jws_client)
                 .addActionFinish())
 
+    def stop(self):
+        """
+        Stops the client operation.
+
+        This cancels pending operations and does cleanup.
+
+        :return: A deferred which files when the client is stopped.
+        """
+        return self._client.stop()
+
     def register(self, new_reg=None):
         """
         Create a new registration with the ACME server.
@@ -675,9 +687,11 @@ class JWSClient(object):
     """
     timeout = _DEFAULT_TIMEOUT
 
-    def __init__(self, treq_client, key, alg,
+    def __init__(self, agent, key, alg,
                  user_agent=u'txacme/{}'.format(__version__).encode('ascii')):
-        self._treq = treq_client
+        self._treq = HTTPClient(agent=agent)
+        self._agent = agent
+        self._current_request = None
         self._key = key
         self._alg = alg
         self._user_agent = user_agent
@@ -765,14 +779,23 @@ class JWSClient(object):
 
         :return: Deferred firing with the HTTP response.
         """
+        def cb_request_done(result):
+            """
+            Called when we got a response from the request.
+            """
+            self._current_request = None
+            return result
+
         action = LOG_JWS_REQUEST(url=url)
         with action.context():
             headers = kwargs.setdefault('headers', Headers())
             headers.setRawHeaders(b'user-agent', [self._user_agent])
             kwargs.setdefault('timeout', self.timeout)
+            self._current_request = self._treq.request(
+                method, url, *args, **kwargs)
             return (
-                DeferredContext(
-                    self._treq.request(method, url, *args, **kwargs))
+                DeferredContext(self._current_request)
+                .addCallback(cb_request_done)
                 .addCallback(
                     tap(lambda r: action.add_success_fields(
                         code=r.code,
@@ -780,6 +803,24 @@ class JWSClient(object):
                             b'content-type', [None])[0])))
                 .addActionFinish())
 
+    def stop(self):
+        """
+        Stops the operation.
+
+        This cancels pending operations and does cleanup.
+
+        :return: A deferred which fires when the client is stopped.
+        """
+        if self._current_request is not None:
+            self._current_request.addErrback(lambda _: None)
+            self._current_request.cancel()
+            self._current_request = None
+
+        agent_pool = getattr(self._agent, '_pool', None)
+        if agent_pool:
+            return agent_pool.closeCachedConnections()
+        return succeed(None)
+
     def head(self, url, *args, **kwargs):
         """
         Send HEAD request without checking the response.
diff --git a/src/txacme/endpoint.py b/src/txacme/endpoint.py
index 5c65960..9696773 100644
--- a/src/txacme/endpoint.py
+++ b/src/txacme/endpoint.py
@@ -52,11 +52,8 @@ class AutoTLSEndpoint(object):
     :param directory: ``twisted.python.url.URL`` for the ACME directory to use
         for issuing certs.
 
-    :type client_creator: Callable[[reactor, ``twisted.python.url.URL``],
-        Deferred[`txacme.client.Client`]]
-    :param client_creator: A callable called with the reactor and directory URL
-        for creating the ACME client.  For example, ``partial(Client.from_url,
-        key=acme_key, alg=RS256)``.
+    :type client: txacme.client.Client
+    :param client: An ACME client used to interact with the server.
     :type cert_store: `txacme.interfaces.ICertificateStore`
     :param cert_store: The certificate
         store containing the certificates to manage.  For example,
@@ -84,7 +81,7 @@ class AutoTLSEndpoint(object):
     reactor = attr.ib()
     directory = attr.ib(
         validator=lambda inst, a, value: check_directory_url_type(value))
-    client_creator = attr.ib()
+    client = attr.ib()
     cert_store = attr.ib()
     cert_mapping = attr.ib()
     sub_endpoint = attr.ib()
@@ -101,8 +98,7 @@ class AutoTLSEndpoint(object):
         def _got_port(port):
             self.service = AcmeIssuingService(
                 cert_store=self.cert_store,
-                client_creator=partial(
-                    self.client_creator, self.reactor, self.directory),
+                client=self.client,
                 clock=self.reactor,
                 responders=[responder],
                 check_interval=self.check_interval,
@@ -174,8 +170,8 @@ def _parse(reactor, directory, pemdir, *args, **kwargs):
     return AutoTLSEndpoint(
         reactor=reactor,
         directory=directory,
-        client_creator=partial(
-            Client.from_url, key=acme_key, alg=RS256, timeout=timeout),
+        client=Client.from_url(
+            reactor, directory, key=acme_key, alg=RS256, timeout=timeout),
         cert_store=DirectoryStore(pem_path),
         cert_mapping=HostDirectoryMap(pem_path),
         sub_endpoint=serverFromString(reactor, sub))
diff --git a/src/txacme/newsfragments/86.bugfix b/src/txacme/newsfragments/86.bugfix
new file mode 100644
index 0000000..3e88c00
--- /dev/null
+++ b/src/txacme/newsfragments/86.bugfix
@@ -0,0 +1,4 @@
+INCOMPATIBLE CHANGE: txacme.service.AcmeIssuingService.stopFactory now
+closes the persisted HTTP client connections.
+This is done to bring the in a state similar to the one before calling
+startFactory.
diff --git a/src/txacme/newsfragments/86.removal b/src/txacme/newsfragments/86.removal
new file mode 100644
index 0000000..dcdaa2f
--- /dev/null
+++ b/src/txacme/newsfragments/86.removal
@@ -0,0 +1,4 @@
+INCOMPATIBLE CHANGE: txacme.client.JWSClient is now initialized with an
+twisted.web.client.Agent instead of treq.client.HTTPClient.
+In this way the usage of Treq is internal to txacme.
+It was changed to make it easier to close the idle persistent connection.
diff --git a/src/txacme/service.py b/src/txacme/service.py
index 0c228bb..af9ce43 100644
--- a/src/txacme/service.py
+++ b/src/txacme/service.py
@@ -35,10 +35,13 @@ class AcmeIssuingService(Service):
     :param cert_store: The certificate store containing the certificates to
         manage.
 
-    :type client_creator: Callable[[], Deferred[`txacme.client.Client`]]
-    :param client_creator: A callable called with no arguments for creating the
-        ACME client.  For example, ``partial(Client.from_url, reactor=reactor,
+    :type client: `txacme.client.Client`
+    :param client: A client which is already set to be used for an
+        environment.  For example, ``Client.from_url(reactor=reactor,
         url=LETSENCRYPT_STAGING_DIRECTORY, key=acme_key, alg=RS256)``.
+        When the service is stopped, it will automatically call the stop
+        method on the client.
+
     :param clock: ``IReactorTime`` provider; usually the reactor, when not
         testing.
 
@@ -65,7 +68,7 @@ class AcmeIssuingService(Service):
         key generation requirements.
     """
     cert_store = attr.ib()
-    _client_creator = attr.ib()
+    _client = attr.ib()
     _clock = attr.ib()
     _responders = attr.ib()
     _email = attr.ib(default=None)
@@ -115,7 +118,7 @@ class AcmeIssuingService(Service):
 
             d1 = (
                 gatherResults(
-                    [self._with_client(self._issue_cert, server_name)
+                    [self._issue_cert(server_name)
                      .addErrback(self._panic, server_name)
                      for server_name in panicing],
                     consumeErrors=True)
@@ -169,7 +172,7 @@ class AcmeIssuingService(Service):
             d_issue, waiting = self._issuing[server_name]
             waiting.append(d)
         else:
-            d_issue = self._with_client(self._issue_cert, server_name)
+            d_issue = self._issue_cert(server_name)
             waiting = [d]
             self._issuing[server_name] = (d_issue, waiting)
             # Add the callback afterwards in case we're using a client
@@ -177,13 +180,7 @@ class AcmeIssuingService(Service):
             d_issue.addBoth(finish)
         return d
 
-    def _with_client(self, f, *a, **kw):
-        """
-        Construct a client, and perform an operation with it.
-        """
-        return self._client_creator().addCallback(f, *a, **kw)
-
-    def _issue_cert(self, client, server_name):
+    def _issue_cert(self, server_name):
         """
         Issue a new cert for a particular name.
         """
@@ -200,10 +197,10 @@ class AcmeIssuingService(Service):
         def answer_and_poll(authzr):
             def got_challenge(stop_responding):
                 return (
-                    poll_until_valid(authzr, self._clock, client)
+                    poll_until_valid(authzr, self._clock, self._client)
                     .addBoth(tap(lambda _: stop_responding())))
             return (
-                answer_challenge(authzr, client, self._responders)
+                answer_challenge(authzr, self._client, self._responders)
                 .addCallback(got_challenge))
 
         def got_cert(certr):
@@ -223,13 +220,13 @@ class AcmeIssuingService(Service):
             return objects
 
         return (
-            client.request_challenges(fqdn_identifier(server_name))
+            self._client.request_challenges(fqdn_identifier(server_name))
             .addCallback(answer_and_poll)
-            .addCallback(lambda ign: client.request_issuance(
+            .addCallback(lambda ign: self._client.request_issuance(
                 CertificateRequest(
                     csr=csr_for_names([server_name], key))))
             .addCallback(got_cert)
-            .addCallback(client.fetch_chain)
+            .addCallback(self._client.fetch_chain)
             .addCallback(got_chain)
             .addCallback(partial(self.cert_store.store, server_name)))
 
@@ -240,9 +237,9 @@ class AcmeIssuingService(Service):
         if self._registered:
             return succeed(None)
         else:
-            return self._with_client(self._register)
+            return self._register()
 
-    def _register(self, client):
+    def _register(self):
         """
         Register and agree to the TOS.
         """
@@ -251,8 +248,8 @@ class AcmeIssuingService(Service):
             self._registered = True
         regr = messages.NewRegistration.from_data(email=self._email)
         return (
-            client.register(regr)
-            .addCallback(client.agree_to_tos)
+            self._client.register(regr)
+            .addCallback(self._client.agree_to_tos)
             .addCallback(_registered))
 
     def when_certs_valid(self):
@@ -291,7 +288,10 @@ class AcmeIssuingService(Service):
         for d in list(self._waiting):
             d.cancel()
         self._waiting = []
-        return self._timer_service.stopService()
+
+        deferred = self._client.stop()
+        deferred.addCallback(lambda _: self._timer_service.stopService())
+        return deferred
 
 
 __all__ = ['AcmeIssuingService']
diff --git a/tox.ini b/tox.ini
index 62de55b..d18ceb0 100644
--- a/tox.ini
+++ b/tox.ini
@@ -1,5 +1,5 @@
 [tox]
-envlist = coverage-clean,{py27,pypy,py35,py36,py37,pypy3}-{twlatest,twtrunk,twlowest}-{aclatest,acmaster}-alldeps,py27-twlatest,flake8,docs,coverage-report
+envlist = coverage-clean,{py36,py37,pypy3}-{twlatest,twtrunk,twlowest}-{aclatest,acmaster}-alldeps,flake8,docs,coverage-report
 
 [testenv]
 setenv =


Test patch:
diff --git a/src/integration/test_client.py b/src/integration/test_client.py
index 49260fe..92abf25 100644
--- a/src/integration/test_client.py
+++ b/src/integration/test_client.py
@@ -14,7 +14,6 @@ from eliot.twisted import DeferredContext
 from twisted.internet import reactor
 from twisted.internet.defer import succeed
 from twisted.internet.endpoints import serverFromString
-from twisted.python.compat import _PY3
 from twisted.python.filepath import FilePath
 from twisted.trial.unittest import TestCase
 from twisted.web.resource import Resource
@@ -162,13 +161,9 @@ def _getenv(name, default=None):
     """
     Sigh.
     """
-    if not _PY3:
-        name = name.encode('utf-8')
     value = getenv(name)
     if value is None:
         return default
-    if not _PY3:
-        value = value.decode('utf-8')
     return value
 
 
@@ -184,16 +179,9 @@ class LetsEncryptStagingTLSSNI01Tests(ClientTestsMixin, TestCase):
     ENDPOINT = _getenv(u'ACME_ENDPOINT')
     if None in [HOST, ENDPOINT]:
         skip = 'Must provide $ACME_HOST and $ACME_ENDPOINT'
-    elif not _PY3:
-        ENDPOINT = ENDPOINT.encode('utf-8')
 
     def _create_client(self, key):
-        return (
-            Client.from_url(reactor, LETSENCRYPT_STAGING_DIRECTORY, key=key)
-            .addCallback(tap(
-                lambda client: self.addCleanup(
-                    client._client._treq._agent._pool.closeCachedConnections)))
-            )
+        return Client.from_url(reactor, LETSENCRYPT_STAGING_DIRECTORY, key=key)
 
     def _create_responder(self):
         action = start_action(action_type=u'integration:create_responder')
@@ -231,12 +219,7 @@ class LetsEncryptStagingLibcloudTests(ClientTestsMixin, TestCase):
         skip = 'Must provide $ACME_HOST and $LIBCLOUD_*'
 
     def _create_client(self, key):
-        return (
-            Client.from_url(reactor, LETSENCRYPT_STAGING_DIRECTORY, key=key)
-            .addCallback(tap(
-                lambda client: self.addCleanup(
-                    client._client._treq._agent._pool.closeCachedConnections)))
-            )
+        return Client.from_url(reactor, LETSENCRYPT_STAGING_DIRECTORY, key=key)
 
     def _create_responder(self):
         with start_action(action_type=u'integration:create_responder'):
diff --git a/src/txacme/test/strategies.py b/src/txacme/test/strategies.py
index e04b62f..f8613f1 100644
--- a/src/txacme/test/strategies.py
+++ b/src/txacme/test/strategies.py
@@ -1,10 +1,7 @@
 """
 Miscellaneous strategies for Hypothesis testing.
 """
-try:
-    from base64 import encodebytes
-except ImportError:
-    from base64 import encodestring as encodebytes
+from base64 import encodebytes
 
 from hypothesis import strategies as s
 from pem import Certificate, RSAPrivateKey
diff --git a/src/txacme/test/test_client.py b/src/txacme/test/test_client.py
index 294dd95..976f1de 100644
--- a/src/txacme/test/test_client.py
+++ b/src/txacme/test/test_client.py
@@ -20,19 +20,20 @@ from testtools import ExpectedException, TestCase
 from testtools.matchers import (
     AfterPreprocessing, Always, ContainsDict, Equals, Is, IsInstance,
     MatchesAll, MatchesListwise, MatchesPredicate, MatchesStructure, Mismatch,
-    Never, Not, StartsWith)
-from testtools.twistedsupport import failed, succeeded
+    Never, Not, StartsWith, HasLength)
+from testtools.twistedsupport import failed, succeeded, has_no_result
 from treq.client import HTTPClient
 from treq.testing import RequestSequence as treq_RequestSequence
 from treq.testing import (
     _SynchronousProducer, RequestTraversalAgent, StringStubbingResource)
 from twisted.internet import reactor
-from twisted.internet.defer import CancelledError, fail, succeed
+from twisted.internet.defer import Deferred, CancelledError, fail, succeed
+from twisted.internet.error import ConnectionClosed
 from twisted.internet.task import Clock
-from twisted.python.compat import _PY3
 from twisted.python.url import URL
 from twisted.test.proto_helpers import MemoryReactor
-from twisted.web import http
+from twisted.web import http, server
+from twisted.web.resource import Resource
 from twisted.web.http_headers import Headers
 from zope.interface import implementer
 
@@ -104,13 +105,45 @@ class Nearly(object):
                     value, self.epsilon, self.expected))
 
 
+@attr.s(auto_attribs=True)
+class ConnectionPoolFixture:
+    _deferred: Deferred = attr.Factory(Deferred)
+    _closing: bool = False
+
+    def started_closing(self):
+        return self._closing
+
+    def finish_closing(self):
+        self._deferred.callback(None)
+
+
+@attr.s(auto_attribs=True)
+class FakePool:
+    _fixture_pool: ConnectionPoolFixture
+
+    def closeCachedConnections(self):  # noqa
+        self._fixture_pool._closing = True
+        return self._fixture_pool._deferred
+
+
 class ClientFixture(Fixture):
     """
     Create a :class:`~txacme.client.Client` for testing.
     """
-    def __init__(self, sequence, key=None, alg=RS256):
+    def __init__(
+        self, sequence, key=None, alg=RS256, use_connection_pool=False
+    ):
         super(ClientFixture, self).__init__()
         self._sequence = sequence
+        if isinstance(sequence, treq_RequestSequence):
+            self._agent = RequestTraversalAgent(
+                StringStubbingResource(self._sequence)
+            )
+        else:
+            self._agent = RequestTraversalAgent(sequence)
+        if use_connection_pool:
+            self.pool = ConnectionPoolFixture()
+            self._agent._pool = FakePool(self.pool)
         self._directory = messages.Directory({
             messages.NewRegistration:
             u'https://example.org/acme/new-reg',
@@ -127,14 +160,15 @@ class ClientFixture(Fixture):
         self._alg = alg
 
     def _setUp(self):  # noqa
-        treq_client = HTTPClient(
-            agent=RequestTraversalAgent(
-                StringStubbingResource(self._sequence)),
-            data_to_body_producer=_SynchronousProducer)
+        jws_client = JWSClient(self._agent, self._key, self._alg)
+        jws_client._treq._data_to_body_producer = _SynchronousProducer
         self.clock = Clock()
         self.client = Client(
             self._directory, self.clock, self._key,
-            jws_client=JWSClient(treq_client, self._key, self._alg))
+            jws_client=jws_client)
+
+    def flush(self):
+        self._agent.flush()
 
 
 def _nonce_response(url, nonce):
@@ -162,10 +196,7 @@ def _nonce_response(url, nonce):
 
 
 def _json_dumps(j):
-    s = json.dumps(j)
-    if _PY3:
-        s = s.encode('utf-8')
-    return s
+    return json.dumps(j).encode("utf-8")
 
 
 class RequestSequence(treq_RequestSequence):
@@ -344,6 +375,57 @@ class ClientTests(TestCase):
             self.assertThat(
                 client.register(reg2),
                 failed_with(IsInstance(errors.UnexpectedUpdate)))
+        self.expectThat(client.stop(), succeeded(Equals(None)))
+
+    def stop_in_progress(self, use_pool=False):
+        requested = []
+
+        class NoAnswerResource(Resource):
+            isLeaf = True       # noqa
+
+            def render(self, request):
+                requested.append(request.notifyFinish())
+                return server.NOT_DONE_YET
+
+        self.client_fixture = self.useFixture(
+            ClientFixture(
+                NoAnswerResource(),
+                key=RSA_KEY_512,
+                use_connection_pool=use_pool,
+            )
+        )
+        client = self.client_fixture.client
+        reg = messages.NewRegistration.from_data(email=u'example@example.com')
+        register_call = client.register(reg)
+        self.expectThat(requested, HasLength(1))
+        self.expectThat(register_call, has_no_result())
+        self.expectThat(requested[0], has_no_result())
+        stop_deferred = client.stop()
+        self.assertThat(register_call, succeeded(Equals(None)))
+        self.client_fixture.flush()
+        self.assertThat(
+            requested[0],
+            failed_with(IsInstance(ConnectionClosed)),
+        )
+        return stop_deferred
+
+    def test_stop_in_progress(self):
+        """
+        If we stop the client while an operation is in progress, it's
+        cancelled.
+        """
+        self.assertThat(self.stop_in_progress(), succeeded(Equals(None)))
+
+    def test_stop_in_progress_with_pool(self):
+        """
+        If we stop the client while an operation is in progress, it will stop.
+        """
+        stopped = self.stop_in_progress(True)
+        self.assertThat(stopped, has_no_result())
+        self.assertThat(self.client_fixture.pool.started_closing(),
+                        Equals(True))
+        self.client_fixture.pool.finish_closing()
+        self.assertThat(stopped, succeeded(Equals(None)))
 
     def test_register(self):
         """
@@ -781,16 +863,15 @@ class ClientTests(TestCase):
                   u'new-authz': u'https://example.org/acme/new-authz',
               })))],
             self.expectThat)
-        treq_client = HTTPClient(
-            agent=RequestTraversalAgent(
-                StringStubbingResource(sequence)),
-            data_to_body_producer=_SynchronousProducer)
+        agent = RequestTraversalAgent(
+            StringStubbingResource(sequence))
+        jws_client = JWSClient(agent, key=RSA_KEY_512, alg=RS256)
+        jws_client._treq._data_to_body_producer = _SynchronousProducer
         with sequence.consume(self.fail):
             d = Client.from_url(
                 reactor, URL.fromText(u'https://example.org/acme/'),
                 key=RSA_KEY_512, alg=RS256,
-                jws_client=JWSClient(
-                    treq_client, key=RSA_KEY_512, alg=RS256))
+                jws_client=jws_client)
             self.assertThat(
                 d,
                 succeeded(
diff --git a/src/txacme/test/test_endpoint.py b/src/txacme/test/test_endpoint.py
index 1de53ce..7e34c9a 100644
--- a/src/txacme/test/test_endpoint.py
+++ b/src/txacme/test/test_endpoint.py
@@ -21,6 +21,7 @@ from twisted.plugin import IPlugin
 from twisted.plugins import txacme_endpoint
 from twisted.python.filepath import FilePath
 from twisted.python.url import URL
+from twisted.test.proto_helpers import MemoryReactorClock
 from txsni.snimap import HostDirectoryMap
 from zope.interface import implementer
 from zope.interface.verify import verifyObject
@@ -29,7 +30,7 @@ from txacme._endpoint_parser import _AcmeParser
 from txacme.endpoint import AutoTLSEndpoint, load_or_create_client_key
 from txacme.store import DirectoryStore
 from txacme.test.test_client import RSA_KEY_512
-from txacme.testing import FakeClient, MemoryStore
+from txacme.testing import FakeClient, MemoryStore, TXACMETestCase
 from txacme.urls import LETSENCRYPT_DIRECTORY, LETSENCRYPT_STAGING_DIRECTORY
 
 
@@ -51,7 +52,30 @@ class DummyEndpoint(object):
         return succeed(DummyPort())
 
 
-class EndpointTests(TestCase):
+class TXAcmeTestCaseTestCase(TestCase):
+    def test_tear_down(self):
+        from twisted.internet import reactor
+
+        def no_op():
+            pass
+
+        no_op()
+
+        garbage_delayed_call = None
+
+        class TestTest(TXACMETestCase):
+            def test_test(self):
+                nonlocal garbage_delayed_call
+                garbage_delayed_call = reactor.callLater(1.0, no_op)
+
+        test_case_under_test = TestTest("test_test")
+        result = test_case_under_test.run()
+        self.assertThat(len(result.failures), Equals(1))
+        self.assertThat(garbage_delayed_call.active(), Equals(False))
+        self.assertThat(reactor.getDelayedCalls(), Equals([]))
+
+
+class EndpointTests(TXACMETestCase):
     """
     Tests for `~txacme.endpoint.AutoTLSEndpoint`.
     """
@@ -64,7 +88,7 @@ class EndpointTests(TestCase):
         self.endpoint = AutoTLSEndpoint(
             reactor=clock,
             directory=URL.fromText(u'https://example.com/'),
-            client_creator=lambda reactor, directory: succeed(client),
+            client=client,
             cert_store=MemoryStore(),
             cert_mapping={},
             sub_endpoint=DummyEndpoint())
@@ -78,7 +102,7 @@ class EndpointTests(TestCase):
             AutoTLSEndpoint(
                 reactor=Clock(),
                 directory='/wrong/kind/of/directory',
-                client_creator=None,
+                client=None,
                 cert_store=None,
                 cert_mapping={},
                 sub_endpoint=DummyEndpoint())
@@ -106,10 +130,11 @@ class EndpointTests(TestCase):
             MatchesStructure(running=Equals(False)))
 
 
-class PluginTests(TestCase):
+class PluginTests(TXACMETestCase):
     """
     Tests for the plugins.
     """
+
     def test_le_parser(self):
         """
         The ``le:`` parser uses the Let's Encrypt production directory, and
@@ -150,7 +175,7 @@ class PluginTests(TestCase):
         tempdir = self.useFixture(TempDir()).path
         temp_path = FilePath(tempdir)
         key_path = temp_path.child('client.key')
-        reactor = object()
+        reactor = MemoryReactorClock()
         self.assertThat(
             parser.parseStreamServer(
                 reactor, tempdir, 'tcp', '443', timeout=0),
@@ -172,11 +197,17 @@ class PluginTests(TestCase):
                         '%r is not a stream server endpoint'))))
         self.assertThat(key_path.isfile(), Equals(True))
         key_data = key_path.getContent()
-        parser.parseStreamServer(reactor, tempdir, 'tcp', '443'),
+
+        # Multiple instances with certificates from the same local directory,
+        # will serve the same certificates.
+        parser.parseStreamServer(reactor, tempdir, 'tcp', '443', timeout=0)
         self.assertThat(key_path.getContent(), Equals(key_data))
 
+        # Check that reactor is clean.
+        self.assertEquals(0, len(reactor.getDelayedCalls()))
+
 
-class LoadClientKeyTests(TestCase):
+class LoadClientKeyTests(TXACMETestCase):
     """
     Tests for `~txacme.endpoint.load_or_create_client_key`.
     """
diff --git a/src/txacme/test/test_service.py b/src/txacme/test/test_service.py
index 861b94d..d9833aa 100644
--- a/src/txacme/test/test_service.py
+++ b/src/txacme/test/test_service.py
@@ -19,7 +19,7 @@ from testtools.matchers import (
 from testtools.twistedsupport import (
     AsynchronousDeferredRunTest, failed, flush_logged_errors,
     has_no_result, succeeded)
-from twisted.internet.defer import CancelledError, Deferred, fail, succeed
+from twisted.internet.defer import CancelledError, Deferred, fail
 from twisted.internet.task import Clock
 from twisted.python.failure import Failure
 
@@ -96,7 +96,8 @@ class FailingClient(object):
     Test client that always fails.
     """
     def __getattr__(self, name):
-        return lambda *a, **kw: fail(RuntimeError('Tried to do something'))
+        return lambda *a, **kw: fail(
+            RuntimeError('Failing at "%s".' % (name,)))
 
 
 class AcmeFixture(Fixture):
@@ -128,7 +129,7 @@ class AcmeFixture(Fixture):
         self.responder = RecordingResponder(set(), u'tls-sni-01')
         args = dict(
             cert_store=self.cert_store,
-            client_creator=lambda: succeed(acme_client),
+            client=acme_client,
             clock=self.clock,
             responders=[self.responder],
             email=self._email,
@@ -332,11 +333,30 @@ class AcmeIssuingServiceTests(TestCase):
         error should be caught and logged.
         """
         with AcmeFixture(client=FailingClient()) as fixture:
+            # Registration is triggered with service starts.
             fixture.service.startService()
+            latest_logs = flush_logged_errors()
+            self.assertThat(latest_logs, HasLength(1))
+            self.assertThat(
+                str(latest_logs[0]), Contains('Failing at "register".'))
+
+            # Forcing a check will trigger again the registration.
             self.assertThat(
                 fixture.service._check_certs(),
                 succeeded(Always()))
-            self.assertThat(flush_logged_errors(), HasLength(2))
+
+            latest_logs = flush_logged_errors()
+            self.assertThat(latest_logs, HasLength(1))
+            self.assertThat(
+                str(latest_logs[0]), Contains('Failing at "register".'))
+
+            # Manually stop the service to not stop it from the fixture
+            # and trigger another failure.
+            self.assertThat(
+                fixture.service.stopService(),
+                failed(AfterPreprocessing(
+                    lambda f: f.value.args[0], Equals('Failing at "stop".'))))
+            latest_logs = flush_logged_errors()
 
     def test_starting_stopping_cancellation(self):
         """
diff --git a/src/txacme/testing.py b/src/txacme/testing.py
index 271085a..d091aad 100644
--- a/src/txacme/testing.py
+++ b/src/txacme/testing.py
@@ -11,6 +11,8 @@ from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.x509.oid import ExtensionOID, NameOID
+from testtools import TestCase
+from twisted.internet import reactor
 from twisted.internet.defer import Deferred, fail, succeed
 from twisted.python.compat import unicode
 from zope.interface import implementer
@@ -19,6 +21,24 @@ from txacme.interfaces import ICertificateStore, IResponder
 from txacme.util import clock_now, generate_private_key
 
 
+class TXACMETestCase(TestCase):
+    """
+    Common code for all tests for the txacme project.
+    """
+
+    def tearDown(self):
+        super(TXACMETestCase, self).tearDown()
+
+        # Make sure the main reactor is clean after each test.
+        junk = []
+        for delayed_call in reactor.getDelayedCalls():
+            junk.append(delayed_call.func)
+            delayed_call.cancel()
+        if junk:
+            raise AssertionError(
+                'Reactor is not clean. DelayedCalls: %s' % (junk,))
+
+
 @attr.s
 class FakeClientController(object):
     """
@@ -122,6 +142,13 @@ class FakeClient(object):
         self._ca_aki = x509.AuthorityKeyIdentifier.from_issuer_public_key(
             self._ca_key.public_key())
 
+    def stop(self):
+        """
+        Called to stop the client and trigger cleanups.
+        """
+        # Nothing to stop as reactor is not spun.
+        return succeed(None)
+
     def register(self, new_reg=None):
         self._registered = True
         if new_reg is None: