Python code coverage for Lib/test/test_asyncore.py

#	count	content
1	n/a	import asyncore
2	n/a	import unittest
3	n/a	import select
4	n/a	import os
5	n/a	import socket
6	n/a	import sys
7	n/a	import time
8	n/a	import errno
9	n/a	import struct
10	n/a
11	n/a	from test import support
12	n/a	from io import BytesIO
13	n/a
14	n/a	if support.PGO:
15	n/a	raise unittest.SkipTest("test is not helpful for PGO")
16	n/a
17	n/a	try:
18	n/a	import threading
19	n/a	except ImportError:
20	n/a	threading = None
21	n/a
22	n/a	TIMEOUT = 3
23	n/a	HAS_UNIX_SOCKETS = hasattr(socket, 'AF_UNIX')
24	n/a
25	n/a	class dummysocket:
26	n/a	def __init__(self):
27	n/a	self.closed = False
28	n/a
29	n/a	def close(self):
30	n/a	self.closed = True
31	n/a
32	n/a	def fileno(self):
33	n/a	return 42
34	n/a
35	n/a	class dummychannel:
36	n/a	def __init__(self):
37	n/a	self.socket = dummysocket()
38	n/a
39	n/a	def close(self):
40	n/a	self.socket.close()
41	n/a
42	n/a	class exitingdummy:
43	n/a	def __init__(self):
44	n/a	pass
45	n/a
46	n/a	def handle_read_event(self):
47	n/a	raise asyncore.ExitNow()
48	n/a
49	n/a	handle_write_event = handle_read_event
50	n/a	handle_close = handle_read_event
51	n/a	handle_expt_event = handle_read_event
52	n/a
53	n/a	class crashingdummy:
54	n/a	def __init__(self):
55	n/a	self.error_handled = False
56	n/a
57	n/a	def handle_read_event(self):
58	n/a	raise Exception()
59	n/a
60	n/a	handle_write_event = handle_read_event
61	n/a	handle_close = handle_read_event
62	n/a	handle_expt_event = handle_read_event
63	n/a
64	n/a	def handle_error(self):
65	n/a	self.error_handled = True
66	n/a
67	n/a	# used when testing senders; just collects what it gets until newline is sent
68	n/a	def capture_server(evt, buf, serv):
69	n/a	try:
70	n/a	serv.listen()
71	n/a	conn, addr = serv.accept()
72	n/a	except socket.timeout:
73	n/a	pass
74	n/a	else:
75	n/a	n = 200
76	n/a	start = time.time()
77	n/a	while n > 0 and time.time() - start < 3.0:
78	n/a	r, w, e = select.select([conn], [], [], 0.1)
79	n/a	if r:
80	n/a	n -= 1
81	n/a	data = conn.recv(10)
82	n/a	# keep everything except for the newline terminator
83	n/a	buf.write(data.replace(b'\n', b''))
84	n/a	if b'\n' in data:
85	n/a	break
86	n/a	time.sleep(0.01)
87	n/a
88	n/a	conn.close()
89	n/a	finally:
90	n/a	serv.close()
91	n/a	evt.set()
92	n/a
93	n/a	def bind_af_aware(sock, addr):
94	n/a	"""Helper function to bind a socket according to its family."""
95	n/a	if HAS_UNIX_SOCKETS and sock.family == socket.AF_UNIX:
96	n/a	# Make sure the path doesn't exist.
97	n/a	support.unlink(addr)
98	n/a	support.bind_unix_socket(sock, addr)
99	n/a	else:
100	n/a	sock.bind(addr)
101	n/a
102	n/a
103	n/a	class HelperFunctionTests(unittest.TestCase):
104	n/a	def test_readwriteexc(self):
105	n/a	# Check exception handling behavior of read, write and _exception
106	n/a
107	n/a	# check that ExitNow exceptions in the object handler method
108	n/a	# bubbles all the way up through asyncore read/write/_exception calls
109	n/a	tr1 = exitingdummy()
110	n/a	self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)
111	n/a	self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)
112	n/a	self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)
113	n/a
114	n/a	# check that an exception other than ExitNow in the object handler
115	n/a	# method causes the handle_error method to get called
116	n/a	tr2 = crashingdummy()
117	n/a	asyncore.read(tr2)
118	n/a	self.assertEqual(tr2.error_handled, True)
119	n/a
120	n/a	tr2 = crashingdummy()
121	n/a	asyncore.write(tr2)
122	n/a	self.assertEqual(tr2.error_handled, True)
123	n/a
124	n/a	tr2 = crashingdummy()
125	n/a	asyncore._exception(tr2)
126	n/a	self.assertEqual(tr2.error_handled, True)
127	n/a
128	n/a	# asyncore.readwrite uses constants in the select module that
129	n/a	# are not present in Windows systems (see this thread:
130	n/a	# http://mail.python.org/pipermail/python-list/2001-October/109973.html)
131	n/a	# These constants should be present as long as poll is available
132	n/a
133	n/a	@unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
134	n/a	def test_readwrite(self):
135	n/a	# Check that correct methods are called by readwrite()
136	n/a
137	n/a	attributes = ('read', 'expt', 'write', 'closed', 'error_handled')
138	n/a
139	n/a	expected = (
140	n/a	(select.POLLIN, 'read'),
141	n/a	(select.POLLPRI, 'expt'),
142	n/a	(select.POLLOUT, 'write'),
143	n/a	(select.POLLERR, 'closed'),
144	n/a	(select.POLLHUP, 'closed'),
145	n/a	(select.POLLNVAL, 'closed'),
146	n/a	)
147	n/a
148	n/a	class testobj:
149	n/a	def __init__(self):
150	n/a	self.read = False
151	n/a	self.write = False
152	n/a	self.closed = False
153	n/a	self.expt = False
154	n/a	self.error_handled = False
155	n/a
156	n/a	def handle_read_event(self):
157	n/a	self.read = True
158	n/a
159	n/a	def handle_write_event(self):
160	n/a	self.write = True
161	n/a
162	n/a	def handle_close(self):
163	n/a	self.closed = True
164	n/a
165	n/a	def handle_expt_event(self):
166	n/a	self.expt = True
167	n/a
168	n/a	def handle_error(self):
169	n/a	self.error_handled = True
170	n/a
171	n/a	for flag, expectedattr in expected:
172	n/a	tobj = testobj()
173	n/a	self.assertEqual(getattr(tobj, expectedattr), False)
174	n/a	asyncore.readwrite(tobj, flag)
175	n/a
176	n/a	# Only the attribute modified by the routine we expect to be
177	n/a	# called should be True.
178	n/a	for attr in attributes:
179	n/a	self.assertEqual(getattr(tobj, attr), attr==expectedattr)
180	n/a
181	n/a	# check that ExitNow exceptions in the object handler method
182	n/a	# bubbles all the way up through asyncore readwrite call
183	n/a	tr1 = exitingdummy()
184	n/a	self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag)
185	n/a
186	n/a	# check that an exception other than ExitNow in the object handler
187	n/a	# method causes the handle_error method to get called
188	n/a	tr2 = crashingdummy()
189	n/a	self.assertEqual(tr2.error_handled, False)
190	n/a	asyncore.readwrite(tr2, flag)
191	n/a	self.assertEqual(tr2.error_handled, True)
192	n/a
193	n/a	def test_closeall(self):
194	n/a	self.closeall_check(False)
195	n/a
196	n/a	def test_closeall_default(self):
197	n/a	self.closeall_check(True)
198	n/a
199	n/a	def closeall_check(self, usedefault):
200	n/a	# Check that close_all() closes everything in a given map
201	n/a
202	n/a	l = []
203	n/a	testmap = {}
204	n/a	for i in range(10):
205	n/a	c = dummychannel()
206	n/a	l.append(c)
207	n/a	self.assertEqual(c.socket.closed, False)
208	n/a	testmap[i] = c
209	n/a
210	n/a	if usedefault:
211	n/a	socketmap = asyncore.socket_map
212	n/a	try:
213	n/a	asyncore.socket_map = testmap
214	n/a	asyncore.close_all()
215	n/a	finally:
216	n/a	testmap, asyncore.socket_map = asyncore.socket_map, socketmap
217	n/a	else:
218	n/a	asyncore.close_all(testmap)
219	n/a
220	n/a	self.assertEqual(len(testmap), 0)
221	n/a
222	n/a	for c in l:
223	n/a	self.assertEqual(c.socket.closed, True)
224	n/a
225	n/a	def test_compact_traceback(self):
226	n/a	try:
227	n/a	raise Exception("I don't like spam!")
228	n/a	except:
229	n/a	real_t, real_v, real_tb = sys.exc_info()
230	n/a	r = asyncore.compact_traceback()
231	n/a	else:
232	n/a	self.fail("Expected exception")
233	n/a
234	n/a	(f, function, line), t, v, info = r
235	n/a	self.assertEqual(os.path.split(f)[-1], 'test_asyncore.py')
236	n/a	self.assertEqual(function, 'test_compact_traceback')
237	n/a	self.assertEqual(t, real_t)
238	n/a	self.assertEqual(v, real_v)
239	n/a	self.assertEqual(info, '[%s\|%s\|%s]' % (f, function, line))
240	n/a
241	n/a
242	n/a	class DispatcherTests(unittest.TestCase):
243	n/a	def setUp(self):
244	n/a	pass
245	n/a
246	n/a	def tearDown(self):
247	n/a	asyncore.close_all()
248	n/a
249	n/a	def test_basic(self):
250	n/a	d = asyncore.dispatcher()
251	n/a	self.assertEqual(d.readable(), True)
252	n/a	self.assertEqual(d.writable(), True)
253	n/a
254	n/a	def test_repr(self):
255	n/a	d = asyncore.dispatcher()
256	n/a	self.assertEqual(repr(d), '<asyncore.dispatcher at %#x>' % id(d))
257	n/a
258	n/a	def test_log(self):
259	n/a	d = asyncore.dispatcher()
260	n/a
261	n/a	# capture output of dispatcher.log() (to stderr)
262	n/a	l1 = "Lovely spam! Wonderful spam!"
263	n/a	l2 = "I don't like spam!"
264	n/a	with support.captured_stderr() as stderr:
265	n/a	d.log(l1)
266	n/a	d.log(l2)
267	n/a
268	n/a	lines = stderr.getvalue().splitlines()
269	n/a	self.assertEqual(lines, ['log: %s' % l1, 'log: %s' % l2])
270	n/a
271	n/a	def test_log_info(self):
272	n/a	d = asyncore.dispatcher()
273	n/a
274	n/a	# capture output of dispatcher.log_info() (to stdout via print)
275	n/a	l1 = "Have you got anything without spam?"
276	n/a	l2 = "Why can't she have egg bacon spam and sausage?"
277	n/a	l3 = "THAT'S got spam in it!"
278	n/a	with support.captured_stdout() as stdout:
279	n/a	d.log_info(l1, 'EGGS')
280	n/a	d.log_info(l2)
281	n/a	d.log_info(l3, 'SPAM')
282	n/a
283	n/a	lines = stdout.getvalue().splitlines()
284	n/a	expected = ['EGGS: %s' % l1, 'info: %s' % l2, 'SPAM: %s' % l3]
285	n/a	self.assertEqual(lines, expected)
286	n/a
287	n/a	def test_unhandled(self):
288	n/a	d = asyncore.dispatcher()
289	n/a	d.ignore_log_types = ()
290	n/a
291	n/a	# capture output of dispatcher.log_info() (to stdout via print)
292	n/a	with support.captured_stdout() as stdout:
293	n/a	d.handle_expt()
294	n/a	d.handle_read()
295	n/a	d.handle_write()
296	n/a	d.handle_connect()
297	n/a
298	n/a	lines = stdout.getvalue().splitlines()
299	n/a	expected = ['warning: unhandled incoming priority event',
300	n/a	'warning: unhandled read event',
301	n/a	'warning: unhandled write event',
302	n/a	'warning: unhandled connect event']
303	n/a	self.assertEqual(lines, expected)
304	n/a
305	n/a	def test_strerror(self):
306	n/a	# refers to bug #8573
307	n/a	err = asyncore._strerror(errno.EPERM)
308	n/a	if hasattr(os, 'strerror'):
309	n/a	self.assertEqual(err, os.strerror(errno.EPERM))
310	n/a	err = asyncore._strerror(-1)
311	n/a	self.assertTrue(err != "")
312	n/a
313	n/a
314	n/a	class dispatcherwithsend_noread(asyncore.dispatcher_with_send):
315	n/a	def readable(self):
316	n/a	return False
317	n/a
318	n/a	def handle_connect(self):
319	n/a	pass
320	n/a
321	n/a
322	n/a	class DispatcherWithSendTests(unittest.TestCase):
323	n/a	def setUp(self):
324	n/a	pass
325	n/a
326	n/a	def tearDown(self):
327	n/a	asyncore.close_all()
328	n/a
329	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
330	n/a	@support.reap_threads
331	n/a	def test_send(self):
332	n/a	evt = threading.Event()
333	n/a	sock = socket.socket()
334	n/a	sock.settimeout(3)
335	n/a	port = support.bind_port(sock)
336	n/a
337	n/a	cap = BytesIO()
338	n/a	args = (evt, cap, sock)
339	n/a	t = threading.Thread(target=capture_server, args=args)
340	n/a	t.start()
341	n/a	try:
342	n/a	# wait a little longer for the server to initialize (it sometimes
343	n/a	# refuses connections on slow machines without this wait)
344	n/a	time.sleep(0.2)
345	n/a
346	n/a	data = b"Suppose there isn't a 16-ton weight?"
347	n/a	d = dispatcherwithsend_noread()
348	n/a	d.create_socket()
349	n/a	d.connect((support.HOST, port))
350	n/a
351	n/a	# give time for socket to connect
352	n/a	time.sleep(0.1)
353	n/a
354	n/a	d.send(data)
355	n/a	d.send(data)
356	n/a	d.send(b'\n')
357	n/a
358	n/a	n = 1000
359	n/a	while d.out_buffer and n > 0:
360	n/a	asyncore.poll()
361	n/a	n -= 1
362	n/a
363	n/a	evt.wait()
364	n/a
365	n/a	self.assertEqual(cap.getvalue(), data*2)
366	n/a	finally:
367	n/a	t.join(timeout=TIMEOUT)
368	n/a	if t.is_alive():
369	n/a	self.fail("join() timed out")
370	n/a
371	n/a
372	n/a	@unittest.skipUnless(hasattr(asyncore, 'file_wrapper'),
373	n/a	'asyncore.file_wrapper required')
374	n/a	class FileWrapperTest(unittest.TestCase):
375	n/a	def setUp(self):
376	n/a	self.d = b"It's not dead, it's sleeping!"
377	n/a	with open(support.TESTFN, 'wb') as file:
378	n/a	file.write(self.d)
379	n/a
380	n/a	def tearDown(self):
381	n/a	support.unlink(support.TESTFN)
382	n/a
383	n/a	def test_recv(self):
384	n/a	fd = os.open(support.TESTFN, os.O_RDONLY)
385	n/a	w = asyncore.file_wrapper(fd)
386	n/a	os.close(fd)
387	n/a
388	n/a	self.assertNotEqual(w.fd, fd)
389	n/a	self.assertNotEqual(w.fileno(), fd)
390	n/a	self.assertEqual(w.recv(13), b"It's not dead")
391	n/a	self.assertEqual(w.read(6), b", it's")
392	n/a	w.close()
393	n/a	self.assertRaises(OSError, w.read, 1)
394	n/a
395	n/a	def test_send(self):
396	n/a	d1 = b"Come again?"
397	n/a	d2 = b"I want to buy some cheese."
398	n/a	fd = os.open(support.TESTFN, os.O_WRONLY \| os.O_APPEND)
399	n/a	w = asyncore.file_wrapper(fd)
400	n/a	os.close(fd)
401	n/a
402	n/a	w.write(d1)
403	n/a	w.send(d2)
404	n/a	w.close()
405	n/a	with open(support.TESTFN, 'rb') as file:
406	n/a	self.assertEqual(file.read(), self.d + d1 + d2)
407	n/a
408	n/a	@unittest.skipUnless(hasattr(asyncore, 'file_dispatcher'),
409	n/a	'asyncore.file_dispatcher required')
410	n/a	def test_dispatcher(self):
411	n/a	fd = os.open(support.TESTFN, os.O_RDONLY)
412	n/a	data = []
413	n/a	class FileDispatcher(asyncore.file_dispatcher):
414	n/a	def handle_read(self):
415	n/a	data.append(self.recv(29))
416	n/a	s = FileDispatcher(fd)
417	n/a	os.close(fd)
418	n/a	asyncore.loop(timeout=0.01, use_poll=True, count=2)
419	n/a	self.assertEqual(b"".join(data), self.d)
420	n/a
421	n/a	def test_resource_warning(self):
422	n/a	# Issue #11453
423	n/a	fd = os.open(support.TESTFN, os.O_RDONLY)
424	n/a	f = asyncore.file_wrapper(fd)
425	n/a
426	n/a	os.close(fd)
427	n/a	with support.check_warnings(('', ResourceWarning)):
428	n/a	f = None
429	n/a	support.gc_collect()
430	n/a
431	n/a	def test_close_twice(self):
432	n/a	fd = os.open(support.TESTFN, os.O_RDONLY)
433	n/a	f = asyncore.file_wrapper(fd)
434	n/a	os.close(fd)
435	n/a
436	n/a	f.close()
437	n/a	self.assertEqual(f.fd, -1)
438	n/a	# calling close twice should not fail
439	n/a	f.close()
440	n/a
441	n/a
442	n/a	class BaseTestHandler(asyncore.dispatcher):
443	n/a
444	n/a	def __init__(self, sock=None):
445	n/a	asyncore.dispatcher.__init__(self, sock)
446	n/a	self.flag = False
447	n/a
448	n/a	def handle_accept(self):
449	n/a	raise Exception("handle_accept not supposed to be called")
450	n/a
451	n/a	def handle_accepted(self):
452	n/a	raise Exception("handle_accepted not supposed to be called")
453	n/a
454	n/a	def handle_connect(self):
455	n/a	raise Exception("handle_connect not supposed to be called")
456	n/a
457	n/a	def handle_expt(self):
458	n/a	raise Exception("handle_expt not supposed to be called")
459	n/a
460	n/a	def handle_close(self):
461	n/a	raise Exception("handle_close not supposed to be called")
462	n/a
463	n/a	def handle_error(self):
464	n/a	raise
465	n/a
466	n/a
467	n/a	class BaseServer(asyncore.dispatcher):
468	n/a	"""A server which listens on an address and dispatches the
469	n/a	connection to a handler.
470	n/a	"""
471	n/a
472	n/a	def __init__(self, family, addr, handler=BaseTestHandler):
473	n/a	asyncore.dispatcher.__init__(self)
474	n/a	self.create_socket(family)
475	n/a	self.set_reuse_addr()
476	n/a	bind_af_aware(self.socket, addr)
477	n/a	self.listen(5)
478	n/a	self.handler = handler
479	n/a
480	n/a	@property
481	n/a	def address(self):
482	n/a	return self.socket.getsockname()
483	n/a
484	n/a	def handle_accepted(self, sock, addr):
485	n/a	self.handler(sock)
486	n/a
487	n/a	def handle_error(self):
488	n/a	raise
489	n/a
490	n/a
491	n/a	class BaseClient(BaseTestHandler):
492	n/a
493	n/a	def __init__(self, family, address):
494	n/a	BaseTestHandler.__init__(self)
495	n/a	self.create_socket(family)
496	n/a	self.connect(address)
497	n/a
498	n/a	def handle_connect(self):
499	n/a	pass
500	n/a
501	n/a
502	n/a	class BaseTestAPI:
503	n/a
504	n/a	def tearDown(self):
505	n/a	asyncore.close_all()
506	n/a
507	n/a	def loop_waiting_for_flag(self, instance, timeout=5):
508	n/a	timeout = float(timeout) / 100
509	n/a	count = 100
510	n/a	while asyncore.socket_map and count > 0:
511	n/a	asyncore.loop(timeout=0.01, count=1, use_poll=self.use_poll)
512	n/a	if instance.flag:
513	n/a	return
514	n/a	count -= 1
515	n/a	time.sleep(timeout)
516	n/a	self.fail("flag not set")
517	n/a
518	n/a	def test_handle_connect(self):
519	n/a	# make sure handle_connect is called on connect()
520	n/a
521	n/a	class TestClient(BaseClient):
522	n/a	def handle_connect(self):
523	n/a	self.flag = True
524	n/a
525	n/a	server = BaseServer(self.family, self.addr)
526	n/a	client = TestClient(self.family, server.address)
527	n/a	self.loop_waiting_for_flag(client)
528	n/a
529	n/a	def test_handle_accept(self):
530	n/a	# make sure handle_accept() is called when a client connects
531	n/a
532	n/a	class TestListener(BaseTestHandler):
533	n/a
534	n/a	def __init__(self, family, addr):
535	n/a	BaseTestHandler.__init__(self)
536	n/a	self.create_socket(family)
537	n/a	bind_af_aware(self.socket, addr)
538	n/a	self.listen(5)
539	n/a	self.address = self.socket.getsockname()
540	n/a
541	n/a	def handle_accept(self):
542	n/a	self.flag = True
543	n/a
544	n/a	server = TestListener(self.family, self.addr)
545	n/a	client = BaseClient(self.family, server.address)
546	n/a	self.loop_waiting_for_flag(server)
547	n/a
548	n/a	def test_handle_accepted(self):
549	n/a	# make sure handle_accepted() is called when a client connects
550	n/a
551	n/a	class TestListener(BaseTestHandler):
552	n/a
553	n/a	def __init__(self, family, addr):
554	n/a	BaseTestHandler.__init__(self)
555	n/a	self.create_socket(family)
556	n/a	bind_af_aware(self.socket, addr)
557	n/a	self.listen(5)
558	n/a	self.address = self.socket.getsockname()
559	n/a
560	n/a	def handle_accept(self):
561	n/a	asyncore.dispatcher.handle_accept(self)
562	n/a
563	n/a	def handle_accepted(self, sock, addr):
564	n/a	sock.close()
565	n/a	self.flag = True
566	n/a
567	n/a	server = TestListener(self.family, self.addr)
568	n/a	client = BaseClient(self.family, server.address)
569	n/a	self.loop_waiting_for_flag(server)
570	n/a
571	n/a
572	n/a	def test_handle_read(self):
573	n/a	# make sure handle_read is called on data received
574	n/a
575	n/a	class TestClient(BaseClient):
576	n/a	def handle_read(self):
577	n/a	self.flag = True
578	n/a
579	n/a	class TestHandler(BaseTestHandler):
580	n/a	def __init__(self, conn):
581	n/a	BaseTestHandler.__init__(self, conn)
582	n/a	self.send(b'x' * 1024)
583	n/a
584	n/a	server = BaseServer(self.family, self.addr, TestHandler)
585	n/a	client = TestClient(self.family, server.address)
586	n/a	self.loop_waiting_for_flag(client)
587	n/a
588	n/a	def test_handle_write(self):
589	n/a	# make sure handle_write is called
590	n/a
591	n/a	class TestClient(BaseClient):
592	n/a	def handle_write(self):
593	n/a	self.flag = True
594	n/a
595	n/a	server = BaseServer(self.family, self.addr)
596	n/a	client = TestClient(self.family, server.address)
597	n/a	self.loop_waiting_for_flag(client)
598	n/a
599	n/a	def test_handle_close(self):
600	n/a	# make sure handle_close is called when the other end closes
601	n/a	# the connection
602	n/a
603	n/a	class TestClient(BaseClient):
604	n/a
605	n/a	def handle_read(self):
606	n/a	# in order to make handle_close be called we are supposed
607	n/a	# to make at least one recv() call
608	n/a	self.recv(1024)
609	n/a
610	n/a	def handle_close(self):
611	n/a	self.flag = True
612	n/a	self.close()
613	n/a
614	n/a	class TestHandler(BaseTestHandler):
615	n/a	def __init__(self, conn):
616	n/a	BaseTestHandler.__init__(self, conn)
617	n/a	self.close()
618	n/a
619	n/a	server = BaseServer(self.family, self.addr, TestHandler)
620	n/a	client = TestClient(self.family, server.address)
621	n/a	self.loop_waiting_for_flag(client)
622	n/a
623	n/a	def test_handle_close_after_conn_broken(self):
624	n/a	# Check that ECONNRESET/EPIPE is correctly handled (issues #5661 and
625	n/a	# #11265).
626	n/a
627	n/a	data = b'\0' * 128
628	n/a
629	n/a	class TestClient(BaseClient):
630	n/a
631	n/a	def handle_write(self):
632	n/a	self.send(data)
633	n/a
634	n/a	def handle_close(self):
635	n/a	self.flag = True
636	n/a	self.close()
637	n/a
638	n/a	def handle_expt(self):
639	n/a	self.flag = True
640	n/a	self.close()
641	n/a
642	n/a	class TestHandler(BaseTestHandler):
643	n/a
644	n/a	def handle_read(self):
645	n/a	self.recv(len(data))
646	n/a	self.close()
647	n/a
648	n/a	def writable(self):
649	n/a	return False
650	n/a
651	n/a	server = BaseServer(self.family, self.addr, TestHandler)
652	n/a	client = TestClient(self.family, server.address)
653	n/a	self.loop_waiting_for_flag(client)
654	n/a
655	n/a	@unittest.skipIf(sys.platform.startswith("sunos"),
656	n/a	"OOB support is broken on Solaris")
657	n/a	def test_handle_expt(self):
658	n/a	# Make sure handle_expt is called on OOB data received.
659	n/a	# Note: this might fail on some platforms as OOB data is
660	n/a	# tenuously supported and rarely used.
661	n/a	if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX:
662	n/a	self.skipTest("Not applicable to AF_UNIX sockets.")
663	n/a
664	n/a	class TestClient(BaseClient):
665	n/a	def handle_expt(self):
666	n/a	self.socket.recv(1024, socket.MSG_OOB)
667	n/a	self.flag = True
668	n/a
669	n/a	class TestHandler(BaseTestHandler):
670	n/a	def __init__(self, conn):
671	n/a	BaseTestHandler.__init__(self, conn)
672	n/a	self.socket.send(bytes(chr(244), 'latin-1'), socket.MSG_OOB)
673	n/a
674	n/a	server = BaseServer(self.family, self.addr, TestHandler)
675	n/a	client = TestClient(self.family, server.address)
676	n/a	self.loop_waiting_for_flag(client)
677	n/a
678	n/a	def test_handle_error(self):
679	n/a
680	n/a	class TestClient(BaseClient):
681	n/a	def handle_write(self):
682	n/a	1.0 / 0
683	n/a	def handle_error(self):
684	n/a	self.flag = True
685	n/a	try:
686	n/a	raise
687	n/a	except ZeroDivisionError:
688	n/a	pass
689	n/a	else:
690	n/a	raise Exception("exception not raised")
691	n/a
692	n/a	server = BaseServer(self.family, self.addr)
693	n/a	client = TestClient(self.family, server.address)
694	n/a	self.loop_waiting_for_flag(client)
695	n/a
696	n/a	def test_connection_attributes(self):
697	n/a	server = BaseServer(self.family, self.addr)
698	n/a	client = BaseClient(self.family, server.address)
699	n/a
700	n/a	# we start disconnected
701	n/a	self.assertFalse(server.connected)
702	n/a	self.assertTrue(server.accepting)
703	n/a	# this can't be taken for granted across all platforms
704	n/a	#self.assertFalse(client.connected)
705	n/a	self.assertFalse(client.accepting)
706	n/a
707	n/a	# execute some loops so that client connects to server
708	n/a	asyncore.loop(timeout=0.01, use_poll=self.use_poll, count=100)
709	n/a	self.assertFalse(server.connected)
710	n/a	self.assertTrue(server.accepting)
711	n/a	self.assertTrue(client.connected)
712	n/a	self.assertFalse(client.accepting)
713	n/a
714	n/a	# disconnect the client
715	n/a	client.close()
716	n/a	self.assertFalse(server.connected)
717	n/a	self.assertTrue(server.accepting)
718	n/a	self.assertFalse(client.connected)
719	n/a	self.assertFalse(client.accepting)
720	n/a
721	n/a	# stop serving
722	n/a	server.close()
723	n/a	self.assertFalse(server.connected)
724	n/a	self.assertFalse(server.accepting)
725	n/a
726	n/a	def test_create_socket(self):
727	n/a	s = asyncore.dispatcher()
728	n/a	s.create_socket(self.family)
729	n/a	self.assertEqual(s.socket.family, self.family)
730	n/a	SOCK_NONBLOCK = getattr(socket, 'SOCK_NONBLOCK', 0)
731	n/a	sock_type = socket.SOCK_STREAM \| SOCK_NONBLOCK
732	n/a	if hasattr(socket, 'SOCK_CLOEXEC'):
733	n/a	self.assertIn(s.socket.type,
734	n/a	(sock_type \| socket.SOCK_CLOEXEC, sock_type))
735	n/a	else:
736	n/a	self.assertEqual(s.socket.type, sock_type)
737	n/a
738	n/a	def test_bind(self):
739	n/a	if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX:
740	n/a	self.skipTest("Not applicable to AF_UNIX sockets.")
741	n/a	s1 = asyncore.dispatcher()
742	n/a	s1.create_socket(self.family)
743	n/a	s1.bind(self.addr)
744	n/a	s1.listen(5)
745	n/a	port = s1.socket.getsockname()[1]
746	n/a
747	n/a	s2 = asyncore.dispatcher()
748	n/a	s2.create_socket(self.family)
749	n/a	# EADDRINUSE indicates the socket was correctly bound
750	n/a	self.assertRaises(OSError, s2.bind, (self.addr[0], port))
751	n/a
752	n/a	def test_set_reuse_addr(self):
753	n/a	if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX:
754	n/a	self.skipTest("Not applicable to AF_UNIX sockets.")
755	n/a	sock = socket.socket(self.family)
756	n/a	try:
757	n/a	sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
758	n/a	except OSError:
759	n/a	unittest.skip("SO_REUSEADDR not supported on this platform")
760	n/a	else:
761	n/a	# if SO_REUSEADDR succeeded for sock we expect asyncore
762	n/a	# to do the same
763	n/a	s = asyncore.dispatcher(socket.socket(self.family))
764	n/a	self.assertFalse(s.socket.getsockopt(socket.SOL_SOCKET,
765	n/a	socket.SO_REUSEADDR))
766	n/a	s.socket.close()
767	n/a	s.create_socket(self.family)
768	n/a	s.set_reuse_addr()
769	n/a	self.assertTrue(s.socket.getsockopt(socket.SOL_SOCKET,
770	n/a	socket.SO_REUSEADDR))
771	n/a	finally:
772	n/a	sock.close()
773	n/a
774	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
775	n/a	@support.reap_threads
776	n/a	def test_quick_connect(self):
777	n/a	# see: http://bugs.python.org/issue10340
778	n/a	if self.family in (socket.AF_INET, getattr(socket, "AF_INET6", object())):
779	n/a	server = BaseServer(self.family, self.addr)
780	n/a	t = threading.Thread(target=lambda: asyncore.loop(timeout=0.1,
781	n/a	count=500))
782	n/a	t.start()
783	n/a	def cleanup():
784	n/a	t.join(timeout=TIMEOUT)
785	n/a	if t.is_alive():
786	n/a	self.fail("join() timed out")
787	n/a	self.addCleanup(cleanup)
788	n/a
789	n/a	s = socket.socket(self.family, socket.SOCK_STREAM)
790	n/a	s.settimeout(.2)
791	n/a	s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
792	n/a	struct.pack('ii', 1, 0))
793	n/a	try:
794	n/a	s.connect(server.address)
795	n/a	except OSError:
796	n/a	pass
797	n/a	finally:
798	n/a	s.close()
799	n/a
800	n/a	class TestAPI_UseIPv4Sockets(BaseTestAPI):
801	n/a	family = socket.AF_INET
802	n/a	addr = (support.HOST, 0)
803	n/a
804	n/a	@unittest.skipUnless(support.IPV6_ENABLED, 'IPv6 support required')
805	n/a	class TestAPI_UseIPv6Sockets(BaseTestAPI):
806	n/a	family = socket.AF_INET6
807	n/a	addr = (support.HOSTv6, 0)
808	n/a
809	n/a	@unittest.skipUnless(HAS_UNIX_SOCKETS, 'Unix sockets required')
810	n/a	class TestAPI_UseUnixSockets(BaseTestAPI):
811	n/a	if HAS_UNIX_SOCKETS:
812	n/a	family = socket.AF_UNIX
813	n/a	addr = support.TESTFN
814	n/a
815	n/a	def tearDown(self):
816	n/a	support.unlink(self.addr)
817	n/a	BaseTestAPI.tearDown(self)
818	n/a
819	n/a	class TestAPI_UseIPv4Select(TestAPI_UseIPv4Sockets, unittest.TestCase):
820	n/a	use_poll = False
821	n/a
822	n/a	@unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
823	n/a	class TestAPI_UseIPv4Poll(TestAPI_UseIPv4Sockets, unittest.TestCase):
824	n/a	use_poll = True
825	n/a
826	n/a	class TestAPI_UseIPv6Select(TestAPI_UseIPv6Sockets, unittest.TestCase):
827	n/a	use_poll = False
828	n/a
829	n/a	@unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
830	n/a	class TestAPI_UseIPv6Poll(TestAPI_UseIPv6Sockets, unittest.TestCase):
831	n/a	use_poll = True
832	n/a
833	n/a	class TestAPI_UseUnixSocketsSelect(TestAPI_UseUnixSockets, unittest.TestCase):
834	n/a	use_poll = False
835	n/a
836	n/a	@unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
837	n/a	class TestAPI_UseUnixSocketsPoll(TestAPI_UseUnixSockets, unittest.TestCase):
838	n/a	use_poll = True
839	n/a
840	n/a	if __name__ == "__main__":
841	n/a	unittest.main()