diff --git a/src/calibre/srv/tests/base.py b/src/calibre/srv/tests/base.py index ddc0f19aa6..e62dcc605d 100644 --- a/src/calibre/srv/tests/base.py +++ b/src/calibre/srv/tests/base.py @@ -115,10 +115,12 @@ class TestServer(Thread): self.loop.stop() self.join(self.loop.opts.shutdown_timeout) - def connect(self, timeout=None): + def connect(self, timeout=None, interface=None): if timeout is None: timeout = self.loop.opts.timeout - return httplib.HTTPConnection(self.address[0], self.address[1], strict=True, timeout=timeout) + if interface is None: + interface = self.address[0] + return httplib.HTTPConnection(interface, self.address[1], strict=True, timeout=timeout) def change_handler(self, handler): from calibre.srv.http_response import create_http_handler diff --git a/src/calibre/srv/tests/loop.py b/src/calibre/srv/tests/loop.py index ff72a0cf3a..a3b1dbb962 100644 --- a/src/calibre/srv/tests/loop.py +++ b/src/calibre/srv/tests/loop.py @@ -126,6 +126,17 @@ class LoopTest(BaseTest): self.assertTrue(b.stopped.wait(5), 'BonJour not stopped') + def test_dual_stack(self): + with TestServer(lambda data:(data.path[0] + data.read()), listen_on='::') as server: + self.ae(server.address[0], '::') + self.ae(server.loop.socket.getsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY), 0) + for interface in ('::1', '127.0.0.1'): + conn = server.connect(interface=interface) + conn.request('GET', '/test', 'body') + r = conn.getresponse() + self.ae(r.status, httplib.OK) + self.ae(r.read(), b'testbody') + def test_ring_buffer(self): 'Test the ring buffer used for reads' class FakeSocket(object):