Python code coverage for Lib/test/test_poplib.py

#	count	content
1	n/a	"""Test script for poplib module."""
2	n/a
3	n/a	# Modified by Giampaolo Rodola' to give poplib.POP3 and poplib.POP3_SSL
4	n/a	# a real test suite
5	n/a
6	n/a	import poplib
7	n/a	import asyncore
8	n/a	import asynchat
9	n/a	import socket
10	n/a	import os
11	n/a	import errno
12	n/a
13	n/a	from unittest import TestCase, skipUnless
14	n/a	from test import support as test_support
15	n/a	threading = test_support.import_module('threading')
16	n/a
17	n/a	HOST = test_support.HOST
18	n/a	PORT = 0
19	n/a
20	n/a	SUPPORTS_SSL = False
21	n/a	if hasattr(poplib, 'POP3_SSL'):
22	n/a	import ssl
23	n/a
24	n/a	SUPPORTS_SSL = True
25	n/a	CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "keycert3.pem")
26	n/a	CAFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "pycacert.pem")
27	n/a
28	n/a	requires_ssl = skipUnless(SUPPORTS_SSL, 'SSL not supported')
29	n/a
30	n/a	# the dummy data returned by server when LIST and RETR commands are issued
31	n/a	LIST_RESP = b'1 1\r\n2 2\r\n3 3\r\n4 4\r\n5 5\r\n.\r\n'
32	n/a	RETR_RESP = b"""From: postmaster@python.org\
33	n/a	\r\nContent-Type: text/plain\r\n\
34	n/a	MIME-Version: 1.0\r\n\
35	n/a	Subject: Dummy\r\n\
36	n/a	\r\n\
37	n/a	line1\r\n\
38	n/a	line2\r\n\
39	n/a	line3\r\n\
40	n/a	.\r\n"""
41	n/a
42	n/a
43	n/a	class DummyPOP3Handler(asynchat.async_chat):
44	n/a
45	n/a	CAPAS = {'UIDL': [], 'IMPLEMENTATION': ['python-testlib-pop-server']}
46	n/a	enable_UTF8 = False
47	n/a
48	n/a	def __init__(self, conn):
49	n/a	asynchat.async_chat.__init__(self, conn)
50	n/a	self.set_terminator(b"\r\n")
51	n/a	self.in_buffer = []
52	n/a	self.push('+OK dummy pop3 server ready. <timestamp>')
53	n/a	self.tls_active = False
54	n/a	self.tls_starting = False
55	n/a
56	n/a	def collect_incoming_data(self, data):
57	n/a	self.in_buffer.append(data)
58	n/a
59	n/a	def found_terminator(self):
60	n/a	line = b''.join(self.in_buffer)
61	n/a	line = str(line, 'ISO-8859-1')
62	n/a	self.in_buffer = []
63	n/a	cmd = line.split(' ')[0].lower()
64	n/a	space = line.find(' ')
65	n/a	if space != -1:
66	n/a	arg = line[space + 1:]
67	n/a	else:
68	n/a	arg = ""
69	n/a	if hasattr(self, 'cmd_' + cmd):
70	n/a	method = getattr(self, 'cmd_' + cmd)
71	n/a	method(arg)
72	n/a	else:
73	n/a	self.push('-ERR unrecognized POP3 command "%s".' %cmd)
74	n/a
75	n/a	def handle_error(self):
76	n/a	raise
77	n/a
78	n/a	def push(self, data):
79	n/a	asynchat.async_chat.push(self, data.encode("ISO-8859-1") + b'\r\n')
80	n/a
81	n/a	def cmd_echo(self, arg):
82	n/a	# sends back the received string (used by the test suite)
83	n/a	self.push(arg)
84	n/a
85	n/a	def cmd_user(self, arg):
86	n/a	if arg != "guido":
87	n/a	self.push("-ERR no such user")
88	n/a	self.push('+OK password required')
89	n/a
90	n/a	def cmd_pass(self, arg):
91	n/a	if arg != "python":
92	n/a	self.push("-ERR wrong password")
93	n/a	self.push('+OK 10 messages')
94	n/a
95	n/a	def cmd_stat(self, arg):
96	n/a	self.push('+OK 10 100')
97	n/a
98	n/a	def cmd_list(self, arg):
99	n/a	if arg:
100	n/a	self.push('+OK %s %s' % (arg, arg))
101	n/a	else:
102	n/a	self.push('+OK')
103	n/a	asynchat.async_chat.push(self, LIST_RESP)
104	n/a
105	n/a	cmd_uidl = cmd_list
106	n/a
107	n/a	def cmd_retr(self, arg):
108	n/a	self.push('+OK %s bytes' %len(RETR_RESP))
109	n/a	asynchat.async_chat.push(self, RETR_RESP)
110	n/a
111	n/a	cmd_top = cmd_retr
112	n/a
113	n/a	def cmd_dele(self, arg):
114	n/a	self.push('+OK message marked for deletion.')
115	n/a
116	n/a	def cmd_noop(self, arg):
117	n/a	self.push('+OK done nothing.')
118	n/a
119	n/a	def cmd_rpop(self, arg):
120	n/a	self.push('+OK done nothing.')
121	n/a
122	n/a	def cmd_apop(self, arg):
123	n/a	self.push('+OK done nothing.')
124	n/a
125	n/a	def cmd_quit(self, arg):
126	n/a	self.push('+OK closing.')
127	n/a	self.close_when_done()
128	n/a
129	n/a	def _get_capas(self):
130	n/a	_capas = dict(self.CAPAS)
131	n/a	if not self.tls_active and SUPPORTS_SSL:
132	n/a	_capas['STLS'] = []
133	n/a	return _capas
134	n/a
135	n/a	def cmd_capa(self, arg):
136	n/a	self.push('+OK Capability list follows')
137	n/a	if self._get_capas():
138	n/a	for cap, params in self._get_capas().items():
139	n/a	_ln = [cap]
140	n/a	if params:
141	n/a	_ln.extend(params)
142	n/a	self.push(' '.join(_ln))
143	n/a	self.push('.')
144	n/a
145	n/a	def cmd_utf8(self, arg):
146	n/a	self.push('+OK I know RFC6856'
147	n/a	if self.enable_UTF8
148	n/a	else '-ERR What is UTF8?!')
149	n/a
150	n/a	if SUPPORTS_SSL:
151	n/a
152	n/a	def cmd_stls(self, arg):
153	n/a	if self.tls_active is False:
154	n/a	self.push('+OK Begin TLS negotiation')
155	n/a	context = ssl.SSLContext()
156	n/a	context.load_cert_chain(CERTFILE)
157	n/a	tls_sock = context.wrap_socket(self.socket,
158	n/a	server_side=True,
159	n/a	do_handshake_on_connect=False,
160	n/a	suppress_ragged_eofs=False)
161	n/a	self.del_channel()
162	n/a	self.set_socket(tls_sock)
163	n/a	self.tls_active = True
164	n/a	self.tls_starting = True
165	n/a	self.in_buffer = []
166	n/a	self._do_tls_handshake()
167	n/a	else:
168	n/a	self.push('-ERR Command not permitted when TLS active')
169	n/a
170	n/a	def _do_tls_handshake(self):
171	n/a	try:
172	n/a	self.socket.do_handshake()
173	n/a	except ssl.SSLError as err:
174	n/a	if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
175	n/a	ssl.SSL_ERROR_WANT_WRITE):
176	n/a	return
177	n/a	elif err.args[0] == ssl.SSL_ERROR_EOF:
178	n/a	return self.handle_close()
179	n/a	raise
180	n/a	except OSError as err:
181	n/a	if err.args[0] == errno.ECONNABORTED:
182	n/a	return self.handle_close()
183	n/a	else:
184	n/a	self.tls_active = True
185	n/a	self.tls_starting = False
186	n/a
187	n/a	def handle_read(self):
188	n/a	if self.tls_starting:
189	n/a	self._do_tls_handshake()
190	n/a	else:
191	n/a	try:
192	n/a	asynchat.async_chat.handle_read(self)
193	n/a	except ssl.SSLEOFError:
194	n/a	self.handle_close()
195	n/a
196	n/a	class DummyPOP3Server(asyncore.dispatcher, threading.Thread):
197	n/a
198	n/a	handler = DummyPOP3Handler
199	n/a
200	n/a	def __init__(self, address, af=socket.AF_INET):
201	n/a	threading.Thread.__init__(self)
202	n/a	asyncore.dispatcher.__init__(self)
203	n/a	self.create_socket(af, socket.SOCK_STREAM)
204	n/a	self.bind(address)
205	n/a	self.listen(5)
206	n/a	self.active = False
207	n/a	self.active_lock = threading.Lock()
208	n/a	self.host, self.port = self.socket.getsockname()[:2]
209	n/a	self.handler_instance = None
210	n/a
211	n/a	def start(self):
212	n/a	assert not self.active
213	n/a	self.__flag = threading.Event()
214	n/a	threading.Thread.start(self)
215	n/a	self.__flag.wait()
216	n/a
217	n/a	def run(self):
218	n/a	self.active = True
219	n/a	self.__flag.set()
220	n/a	while self.active and asyncore.socket_map:
221	n/a	self.active_lock.acquire()
222	n/a	asyncore.loop(timeout=0.1, count=1)
223	n/a	self.active_lock.release()
224	n/a	asyncore.close_all(ignore_all=True)
225	n/a
226	n/a	def stop(self):
227	n/a	assert self.active
228	n/a	self.active = False
229	n/a	self.join()
230	n/a
231	n/a	def handle_accepted(self, conn, addr):
232	n/a	self.handler_instance = self.handler(conn)
233	n/a
234	n/a	def handle_connect(self):
235	n/a	self.close()
236	n/a	handle_read = handle_connect
237	n/a
238	n/a	def writable(self):
239	n/a	return 0
240	n/a
241	n/a	def handle_error(self):
242	n/a	raise
243	n/a
244	n/a
245	n/a	class TestPOP3Class(TestCase):
246	n/a	def assertOK(self, resp):
247	n/a	self.assertTrue(resp.startswith(b"+OK"))
248	n/a
249	n/a	def setUp(self):
250	n/a	self.server = DummyPOP3Server((HOST, PORT))
251	n/a	self.server.start()
252	n/a	self.client = poplib.POP3(self.server.host, self.server.port, timeout=3)
253	n/a
254	n/a	def tearDown(self):
255	n/a	self.client.close()
256	n/a	self.server.stop()
257	n/a
258	n/a	def test_getwelcome(self):
259	n/a	self.assertEqual(self.client.getwelcome(),
260	n/a	b'+OK dummy pop3 server ready. <timestamp>')
261	n/a
262	n/a	def test_exceptions(self):
263	n/a	self.assertRaises(poplib.error_proto, self.client._shortcmd, 'echo -err')
264	n/a
265	n/a	def test_user(self):
266	n/a	self.assertOK(self.client.user('guido'))
267	n/a	self.assertRaises(poplib.error_proto, self.client.user, 'invalid')
268	n/a
269	n/a	def test_pass_(self):
270	n/a	self.assertOK(self.client.pass_('python'))
271	n/a	self.assertRaises(poplib.error_proto, self.client.user, 'invalid')
272	n/a
273	n/a	def test_stat(self):
274	n/a	self.assertEqual(self.client.stat(), (10, 100))
275	n/a
276	n/a	def test_list(self):
277	n/a	self.assertEqual(self.client.list()[1:],
278	n/a	([b'1 1', b'2 2', b'3 3', b'4 4', b'5 5'],
279	n/a	25))
280	n/a	self.assertTrue(self.client.list('1').endswith(b"OK 1 1"))
281	n/a
282	n/a	def test_retr(self):
283	n/a	expected = (b'+OK 116 bytes',
284	n/a	[b'From: postmaster@python.org', b'Content-Type: text/plain',
285	n/a	b'MIME-Version: 1.0', b'Subject: Dummy',
286	n/a	b'', b'line1', b'line2', b'line3'],
287	n/a	113)
288	n/a	foo = self.client.retr('foo')
289	n/a	self.assertEqual(foo, expected)
290	n/a
291	n/a	def test_too_long_lines(self):
292	n/a	self.assertRaises(poplib.error_proto, self.client._shortcmd,
293	n/a	'echo +%s' % ((poplib._MAXLINE + 10) * 'a'))
294	n/a
295	n/a	def test_dele(self):
296	n/a	self.assertOK(self.client.dele('foo'))
297	n/a
298	n/a	def test_noop(self):
299	n/a	self.assertOK(self.client.noop())
300	n/a
301	n/a	def test_rpop(self):
302	n/a	self.assertOK(self.client.rpop('foo'))
303	n/a
304	n/a	def test_apop(self):
305	n/a	self.assertOK(self.client.apop('foo', 'dummypassword'))
306	n/a
307	n/a	def test_top(self):
308	n/a	expected = (b'+OK 116 bytes',
309	n/a	[b'From: postmaster@python.org', b'Content-Type: text/plain',
310	n/a	b'MIME-Version: 1.0', b'Subject: Dummy', b'',
311	n/a	b'line1', b'line2', b'line3'],
312	n/a	113)
313	n/a	self.assertEqual(self.client.top(1, 1), expected)
314	n/a
315	n/a	def test_uidl(self):
316	n/a	self.client.uidl()
317	n/a	self.client.uidl('foo')
318	n/a
319	n/a	def test_utf8_raises_if_unsupported(self):
320	n/a	self.server.handler.enable_UTF8 = False
321	n/a	self.assertRaises(poplib.error_proto, self.client.utf8)
322	n/a
323	n/a	def test_utf8(self):
324	n/a	self.server.handler.enable_UTF8 = True
325	n/a	expected = b'+OK I know RFC6856'
326	n/a	result = self.client.utf8()
327	n/a	self.assertEqual(result, expected)
328	n/a
329	n/a	def test_capa(self):
330	n/a	capa = self.client.capa()
331	n/a	self.assertTrue('IMPLEMENTATION' in capa.keys())
332	n/a
333	n/a	def test_quit(self):
334	n/a	resp = self.client.quit()
335	n/a	self.assertTrue(resp)
336	n/a	self.assertIsNone(self.client.sock)
337	n/a	self.assertIsNone(self.client.file)
338	n/a
339	n/a	@requires_ssl
340	n/a	def test_stls_capa(self):
341	n/a	capa = self.client.capa()
342	n/a	self.assertTrue('STLS' in capa.keys())
343	n/a
344	n/a	@requires_ssl
345	n/a	def test_stls(self):
346	n/a	expected = b'+OK Begin TLS negotiation'
347	n/a	resp = self.client.stls()
348	n/a	self.assertEqual(resp, expected)
349	n/a
350	n/a	@requires_ssl
351	n/a	def test_stls_context(self):
352	n/a	expected = b'+OK Begin TLS negotiation'
353	n/a	ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
354	n/a	ctx.load_verify_locations(CAFILE)
355	n/a	ctx.verify_mode = ssl.CERT_REQUIRED
356	n/a	ctx.check_hostname = True
357	n/a	with self.assertRaises(ssl.CertificateError):
358	n/a	resp = self.client.stls(context=ctx)
359	n/a	self.client = poplib.POP3("localhost", self.server.port, timeout=3)
360	n/a	resp = self.client.stls(context=ctx)
361	n/a	self.assertEqual(resp, expected)
362	n/a
363	n/a
364	n/a	if SUPPORTS_SSL:
365	n/a	from test.test_ftplib import SSLConnection
366	n/a
367	n/a	class DummyPOP3_SSLHandler(SSLConnection, DummyPOP3Handler):
368	n/a
369	n/a	def __init__(self, conn):
370	n/a	asynchat.async_chat.__init__(self, conn)
371	n/a	self.secure_connection()
372	n/a	self.set_terminator(b"\r\n")
373	n/a	self.in_buffer = []
374	n/a	self.push('+OK dummy pop3 server ready. <timestamp>')
375	n/a	self.tls_active = True
376	n/a	self.tls_starting = False
377	n/a
378	n/a
379	n/a	@requires_ssl
380	n/a	class TestPOP3_SSLClass(TestPOP3Class):
381	n/a	# repeat previous tests by using poplib.POP3_SSL
382	n/a
383	n/a	def setUp(self):
384	n/a	self.server = DummyPOP3Server((HOST, PORT))
385	n/a	self.server.handler = DummyPOP3_SSLHandler
386	n/a	self.server.start()
387	n/a	self.client = poplib.POP3_SSL(self.server.host, self.server.port)
388	n/a
389	n/a	def test__all__(self):
390	n/a	self.assertIn('POP3_SSL', poplib.__all__)
391	n/a
392	n/a	def test_context(self):
393	n/a	ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
394	n/a	self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
395	n/a	self.server.port, keyfile=CERTFILE, context=ctx)
396	n/a	self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
397	n/a	self.server.port, certfile=CERTFILE, context=ctx)
398	n/a	self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
399	n/a	self.server.port, keyfile=CERTFILE,
400	n/a	certfile=CERTFILE, context=ctx)
401	n/a
402	n/a	self.client.quit()
403	n/a	self.client = poplib.POP3_SSL(self.server.host, self.server.port,
404	n/a	context=ctx)
405	n/a	self.assertIsInstance(self.client.sock, ssl.SSLSocket)
406	n/a	self.assertIs(self.client.sock.context, ctx)
407	n/a	self.assertTrue(self.client.noop().startswith(b'+OK'))
408	n/a
409	n/a	def test_stls(self):
410	n/a	self.assertRaises(poplib.error_proto, self.client.stls)
411	n/a
412	n/a	test_stls_context = test_stls
413	n/a
414	n/a	def test_stls_capa(self):
415	n/a	capa = self.client.capa()
416	n/a	self.assertFalse('STLS' in capa.keys())
417	n/a
418	n/a
419	n/a	@requires_ssl
420	n/a	class TestPOP3_TLSClass(TestPOP3Class):
421	n/a	# repeat previous tests by using poplib.POP3.stls()
422	n/a
423	n/a	def setUp(self):
424	n/a	self.server = DummyPOP3Server((HOST, PORT))
425	n/a	self.server.start()
426	n/a	self.client = poplib.POP3(self.server.host, self.server.port, timeout=3)
427	n/a	self.client.stls()
428	n/a
429	n/a	def tearDown(self):
430	n/a	if self.client.file is not None and self.client.sock is not None:
431	n/a	try:
432	n/a	self.client.quit()
433	n/a	except poplib.error_proto:
434	n/a	# happens in the test_too_long_lines case; the overlong
435	n/a	# response will be treated as response to QUIT and raise
436	n/a	# this exception
437	n/a	self.client.close()
438	n/a	self.server.stop()
439	n/a
440	n/a	def test_stls(self):
441	n/a	self.assertRaises(poplib.error_proto, self.client.stls)
442	n/a
443	n/a	test_stls_context = test_stls
444	n/a
445	n/a	def test_stls_capa(self):
446	n/a	capa = self.client.capa()
447	n/a	self.assertFalse(b'STLS' in capa.keys())
448	n/a
449	n/a
450	n/a	class TestTimeouts(TestCase):
451	n/a
452	n/a	def setUp(self):
453	n/a	self.evt = threading.Event()
454	n/a	self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
455	n/a	self.sock.settimeout(60) # Safety net. Look issue 11812
456	n/a	self.port = test_support.bind_port(self.sock)
457	n/a	self.thread = threading.Thread(target=self.server, args=(self.evt,self.sock))
458	n/a	self.thread.setDaemon(True)
459	n/a	self.thread.start()
460	n/a	self.evt.wait()
461	n/a
462	n/a	def tearDown(self):
463	n/a	self.thread.join()
464	n/a	del self.thread # Clear out any dangling Thread objects.
465	n/a
466	n/a	def server(self, evt, serv):
467	n/a	serv.listen()
468	n/a	evt.set()
469	n/a	try:
470	n/a	conn, addr = serv.accept()
471	n/a	conn.send(b"+ Hola mundo\n")
472	n/a	conn.close()
473	n/a	except socket.timeout:
474	n/a	pass
475	n/a	finally:
476	n/a	serv.close()
477	n/a
478	n/a	def testTimeoutDefault(self):
479	n/a	self.assertIsNone(socket.getdefaulttimeout())
480	n/a	socket.setdefaulttimeout(30)
481	n/a	try:
482	n/a	pop = poplib.POP3(HOST, self.port)
483	n/a	finally:
484	n/a	socket.setdefaulttimeout(None)
485	n/a	self.assertEqual(pop.sock.gettimeout(), 30)
486	n/a	pop.sock.close()
487	n/a
488	n/a	def testTimeoutNone(self):
489	n/a	self.assertIsNone(socket.getdefaulttimeout())
490	n/a	socket.setdefaulttimeout(30)
491	n/a	try:
492	n/a	pop = poplib.POP3(HOST, self.port, timeout=None)
493	n/a	finally:
494	n/a	socket.setdefaulttimeout(None)
495	n/a	self.assertIsNone(pop.sock.gettimeout())
496	n/a	pop.sock.close()
497	n/a
498	n/a	def testTimeoutValue(self):
499	n/a	pop = poplib.POP3(HOST, self.port, timeout=30)
500	n/a	self.assertEqual(pop.sock.gettimeout(), 30)
501	n/a	pop.sock.close()
502	n/a
503	n/a
504	n/a	def test_main():
505	n/a	tests = [TestPOP3Class, TestTimeouts,
506	n/a	TestPOP3_SSLClass, TestPOP3_TLSClass]
507	n/a	thread_info = test_support.threading_setup()
508	n/a	try:
509	n/a	test_support.run_unittest(*tests)
510	n/a	finally:
511	n/a	test_support.threading_cleanup(*thread_info)
512	n/a
513	n/a
514	n/a	if __name__ == '__main__':
515	n/a	test_main()