Cutlery/calibre
1
0
Fork 0
mirror of https://github.com/kovidgoyal/calibre.git synced 2025-07-09 03:04:10 -04:00
Code Issues Packages Projects Releases Wiki Activity

Add a test for dual stack sockets

Browse Source
This commit is contained in:
Kovid Goyal 2019-01-16 09:57:52 +05:30
parent 0870125019
commit b00246070e
No known key found for this signature in database
GPG Key ID: 06BC317B515ACE7C
2 changed files with 15 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/calibre/srv/tests/base.py
View File

@ -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

11
src/calibre/srv/tests/loop.py
View File

@ -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):