1berry/cpython
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix/optimize test_asyncore.test_quick_connect() (#1188)

Browse Source 
Don't use addCleanup() in test_quick_connect() because it keeps the
Thread object alive and so @reap_threads fails on its timeout of 1
second. "./python -m test -v test_asyncore -m test_quick_connect"
now takes 185 ms, instead of 11 seconds.

Other minor changes:

* Use "with sock:" to close the socket instead of
  try/finally: sock.close()
* Use self.skipTest() in test_quick_connect() to remove one
  indentation level and notice user that the test is specific to
  AF_INET and AF_INET6
This commit is contained in:
Victor Stinner 2017-04-19 23:42:46 +02:00 committed by GitHub
parent 1e62bf145b
commit 7b9619ae24
1 changed files with 37 additions and 38 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
Lib/test/test_asyncore.py
View File

@ -755,50 +755,49 @@ class BaseTestAPI:
def test_set_reuse_addr(self): def test_set_reuse_addr(self):
if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX: if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX:
self.skipTest("Not applicable to AF_UNIX sockets.") self.skipTest("Not applicable to AF_UNIX sockets.")
sock = socket.socket(self.family)
try: with socket.socket(self.family) as sock:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) try:
except OSError: sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
unittest.skip("SO_REUSEADDR not supported on this platform") except OSError:
else: unittest.skip("SO_REUSEADDR not supported on this platform")
# if SO_REUSEADDR succeeded for sock we expect asyncore else:
# to do the same # if SO_REUSEADDR succeeded for sock we expect asyncore
s = asyncore.dispatcher(socket.socket(self.family)) # to do the same
self.assertFalse(s.socket.getsockopt(socket.SOL_SOCKET, s = asyncore.dispatcher(socket.socket(self.family))
socket.SO_REUSEADDR)) self.assertFalse(s.socket.getsockopt(socket.SOL_SOCKET,
s.socket.close() socket.SO_REUSEADDR))
s.create_socket(self.family) s.socket.close()
s.set_reuse_addr() s.create_socket(self.family)
self.assertTrue(s.socket.getsockopt(socket.SOL_SOCKET, s.set_reuse_addr()
socket.SO_REUSEADDR)) self.assertTrue(s.socket.getsockopt(socket.SOL_SOCKET,
finally: socket.SO_REUSEADDR))
sock.close()
@unittest.skipUnless(threading, 'Threading required for this test.') @unittest.skipUnless(threading, 'Threading required for this test.')
@support.reap_threads @support.reap_threads
def test_quick_connect(self): def test_quick_connect(self):
# see: http://bugs.python.org/issue10340 # see: http://bugs.python.org/issue10340
if self.family in (socket.AF_INET, getattr(socket, "AF_INET6", object())): if self.family not in (socket.AF_INET, getattr(socket, "AF_INET6", object())):
server = BaseServer(self.family, self.addr) self.skipTest("test specific to AF_INET and AF_INET6")
t = threading.Thread(target=lambda: asyncore.loop(timeout=0.1,
count=500))
t.start()
def cleanup():
t.join(timeout=TIMEOUT)
if t.is_alive():
self.fail("join() timed out")
self.addCleanup(cleanup)
s = socket.socket(self.family, socket.SOCK_STREAM) server = BaseServer(self.family, self.addr)
s.settimeout(.2) t = threading.Thread(target=lambda: asyncore.loop(timeout=0.1,
s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, count=500), name="ident")
struct.pack('ii', 1, 0)) t.start()
try: try:
s.connect(server.address) with socket.socket(self.family, socket.SOCK_STREAM) as s:
except OSError: s.settimeout(.2)
pass s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
finally: struct.pack('ii', 1, 0))
s.close()
try:
s.connect(server.address)
except OSError:
pass
finally:
t.join(timeout=TIMEOUT)
if t.is_alive():
self.fail("join() timed out")
class TestAPI_UseIPv4Sockets(BaseTestAPI): class TestAPI_UseIPv4Sockets(BaseTestAPI):
family = socket.AF_INET family = socket.AF_INET