mirror of
https://github.com/certbot/certbot.git
synced 2026-07-28 00:35:50 +02:00
100% coverage for standalone.ServerManager
This commit is contained in:
@@ -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``."""
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user