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

Make the socket activation test simpler and more robust

Browse Source
This commit is contained in:
Kovid Goyal 2015-05-27 13:50:21 +05:30
parent b653cd2060
commit 1bc5dfc47f
1 changed files with 13 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -6,21 +6,19 @@ from __future__ import (unicode_literals, division, absolute_import,
__license__ = 'GPL v3' __license__ = 'GPL v3'
__copyright__ = '2015, Kovid Goyal <kovid at kovidgoyal.net>' __copyright__ = '2015, Kovid Goyal <kovid at kovidgoyal.net>'
import httplib, ssl, os, subprocess, time, socket import httplib, ssl, os, socket
from unittest import skipIf, skipUnless from unittest import skipIf
from calibre.constants import islinux
try: try:
from calibre.utils.certgen import create_server_cert from calibre.utils.certgen import create_server_cert
except ImportError: except ImportError:
create_server_cert = None create_server_cert = None
from calibre.srv.pre_activated import pre_activated_socket
from calibre.srv.tests.base import BaseTest, TestServer from calibre.srv.tests.base import BaseTest, TestServer
from calibre.ptempfile import TemporaryDirectory from calibre.ptempfile import TemporaryDirectory
SYSTEMD = '/usr/lib/systemd/systemd-activate'
class LoopTest(BaseTest): class LoopTest(BaseTest):
@skipIf(create_server_cert is None, 'certgen module not available') @skipIf(create_server_cert is None, 'certgen module not available')
@ -42,33 +40,20 @@ class LoopTest(BaseTest):
subject = dict(x[0] for x in cert['subject']) subject = dict(x[0] for x in cert['subject'])
self.ae(subject['commonName'], address) self.ae(subject['commonName'], address)
@skipUnless(islinux and os.access(SYSTEMD, os.X_OK), 'systemd-activate not available') @skipIf(pre_activated_socket is None, 'pre_activated_socket not available')
def test_systemd(self): def test_socket_activation(self):
'Test systemd socket activation' 'Test socket activation'
os.closerange(3, 4) # Ensure the socket gets fileno == 3
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
s.bind(('localhost', 0)) s.bind(('localhost', 0))
port = s.getsockname()[1] port = s.getsockname()[1]
f = os.path.abspath(__file__).rstrip('c') self.ae(s.fileno(), 3)
s.close() os.environ['LISTEN_PID'] = str(os.getpid())
p = subprocess.Popen([SYSTEMD, '-l', '127.0.0.1:%d' % port, 'calibre-debug', f], stdout=open(os.devnull, 'wb'), stderr=subprocess.STDOUT) os.environ['LISTEN_FDS'] = '1'
try: with TestServer(lambda data:(data.path[0] + data.read()), allow_socket_preallocation=True) as server:
conn = httplib.HTTPConnection('localhost', port, strict=True, timeout=1) conn = server.connect()
conn.request('GET', '/test', 'body') conn.request('GET', '/test', 'body')
r = conn.getresponse() r = conn.getresponse()
self.ae(r.status, httplib.OK) self.ae(r.status, httplib.OK)
self.ae(r.read(), b'testbody') self.ae(r.read(), b'testbody')
conn.request('GET', '/quit') self.ae(server.loop.bound_address[1], port)
st = time.time()
while p.poll() is None and time.time() - st < 2:
time.sleep(0.01)
finally:
if p.poll() is None:
p.kill()
if __name__ == '__main__':
def handler(data):
if data.path and data.path[0] == 'quit':
raise SystemExit(0)
return (data.path[0] + data.read()) if data.path else ''
with TestServer(handler, allow_socket_preallocation=True) as server:
server.join()