Don't use hardcoded port in tests (#6145)

* Don't use port 1234 in standalone tests.

* rename unused variable

* add back failure case

* Add back probe connection error test.

* fix lint

* remove unused import

* fix test file coverage

* prevent future heisenbug

acme/acme/crypto_util_test.py

@@ -42,28 +42,38 @@ class SSLSocketAndProbeSNITest(unittest.TestCase):
         self.server_thread = threading.Thread(
             # pylint: disable=no-member
             target=self.server.handle_request)
-        self.server_thread.start()
-        time.sleep(1)  # TODO: avoid race conditions in other way
 
     def tearDown(self):
-        self.server_thread.join()
+        if self.server_thread.is_alive():
+            # The thread may have already terminated.
+            self.server_thread.join()  # pragma: no cover
 
     def _probe(self, name):
         from acme.crypto_util import probe_sni
         return jose.ComparableX509(probe_sni(
             name, host='127.0.0.1', port=self.port))
 
+    def _start_server(self):
+        self.server_thread.start()
+        time.sleep(1)  # TODO: avoid race conditions in other way
+
     def test_probe_ok(self):
+        self._start_server()
         self.assertEqual(self.cert, self._probe(b'foo'))
 
     def test_probe_not_recognized_name(self):
+        self._start_server()
         self.assertRaises(errors.Error, self._probe, b'bar')
 
-    # TODO: py33/py34 tox hangs forever on do_handshake in second probe
+    def test_probe_connection_error(self):
-    #def probe_connection_error(self):
+        # pylint has a hard time with six
-    #    self._probe(b'foo')
+        self.server.server_close()  # pylint: disable=no-member
-    #    #time.sleep(1)  # TODO: avoid race conditions in other way
+        original_timeout = socket.getdefaulttimeout()
-    #    self.assertRaises(errors.Error, self._probe, b'bar')
+        try:
+            socket.setdefaulttimeout(1)
+            self.assertRaises(errors.Error, self._probe, b'bar')
+        finally:
+            socket.setdefaulttimeout(original_timeout)
 
 
 class PyOpenSSLCertOrReqAllNamesTest(unittest.TestCase):

acme/acme/standalone_test.py

@@ -4,10 +4,10 @@ import shutil
 import socket
 import threading
 import tempfile
-import time
 import unittest
 
 from six.moves import http_client  # pylint: disable=import-error
+from six.moves import queue  # pylint: disable=import-error
 from six.moves import socketserver  # type: ignore  # pylint: disable=import-error
 
 import josepy as jose
@@ -16,7 +16,6 @@ import requests
 
 from acme import challenges
 from acme import crypto_util
-from acme import errors
 from acme import test_util
 from acme.magic_typing import Set # pylint: disable=unused-import, no-name-in-module
 
@@ -261,10 +260,9 @@ class TestSimpleTLSSNI01Server(unittest.TestCase):
                     os.path.join(localhost_dir, 'key.pem'))
 
         from acme.standalone import simple_tls_sni_01_server
-        self.port = 1234
         self.thread = threading.Thread(
             target=simple_tls_sni_01_server, kwargs={
-                'cli_args': ('xxx', '--port', str(self.port)),
+                'cli_args': ('filename',),
                 'forever': False,
             },
         )
@@ -276,25 +274,20 @@ class TestSimpleTLSSNI01Server(unittest.TestCase):
         self.thread.join()
         shutil.rmtree(self.test_cwd)
 
-    def test_it(self):
+    @mock.patch('acme.standalone.logger')
-        max_attempts = 5
+    def test_it(self, mock_logger):
-        for attempt in range(max_attempts):
+        # Use a Queue because mock objects aren't thread safe.
-            try:
+        q = queue.Queue()  # type: queue.Queue[int]
-                cert = crypto_util.probe_sni(
+        # Add port number to the queue.
-                    b'localhost', b'0.0.0.0', self.port)
+        mock_logger.info.side_effect = lambda *args: q.put(args[-1])
-            except errors.Error:
+        self.thread.start()
-                self.assertTrue(attempt + 1 < max_attempts, "Timeout!")
-                time.sleep(1)  # wait until thread starts
-            else:
-                self.assertEqual(jose.ComparableX509(cert),
-                                 test_util.load_comparable_cert(
-                                     'rsa2048_cert.pem'))
-                break
 
-            if attempt == 0:
+        # After the timeout, an exception is raised if the queue is empty.
-                # the first attempt is always meant to fail, so we can test
+        port = q.get(timeout=5)
-                # the socket failure code-path for probe_sni, as well
+        cert = crypto_util.probe_sni(b'localhost', b'0.0.0.0', port)
-                self.thread.start()
+        self.assertEqual(jose.ComparableX509(cert),
+                         test_util.load_comparable_cert(
+                             'rsa2048_cert.pem'))
 
 
 if __name__ == "__main__":