Infrastructure for testing http server

src/calibre/srv/tests/base.py

@@ -7,8 +7,9 @@ __license__   = 'GPL v3'
 __copyright__ = '2011, Kovid Goyal <kovid@kovidgoyal.net>'
 __docformat__ = 'restructuredtext en'
 
-import unittest, shutil
+import unittest, shutil, time, httplib
 from functools import partial
+from threading import Thread
 
 rmtree = partial(shutil.rmtree, ignore_errors=True)
 
@@ -20,3 +21,37 @@ class BaseTest(unittest.TestCase):
     ae = unittest.TestCase.assertEqual
 
 
+class TestServer(Thread):
+
+    daemon = True
+
+    def __init__(self, handler):
+        Thread.__init__(self, name='ServerMain')
+        from calibre.srv.opts import Options
+        from calibre.srv.loop import ServerLoop
+        from calibre.srv.http import create_http_handler
+        from calibre.utils.logging import ThreadSafeLog
+        self.loop = ServerLoop(
+            opts=Options(shutdown_timeout=0.1),
+            bind_address=('localhost', 0), http_handler=create_http_handler(handler),
+            log=ThreadSafeLog(level=ThreadSafeLog.WARN),
+        )
+
+    def run(self):
+        try:
+            self.loop.serve_forever()
+        except KeyboardInterrupt:
+            pass
+
+    def __enter__(self):
+        self.start()
+        while not self.loop.ready and self.is_alive():
+            time.sleep(0.01)
+        self.address = self.loop.bound_address[:2]
+        return self
+
+    def __exit__(self, *args):
+        self.loop.stop()
+
+    def connect(self):
+        return httplib.HTTPConnection(self.address[0], self.address[1], strict=True, timeout=0.1)

src/calibre/srv/tests/http.py

@@ -6,17 +6,17 @@ from __future__ import (unicode_literals, division, absolute_import,
 __license__ = 'GPL v3'
 __copyright__ = '2015, Kovid Goyal <kovid at kovidgoyal.net>'
 
-import textwrap
+import textwrap, httplib
 from io import BytesIO
 
-from calibre.srv.tests.base import BaseTest
+from calibre.srv.tests.base import BaseTest, TestServer
 
 def headers(raw):
     return BytesIO(textwrap.dedent(raw).encode('utf-8'))
 
 class TestHTTP(BaseTest):
 
-    def test_header_parsing(self):
+    def test_header_parsing(self):  # {{{
         'Test parsing of HTTP headers'
         from calibre.srv.http import read_headers
 
@@ -44,8 +44,9 @@ class TestHTTP(BaseTest):
             read_headers(headers('Connection:a\r\n').readline)
             read_headers(headers('Connection:a\n').readline)
             read_headers(headers(' Connection:a\n').readline)
+    # }}}
 
-    def test_accept_encoding(self):
+    def test_accept_encoding(self):  # {{{
         'Test parsing of Accept-Encoding'
         from calibre.srv.http import acceptable_encoding
         def test(name, val, ans, allowed={'gzip'}):
@@ -55,4 +56,16 @@ class TestHTTP(BaseTest):
         test('Case insensitive', 'GZIp', 'gzip')
         test('Multiple', 'gzip, identity', 'gzip')
         test('Priority', '1;q=0.5, 2;q=0.75, 3;q=1.0', '3', {'1', '2', '3'})
+        # }}}
 
+    def test_http_basic(self):  # {{{
+        'Test basic HTTP protocol conformance'
+        from calibre.srv.errors import HTTP404
+        def handler(conn):
+            raise HTTP404('Requested resource not found')
+        with TestServer(handler) as server:
+            conn = server.connect()
+            conn.request('HEAD', '/moose')
+            r = conn.getresponse()
+            self.ae(r.status, httplib.NOT_FOUND)
+    # }}}