Run python setup.py test
to test whether socket ignores signals on
various Python versions. If it does then the test hangs.
Or just:
$ pythonX.Y test_socket_ignore_signal.py
to test Python X.Y version.
/.tox/ | |
/*.egg-info/ | |
/tox-*.egg/ |
Run python setup.py test
to test whether socket ignores signals on
various Python versions. If it does then the test hangs.
Or just:
$ pythonX.Y test_socket_ignore_signal.py
to test Python X.Y version.
#!/usr/bin/env python | |
"""Server program that doesn't respond. | |
From http://docs.python.org/release/3.3.0/library/socket.html | |
""" | |
import socket | |
import sys | |
def sprint(*args): | |
print((("server",) + args)) # print tuple to support Python 2.4+ | |
HOST = None # Symbolic name meaning all available interfaces | |
if len(sys.argv) > 1: | |
HOST, _, PORT = sys.argv[1].partition(':') | |
PORT = int(PORT) | |
else: | |
PORT = 31950 # arbitrary non-privileged port | |
s = None | |
for res in socket.getaddrinfo(HOST, PORT, | |
socket.AF_UNSPEC, | |
socket.SOCK_STREAM, | |
0, | |
socket.AI_PASSIVE): | |
af, socktype, proto, canonname, sa = res | |
sprint(sa) | |
try: | |
s = socket.socket(af, socktype, proto) | |
except socket.error: # use `socket.error` to support Python 2.4+ | |
s = None | |
continue | |
try: | |
s.bind(sa) | |
s.listen(1) | |
except socket.error: | |
s.close() | |
s = None | |
continue | |
break | |
if s is None: | |
sprint('could not open socket') | |
sys.exit(1) | |
conn, addr = s.accept() | |
sprint('Connected by', addr) | |
while True: | |
data = conn.recv(1024) | |
if not data: | |
break | |
sprint(repr(data)) | |
# do not respond | |
conn.close() |
import sys | |
from setuptools import setup | |
from setuptools.command.test import test as TestCommand | |
# `python setup.py test` installs and runs `tox` | |
class Tox(TestCommand): | |
def finalize_options(self): | |
TestCommand.finalize_options(self) | |
self.test_args = [] | |
self.test_suite = True | |
def run_tests(self): | |
#import here, cause outside the eggs aren't loaded | |
import tox | |
errno = tox.cmdline(self.test_args) | |
sys.exit(errno) | |
setup( | |
name='test-socket-ignore-signal', | |
tests_require=['tox'], | |
cmdclass={'test': Tox}, | |
url='https://gist.github.com/4069063', | |
) |
#!/usr/bin/env python | |
"""Test whether socket ignores signals.""" | |
import os | |
import signal | |
import socket | |
import sys | |
import time | |
from subprocess import Popen | |
class Alarm(Exception): | |
pass | |
def sigalrm_handler(signum, frame): | |
raise Alarm | |
def sprint(*args): | |
print((("client",) + args)) # print tuple to support Python 2.4+ | |
def readline(host, port): | |
"""Read a single line from host:port. | |
From http://docs.python.org/release/3.3.0/library/socket.html | |
""" | |
s = None | |
for _ in range(10): | |
for res in socket.getaddrinfo(host, port, | |
socket.AF_UNSPEC, socket.SOCK_STREAM): | |
af, socktype, proto, canonname, sa = res | |
sprint(sa) | |
try: | |
s = socket.socket(af, socktype, proto) | |
except socket.error: | |
s = None | |
continue | |
try: | |
s.connect(sa) | |
except socket.error: | |
s.close() | |
s = None | |
continue | |
break | |
else: | |
# no break; server might not be ready yet | |
if server_process.poll() is not None: | |
break | |
time.sleep(1) # allow server to start | |
continue | |
break | |
if s is None: | |
sprint('could not open socket') | |
sys.exit(1) | |
s.sendall('Hello, world'.encode('ascii')) | |
f = s.makefile('rb') | |
try: | |
data = f.readline() # should block | |
finally: | |
f.close() | |
s.close() | |
sprint('Received', repr(data)) | |
return data | |
port = 12345 | |
if len(sys.argv) > 1: | |
host = sys.argv[1] | |
else: | |
# get public network interface (according to docs) | |
host = socket.gethostbyname(socket.gethostname()) | |
# start server that doesn't respond | |
scriptdir = os.path.dirname(os.path.abspath(__file__)) | |
server_script = os.path.join(scriptdir, 'server.py') | |
server_process = Popen([sys.executable, server_script, "%s:%s" % (host, port)]) | |
# setup signal handler & alarm | |
signal.signal(signal.SIGALRM, sigalrm_handler) | |
delay = 3 # seconds | |
signal.alarm(delay) | |
sprint("wait %d seconds" % (delay,)) | |
try: | |
readline(host, port) | |
except Alarm: # got alarm | |
sprint('ALARM WORKS') | |
else: | |
assert 0, "shouldn't get here, readline() must block" |
[tox] | |
envlist = py25,py26,py27,pypy,py31,py32,py33 | |
[testenv] | |
commands= | |
python -V | |
python test_socket_ignore_signal.py [] |