blob: c0929a06b3a83d583279f05d1e3b39f0a79e1036 [file] [log] [blame]
Christian Heimesd3956292008-11-05 19:48:27 +00001"""Test script for poplib module."""
2
3# Modified by Giampaolo Rodola' to give poplib.POP3 and poplib.POP3_SSL
4# a real test suite
5
Guido van Rossumd8faa362007-04-27 19:54:29 +00006import poplib
Christian Heimesd3956292008-11-05 19:48:27 +00007import asyncore
8import asynchat
9import socket
10import os
Guido van Rossumd8faa362007-04-27 19:54:29 +000011import time
Antoine Pitroud3f8ab82010-04-24 21:26:44 +000012import errno
Guido van Rossumd8faa362007-04-27 19:54:29 +000013
14from unittest import TestCase
Christian Heimesd3956292008-11-05 19:48:27 +000015from test import support as test_support
Victor Stinner45df8202010-04-28 22:31:17 +000016threading = test_support.import_module('threading')
Guido van Rossumd8faa362007-04-27 19:54:29 +000017
Christian Heimesd3956292008-11-05 19:48:27 +000018HOST = test_support.HOST
19PORT = 0
Guido van Rossumd8faa362007-04-27 19:54:29 +000020
Christian Heimesd3956292008-11-05 19:48:27 +000021# the dummy data returned by server when LIST and RETR commands are issued
22LIST_RESP = b'1 1\r\n2 2\r\n3 3\r\n4 4\r\n5 5\r\n.\r\n'
23RETR_RESP = b"""From: postmaster@python.org\
24\r\nContent-Type: text/plain\r\n\
25MIME-Version: 1.0\r\n\
26Subject: Dummy\r\n\
27\r\n\
28line1\r\n\
29line2\r\n\
30line3\r\n\
31.\r\n"""
Guido van Rossumd8faa362007-04-27 19:54:29 +000032
Christian Heimesd3956292008-11-05 19:48:27 +000033
34class DummyPOP3Handler(asynchat.async_chat):
35
36 def __init__(self, conn):
37 asynchat.async_chat.__init__(self, conn)
38 self.set_terminator(b"\r\n")
39 self.in_buffer = []
Mark Dickinsonea1158f2009-08-06 16:06:25 +000040 self.push('+OK dummy pop3 server ready. <timestamp>')
Christian Heimesd3956292008-11-05 19:48:27 +000041
42 def collect_incoming_data(self, data):
43 self.in_buffer.append(data)
44
45 def found_terminator(self):
46 line = b''.join(self.in_buffer)
47 line = str(line, 'ISO-8859-1')
48 self.in_buffer = []
49 cmd = line.split(' ')[0].lower()
50 space = line.find(' ')
51 if space != -1:
52 arg = line[space + 1:]
53 else:
54 arg = ""
55 if hasattr(self, 'cmd_' + cmd):
56 method = getattr(self, 'cmd_' + cmd)
57 method(arg)
58 else:
59 self.push('-ERR unrecognized POP3 command "%s".' %cmd)
60
61 def handle_error(self):
62 raise
63
64 def push(self, data):
65 asynchat.async_chat.push(self, data.encode("ISO-8859-1") + b'\r\n')
66
67 def cmd_echo(self, arg):
68 # sends back the received string (used by the test suite)
69 self.push(arg)
70
71 def cmd_user(self, arg):
72 if arg != "guido":
73 self.push("-ERR no such user")
74 self.push('+OK password required')
75
76 def cmd_pass(self, arg):
77 if arg != "python":
78 self.push("-ERR wrong password")
79 self.push('+OK 10 messages')
80
81 def cmd_stat(self, arg):
82 self.push('+OK 10 100')
83
84 def cmd_list(self, arg):
85 if arg:
86 self.push('+OK %s %s' %(arg, arg))
87 else:
88 self.push('+OK')
89 asynchat.async_chat.push(self, LIST_RESP)
90
91 cmd_uidl = cmd_list
92
93 def cmd_retr(self, arg):
94 self.push('+OK %s bytes' %len(RETR_RESP))
95 asynchat.async_chat.push(self, RETR_RESP)
96
97 cmd_top = cmd_retr
98
99 def cmd_dele(self, arg):
100 self.push('+OK message marked for deletion.')
101
102 def cmd_noop(self, arg):
103 self.push('+OK done nothing.')
104
105 def cmd_rpop(self, arg):
106 self.push('+OK done nothing.')
107
Mark Dickinsonea1158f2009-08-06 16:06:25 +0000108 def cmd_apop(self, arg):
109 self.push('+OK done nothing.')
110
Giampaolo Rodolà95bcb932011-02-25 22:28:24 +0000111 def cmd_quit(self, arg):
112 self.push('+OK closing.')
113 self.close_when_done()
114
Christian Heimesd3956292008-11-05 19:48:27 +0000115
116class DummyPOP3Server(asyncore.dispatcher, threading.Thread):
117
118 handler = DummyPOP3Handler
119
120 def __init__(self, address, af=socket.AF_INET):
121 threading.Thread.__init__(self)
122 asyncore.dispatcher.__init__(self)
123 self.create_socket(af, socket.SOCK_STREAM)
124 self.bind(address)
125 self.listen(5)
126 self.active = False
127 self.active_lock = threading.Lock()
128 self.host, self.port = self.socket.getsockname()[:2]
Giampaolo Rodolà42382fe2010-08-17 16:09:53 +0000129 self.handler_instance = None
Christian Heimesd3956292008-11-05 19:48:27 +0000130
131 def start(self):
132 assert not self.active
133 self.__flag = threading.Event()
134 threading.Thread.start(self)
135 self.__flag.wait()
136
137 def run(self):
138 self.active = True
139 self.__flag.set()
140 while self.active and asyncore.socket_map:
141 self.active_lock.acquire()
142 asyncore.loop(timeout=0.1, count=1)
143 self.active_lock.release()
144 asyncore.close_all(ignore_all=True)
145
146 def stop(self):
147 assert self.active
148 self.active = False
149 self.join()
150
Giampaolo Rodolà977c7072010-10-04 21:08:36 +0000151 def handle_accepted(self, conn, addr):
Giampaolo Rodolà42382fe2010-08-17 16:09:53 +0000152 self.handler_instance = self.handler(conn)
Christian Heimesd3956292008-11-05 19:48:27 +0000153
154 def handle_connect(self):
155 self.close()
156 handle_read = handle_connect
157
158 def writable(self):
159 return 0
160
161 def handle_error(self):
162 raise
163
164
165class TestPOP3Class(TestCase):
166 def assertOK(self, resp):
167 self.assertTrue(resp.startswith(b"+OK"))
168
169 def setUp(self):
170 self.server = DummyPOP3Server((HOST, PORT))
171 self.server.start()
Giampaolo Rodolà95bcb932011-02-25 22:28:24 +0000172 self.client = poplib.POP3(self.server.host, self.server.port, timeout=3)
Christian Heimesd3956292008-11-05 19:48:27 +0000173
174 def tearDown(self):
Giampaolo Rodolà95bcb932011-02-25 22:28:24 +0000175 self.client.close()
Christian Heimesd3956292008-11-05 19:48:27 +0000176 self.server.stop()
177
178 def test_getwelcome(self):
Mark Dickinsonea1158f2009-08-06 16:06:25 +0000179 self.assertEqual(self.client.getwelcome(),
180 b'+OK dummy pop3 server ready. <timestamp>')
Christian Heimesd3956292008-11-05 19:48:27 +0000181
182 def test_exceptions(self):
183 self.assertRaises(poplib.error_proto, self.client._shortcmd, 'echo -err')
184
185 def test_user(self):
186 self.assertOK(self.client.user('guido'))
187 self.assertRaises(poplib.error_proto, self.client.user, 'invalid')
188
189 def test_pass_(self):
190 self.assertOK(self.client.pass_('python'))
191 self.assertRaises(poplib.error_proto, self.client.user, 'invalid')
192
193 def test_stat(self):
194 self.assertEqual(self.client.stat(), (10, 100))
195
196 def test_list(self):
197 self.assertEqual(self.client.list()[1:],
198 ([b'1 1', b'2 2', b'3 3', b'4 4', b'5 5'],
199 25))
200 self.assertTrue(self.client.list('1').endswith(b"OK 1 1"))
201
202 def test_retr(self):
203 expected = (b'+OK 116 bytes',
204 [b'From: postmaster@python.org', b'Content-Type: text/plain',
205 b'MIME-Version: 1.0', b'Subject: Dummy',
206 b'', b'line1', b'line2', b'line3'],
207 113)
208 foo = self.client.retr('foo')
209 self.assertEqual(foo, expected)
210
211 def test_dele(self):
212 self.assertOK(self.client.dele('foo'))
213
214 def test_noop(self):
215 self.assertOK(self.client.noop())
216
217 def test_rpop(self):
218 self.assertOK(self.client.rpop('foo'))
219
Mark Dickinsonea1158f2009-08-06 16:06:25 +0000220 def test_apop(self):
221 self.assertOK(self.client.apop('foo', 'dummypassword'))
222
Christian Heimesd3956292008-11-05 19:48:27 +0000223 def test_top(self):
224 expected = (b'+OK 116 bytes',
225 [b'From: postmaster@python.org', b'Content-Type: text/plain',
226 b'MIME-Version: 1.0', b'Subject: Dummy', b'',
227 b'line1', b'line2', b'line3'],
228 113)
229 self.assertEqual(self.client.top(1, 1), expected)
230
231 def test_uidl(self):
232 self.client.uidl()
233 self.client.uidl('foo')
234
Giampaolo Rodolà95bcb932011-02-25 22:28:24 +0000235 def test_quit(self):
236 resp = self.client.quit()
237 self.assertTrue(resp)
238 self.assertIsNone(self.client.sock)
239 self.assertIsNone(self.client.file)
240
Christian Heimesd3956292008-11-05 19:48:27 +0000241
242SUPPORTS_SSL = False
243if hasattr(poplib, 'POP3_SSL'):
244 import ssl
245
246 SUPPORTS_SSL = True
247 CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "keycert.pem")
248
249 class DummyPOP3_SSLHandler(DummyPOP3Handler):
250
251 def __init__(self, conn):
252 asynchat.async_chat.__init__(self, conn)
253 ssl_socket = ssl.wrap_socket(self.socket, certfile=CERTFILE,
Antoine Pitroud3f8ab82010-04-24 21:26:44 +0000254 server_side=True,
255 do_handshake_on_connect=False)
Christian Heimesd3956292008-11-05 19:48:27 +0000256 self.del_channel()
257 self.set_socket(ssl_socket)
Antoine Pitroud3f8ab82010-04-24 21:26:44 +0000258 # Must try handshake before calling push()
259 self._ssl_accepting = True
260 self._do_ssl_handshake()
Christian Heimesd3956292008-11-05 19:48:27 +0000261 self.set_terminator(b"\r\n")
262 self.in_buffer = []
Mark Dickinsonea1158f2009-08-06 16:06:25 +0000263 self.push('+OK dummy pop3 server ready. <timestamp>')
Christian Heimesd3956292008-11-05 19:48:27 +0000264
Antoine Pitroud3f8ab82010-04-24 21:26:44 +0000265 def _do_ssl_handshake(self):
266 try:
267 self.socket.do_handshake()
268 except ssl.SSLError as err:
269 if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
270 ssl.SSL_ERROR_WANT_WRITE):
271 return
272 elif err.args[0] == ssl.SSL_ERROR_EOF:
273 return self.handle_close()
274 raise
275 except socket.error as err:
276 if err.args[0] == errno.ECONNABORTED:
277 return self.handle_close()
278 else:
279 self._ssl_accepting = False
280
281 def handle_read(self):
282 if self._ssl_accepting:
283 self._do_ssl_handshake()
284 else:
285 DummyPOP3Handler.handle_read(self)
286
Giampaolo Rodolà95bcb932011-02-25 22:28:24 +0000287
Christian Heimesd3956292008-11-05 19:48:27 +0000288 class TestPOP3_SSLClass(TestPOP3Class):
289 # repeat previous tests by using poplib.POP3_SSL
290
291 def setUp(self):
292 self.server = DummyPOP3Server((HOST, PORT))
293 self.server.handler = DummyPOP3_SSLHandler
294 self.server.start()
295 self.client = poplib.POP3_SSL(self.server.host, self.server.port)
296
297 def test__all__(self):
Benjamin Peterson577473f2010-01-19 00:09:57 +0000298 self.assertIn('POP3_SSL', poplib.__all__)
Christian Heimesd3956292008-11-05 19:48:27 +0000299
Giampaolo Rodolà42382fe2010-08-17 16:09:53 +0000300 def test_context(self):
301 ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
302 self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
303 self.server.port, keyfile=CERTFILE, context=ctx)
304 self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
305 self.server.port, certfile=CERTFILE, context=ctx)
306 self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
307 self.server.port, keyfile=CERTFILE,
308 certfile=CERTFILE, context=ctx)
309
310 self.client.quit()
311 self.client = poplib.POP3_SSL(self.server.host, self.server.port,
312 context=ctx)
313 self.assertIsInstance(self.client.sock, ssl.SSLSocket)
314 self.assertIs(self.client.sock.context, ctx)
315 self.assertTrue(self.client.noop().startswith(b'+OK'))
316
Christian Heimesd3956292008-11-05 19:48:27 +0000317
318class TestTimeouts(TestCase):
Guido van Rossumd8faa362007-04-27 19:54:29 +0000319
320 def setUp(self):
321 self.evt = threading.Event()
Christian Heimes5e696852008-04-09 08:37:03 +0000322 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
Charles-François Natali83ef2542011-12-14 19:28:56 +0100323 self.sock.settimeout(60) # Safety net. Look issue 11812
Christian Heimesd3956292008-11-05 19:48:27 +0000324 self.port = test_support.bind_port(self.sock)
Charles-François Natali83ef2542011-12-14 19:28:56 +0100325 self.thread = threading.Thread(target=self.server, args=(self.evt,self.sock))
326 self.thread.setDaemon(True)
327 self.thread.start()
328 self.evt.wait()
Guido van Rossumd8faa362007-04-27 19:54:29 +0000329
330 def tearDown(self):
Charles-François Natali83ef2542011-12-14 19:28:56 +0100331 self.thread.join()
332 del self.thread # Clear out any dangling Thread objects.
Guido van Rossumd8faa362007-04-27 19:54:29 +0000333
Christian Heimesd3956292008-11-05 19:48:27 +0000334 def server(self, evt, serv):
335 serv.listen(5)
Charles-François Natali83ef2542011-12-14 19:28:56 +0100336 evt.set()
Christian Heimesd3956292008-11-05 19:48:27 +0000337 try:
338 conn, addr = serv.accept()
Christian Heimesd3956292008-11-05 19:48:27 +0000339 conn.send(b"+ Hola mundo\n")
340 conn.close()
Charles-François Natali83ef2542011-12-14 19:28:56 +0100341 except socket.timeout:
342 pass
Christian Heimesd3956292008-11-05 19:48:27 +0000343 finally:
344 serv.close()
Guido van Rossumd8faa362007-04-27 19:54:29 +0000345
346 def testTimeoutDefault(self):
Georg Brandlf78e02b2008-06-10 17:40:04 +0000347 self.assertTrue(socket.getdefaulttimeout() is None)
348 socket.setdefaulttimeout(30)
349 try:
Charles-François Natali83ef2542011-12-14 19:28:56 +0100350 pop = poplib.POP3(HOST, self.port)
Georg Brandlf78e02b2008-06-10 17:40:04 +0000351 finally:
352 socket.setdefaulttimeout(None)
Guido van Rossumd8faa362007-04-27 19:54:29 +0000353 self.assertEqual(pop.sock.gettimeout(), 30)
354 pop.sock.close()
355
356 def testTimeoutNone(self):
Georg Brandlf78e02b2008-06-10 17:40:04 +0000357 self.assertTrue(socket.getdefaulttimeout() is None)
Guido van Rossumd8faa362007-04-27 19:54:29 +0000358 socket.setdefaulttimeout(30)
359 try:
Christian Heimes5e696852008-04-09 08:37:03 +0000360 pop = poplib.POP3(HOST, self.port, timeout=None)
Guido van Rossumd8faa362007-04-27 19:54:29 +0000361 finally:
Georg Brandlf78e02b2008-06-10 17:40:04 +0000362 socket.setdefaulttimeout(None)
363 self.assertTrue(pop.sock.gettimeout() is None)
Guido van Rossumd8faa362007-04-27 19:54:29 +0000364 pop.sock.close()
365
Georg Brandlf78e02b2008-06-10 17:40:04 +0000366 def testTimeoutValue(self):
Charles-François Natali83ef2542011-12-14 19:28:56 +0100367 pop = poplib.POP3(HOST, self.port, timeout=30)
Georg Brandlf78e02b2008-06-10 17:40:04 +0000368 self.assertEqual(pop.sock.gettimeout(), 30)
369 pop.sock.close()
Guido van Rossumd8faa362007-04-27 19:54:29 +0000370
371
Christian Heimesd3956292008-11-05 19:48:27 +0000372def test_main():
373 tests = [TestPOP3Class, TestTimeouts]
374 if SUPPORTS_SSL:
375 tests.append(TestPOP3_SSLClass)
376 thread_info = test_support.threading_setup()
377 try:
378 test_support.run_unittest(*tests)
379 finally:
380 test_support.threading_cleanup(*thread_info)
381
Guido van Rossumd8faa362007-04-27 19:54:29 +0000382
383if __name__ == '__main__':
384 test_main()