From 7687ecd6e36dbb15fcc633bc3c98123cf9605859 Mon Sep 17 00:00:00 2001 From: Jakub Warmuz Date: Sun, 4 Oct 2015 11:41:19 +0000 Subject: [PATCH] 100% coverage for standalone.ServerManager --- letsencrypt/plugins/standalone.py | 9 ++++--- letsencrypt/plugins/standalone_test.py | 34 ++++++++++++++++++++++++++ 2 files changed, 40 insertions(+), 3 deletions(-) diff --git a/letsencrypt/plugins/standalone.py b/letsencrypt/plugins/standalone.py index 071fdbf42..b3fb92ad1 100644 --- a/letsencrypt/plugins/standalone.py +++ b/letsencrypt/plugins/standalone.py @@ -51,12 +51,15 @@ class ServerManager(object): try: server = cls(("", port), handler) except socket.error as error: - errors.StandaloneBindError(error, port) + raise errors.StandaloneBindError(error, port) + + # if port == 0, then random free port on OS is taken + real_port = server.socket.getsockname() thread = threading.Thread(target=server.serve_forever2) thread.start() - self.servers[port] = (server, thread) - return self.servers[port] + self.servers[real_port] = (server, thread) + return self.servers[real_port] def stop(self, port): """Stop ACME server running on the specified ``port``.""" diff --git a/letsencrypt/plugins/standalone_test.py b/letsencrypt/plugins/standalone_test.py index 8c0dd6f97..ad8122c78 100644 --- a/letsencrypt/plugins/standalone_test.py +++ b/letsencrypt/plugins/standalone_test.py @@ -1,6 +1,9 @@ """Tests for letsencrypt.plugins.standalone.""" +import socket import unittest +from letsencrypt import errors + class ServerManagerTest(unittest.TestCase): """Tests for letsencrypt.plugins.standalone.ServerManager.""" @@ -16,6 +19,37 @@ class ServerManagerTest(unittest.TestCase): self.assertTrue( self.mgr.simple_http_resources is self.simple_http_resources) + def test_run_stop_non_tls(self): + server, thread = self.mgr.run(port=0, tls=False) + self.mgr.stop(port=server.socket.getsockname()) + + def test_run_stop_tls(self): + server, thread = self.mgr.run(port=0, tls=True) + self.mgr.stop(port=server.socket.getsockname()) + + def test_run_idempotent(self): + server, thread = self.mgr.run(port=0, tls=False) + port = server.socket.getsockname() + server2, thread2 = self.mgr.run(port=port, tls=False) + self.assertTrue(server is server2) + self.assertTrue(thread2 is thread2) + self.mgr.stop(port) + + def test_run_bind_error(self): + some_server = socket.socket() + some_server.bind(("", 0)) + port = some_server.getsockname()[1] + self.assertRaises( + errors.StandaloneBindError, self.mgr.run, port, tls=False) + + def test_items(self): + server, thread = self.mgr.run(port=0, tls=True) + port = server.socket.getsockname() + self.assertEqual(port, self.mgr.items()[0][0]) + self.assertTrue(self.mgr.items()[0][1][0] is server) + self.assertTrue(self.mgr.items()[0][1][1] is thread) + self.mgr.stop(port=port) + if __name__ == "__main__": unittest.main() # pragma: no cover