Shutting down an HTTPServer

For integration tests it can be helpful to have a fake HTTP server whose behaviour the tests can control. All necessary building blocks are even included in Python standard library. However, the BaseHTTPServer is surprisingly hard to shut down properly, so that it gives up the socket and everything.

While working on gocept.selenium, we came up with some code that does the trick (together with Jan-Wijbrand Kolman and Jan-Jaap Driessen).

class HTTPServer(BaseHTTPServer.HTTPServer):

    _continue = True

    def serve_until_shutdown(self):
        while self._continue:
            self.handle_request()

    def shutdown(self):
        self._continue = False
        # We fire a last request at the server in order to take it out of the
        # while loop in `self.serve_until_shutdown`.
        try:
            urllib2.urlopen(
                'http://%s:%s/' % (self.server_name, self.server_port))
        except urllib2.URLError:
            # If the server is already shut down, we receive a socket error,
            # which we ignore.
            pass
        self.server_close()

You might use this in a zope.testrunner layer like this:

class SilentRequestHandler(BaseHTTPServer.BaseHTTPRequestHandler):

    def log_message(self, format, *args):
        pass


class HTTPServerLayer(object):

    host = 'localhost'

    def setUp(self):
        self.server = None
        self.port = random.randint(30000, 40000)
        self.start_server()

    def start_server(self):
        self.server = HTTPServer((self.host, self.port), SilentRequestHandler)
        self.server_thread = threading.Thread(
            target=self.server.serve_until_shutdown)
        self.server_thread.daemon = True
        self.server_thread.start()
        # Kludge: Wait a little as it sometimes takes a while to get the server
        # started.
        time.sleep(0.25)

    def stop_server(self):
        if self.server is None:
            return
        self.server.shutdown()
        self.server_thread.join()

    def tearDown(self):
        self.stop_server()