1 """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 6 import poplib 7 import asyncore 8 import asynchat 9 import socket 10 import os 11 import time 12 import errno 13 14 from unittest import TestCase, skipUnless 15 from test import test_support 16 from test.test_support import HOST 17 threading = test_support.import_module('threading') 18 19 20 # the dummy data returned by server when LIST and RETR commands are issued 21 LIST_RESP = '1 1\r\n2 2\r\n3 3\r\n4 4\r\n5 5\r\n.\r\n' 22 RETR_RESP = """From: postmaster (at] python.org\ 23 \r\nContent-Type: text/plain\r\n\ 24 MIME-Version: 1.0\r\n\ 25 Subject: Dummy\r\n\ 26 \r\n\ 27 line1\r\n\ 28 line2\r\n\ 29 line3\r\n\ 30 .\r\n""" 31 32 33 class DummyPOP3Handler(asynchat.async_chat): 34 35 def __init__(self, conn): 36 asynchat.async_chat.__init__(self, conn) 37 self.set_terminator("\r\n") 38 self.in_buffer = [] 39 self.push('+OK dummy pop3 server ready.') 40 41 def collect_incoming_data(self, data): 42 self.in_buffer.append(data) 43 44 def found_terminator(self): 45 line = ''.join(self.in_buffer) 46 self.in_buffer = [] 47 cmd = line.split(' ')[0].lower() 48 space = line.find(' ') 49 if space != -1: 50 arg = line[space + 1:] 51 else: 52 arg = "" 53 if hasattr(self, 'cmd_' + cmd): 54 method = getattr(self, 'cmd_' + cmd) 55 method(arg) 56 else: 57 self.push('-ERR unrecognized POP3 command "%s".' %cmd) 58 59 def handle_error(self): 60 raise 61 62 def push(self, data): 63 asynchat.async_chat.push(self, data + '\r\n') 64 65 def cmd_echo(self, arg): 66 # sends back the received string (used by the test suite) 67 self.push(arg) 68 69 def cmd_user(self, arg): 70 if arg != "guido": 71 self.push("-ERR no such user") 72 self.push('+OK password required') 73 74 def cmd_pass(self, arg): 75 if arg != "python": 76 self.push("-ERR wrong password") 77 self.push('+OK 10 messages') 78 79 def cmd_stat(self, arg): 80 self.push('+OK 10 100') 81 82 def cmd_list(self, arg): 83 if arg: 84 self.push('+OK %s %s' %(arg, arg)) 85 else: 86 self.push('+OK') 87 asynchat.async_chat.push(self, LIST_RESP) 88 89 cmd_uidl = cmd_list 90 91 def cmd_retr(self, arg): 92 self.push('+OK %s bytes' %len(RETR_RESP)) 93 asynchat.async_chat.push(self, RETR_RESP) 94 95 cmd_top = cmd_retr 96 97 def cmd_dele(self, arg): 98 self.push('+OK message marked for deletion.') 99 100 def cmd_noop(self, arg): 101 self.push('+OK done nothing.') 102 103 def cmd_rpop(self, arg): 104 self.push('+OK done nothing.') 105 106 107 class DummyPOP3Server(asyncore.dispatcher, threading.Thread): 108 109 handler = DummyPOP3Handler 110 111 def __init__(self, address, af=socket.AF_INET): 112 threading.Thread.__init__(self) 113 asyncore.dispatcher.__init__(self) 114 self.create_socket(af, socket.SOCK_STREAM) 115 self.bind(address) 116 self.listen(5) 117 self.active = False 118 self.active_lock = threading.Lock() 119 self.host, self.port = self.socket.getsockname()[:2] 120 121 def start(self): 122 assert not self.active 123 self.__flag = threading.Event() 124 threading.Thread.start(self) 125 self.__flag.wait() 126 127 def run(self): 128 self.active = True 129 self.__flag.set() 130 while self.active and asyncore.socket_map: 131 self.active_lock.acquire() 132 asyncore.loop(timeout=0.1, count=1) 133 self.active_lock.release() 134 asyncore.close_all(ignore_all=True) 135 136 def stop(self): 137 assert self.active 138 self.active = False 139 self.join() 140 141 def handle_accept(self): 142 conn, addr = self.accept() 143 self.handler = self.handler(conn) 144 self.close() 145 146 def handle_connect(self): 147 self.close() 148 handle_read = handle_connect 149 150 def writable(self): 151 return 0 152 153 def handle_error(self): 154 raise 155 156 157 class TestPOP3Class(TestCase): 158 159 def assertOK(self, resp): 160 self.assertTrue(resp.startswith("+OK")) 161 162 def setUp(self): 163 self.server = DummyPOP3Server((HOST, 0)) 164 self.server.start() 165 self.client = poplib.POP3(self.server.host, self.server.port) 166 167 def tearDown(self): 168 self.client.quit() 169 self.server.stop() 170 171 def test_getwelcome(self): 172 self.assertEqual(self.client.getwelcome(), '+OK dummy pop3 server ready.') 173 174 def test_exceptions(self): 175 self.assertRaises(poplib.error_proto, self.client._shortcmd, 'echo -err') 176 177 def test_user(self): 178 self.assertOK(self.client.user('guido')) 179 self.assertRaises(poplib.error_proto, self.client.user, 'invalid') 180 181 def test_pass_(self): 182 self.assertOK(self.client.pass_('python')) 183 self.assertRaises(poplib.error_proto, self.client.user, 'invalid') 184 185 def test_stat(self): 186 self.assertEqual(self.client.stat(), (10, 100)) 187 188 def test_list(self): 189 self.assertEqual(self.client.list()[1:], 190 (['1 1', '2 2', '3 3', '4 4', '5 5'], 25)) 191 self.assertTrue(self.client.list('1').endswith("OK 1 1")) 192 193 def test_retr(self): 194 expected = ('+OK 116 bytes', 195 ['From: postmaster (at] python.org', 'Content-Type: text/plain', 196 'MIME-Version: 1.0', 'Subject: Dummy', 197 '', 'line1', 'line2', 'line3'], 198 113) 199 self.assertEqual(self.client.retr('foo'), expected) 200 201 def test_too_long_lines(self): 202 self.assertRaises(poplib.error_proto, self.client._shortcmd, 203 'echo +%s' % ((poplib._MAXLINE + 10) * 'a')) 204 205 def test_dele(self): 206 self.assertOK(self.client.dele('foo')) 207 208 def test_noop(self): 209 self.assertOK(self.client.noop()) 210 211 def test_rpop(self): 212 self.assertOK(self.client.rpop('foo')) 213 214 def test_top(self): 215 expected = ('+OK 116 bytes', 216 ['From: postmaster (at] python.org', 'Content-Type: text/plain', 217 'MIME-Version: 1.0', 'Subject: Dummy', '', 218 'line1', 'line2', 'line3'], 219 113) 220 self.assertEqual(self.client.top(1, 1), expected) 221 222 def test_uidl(self): 223 self.client.uidl() 224 self.client.uidl('foo') 225 226 227 SUPPORTS_SSL = False 228 if hasattr(poplib, 'POP3_SSL'): 229 import ssl 230 231 SUPPORTS_SSL = True 232 CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir, "keycert.pem") 233 234 class DummyPOP3_SSLHandler(DummyPOP3Handler): 235 236 def __init__(self, conn): 237 asynchat.async_chat.__init__(self, conn) 238 self.socket = ssl.wrap_socket(self.socket, certfile=CERTFILE, 239 server_side=True, 240 do_handshake_on_connect=False) 241 # Must try handshake before calling push() 242 self._ssl_accepting = True 243 self._do_ssl_handshake() 244 self.set_terminator("\r\n") 245 self.in_buffer = [] 246 self.push('+OK dummy pop3 server ready.') 247 248 def _do_ssl_handshake(self): 249 try: 250 self.socket.do_handshake() 251 except ssl.SSLError, err: 252 if err.args[0] in (ssl.SSL_ERROR_WANT_READ, 253 ssl.SSL_ERROR_WANT_WRITE): 254 return 255 elif err.args[0] == ssl.SSL_ERROR_EOF: 256 return self.handle_close() 257 raise 258 except socket.error, err: 259 if err.args[0] == errno.ECONNABORTED: 260 return self.handle_close() 261 else: 262 self._ssl_accepting = False 263 264 def handle_read(self): 265 if self._ssl_accepting: 266 self._do_ssl_handshake() 267 else: 268 DummyPOP3Handler.handle_read(self) 269 270 requires_ssl = skipUnless(SUPPORTS_SSL, 'SSL not supported') 271 272 @requires_ssl 273 class TestPOP3_SSLClass(TestPOP3Class): 274 # repeat previous tests by using poplib.POP3_SSL 275 276 def setUp(self): 277 self.server = DummyPOP3Server((HOST, 0)) 278 self.server.handler = DummyPOP3_SSLHandler 279 self.server.start() 280 self.client = poplib.POP3_SSL(self.server.host, self.server.port) 281 282 def test__all__(self): 283 self.assertIn('POP3_SSL', poplib.__all__) 284 285 286 class TestTimeouts(TestCase): 287 288 def setUp(self): 289 self.evt = threading.Event() 290 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) 291 self.sock.settimeout(60) # Safety net. Look issue 11812 292 self.port = test_support.bind_port(self.sock) 293 self.thread = threading.Thread(target=self.server, args=(self.evt,self.sock)) 294 self.thread.setDaemon(True) 295 self.thread.start() 296 self.evt.wait() 297 298 def tearDown(self): 299 self.thread.join() 300 del self.thread # Clear out any dangling Thread objects. 301 302 def server(self, evt, serv): 303 serv.listen(5) 304 evt.set() 305 try: 306 conn, addr = serv.accept() 307 conn.send("+ Hola mundo\n") 308 conn.close() 309 except socket.timeout: 310 pass 311 finally: 312 serv.close() 313 314 def testTimeoutDefault(self): 315 self.assertIsNone(socket.getdefaulttimeout()) 316 socket.setdefaulttimeout(30) 317 try: 318 pop = poplib.POP3(HOST, self.port) 319 finally: 320 socket.setdefaulttimeout(None) 321 self.assertEqual(pop.sock.gettimeout(), 30) 322 pop.sock.close() 323 324 def testTimeoutNone(self): 325 self.assertIsNone(socket.getdefaulttimeout()) 326 socket.setdefaulttimeout(30) 327 try: 328 pop = poplib.POP3(HOST, self.port, timeout=None) 329 finally: 330 socket.setdefaulttimeout(None) 331 self.assertIsNone(pop.sock.gettimeout()) 332 pop.sock.close() 333 334 def testTimeoutValue(self): 335 pop = poplib.POP3(HOST, self.port, timeout=30) 336 self.assertEqual(pop.sock.gettimeout(), 30) 337 pop.sock.close() 338 339 340 def test_main(): 341 tests = [TestPOP3Class, TestTimeouts, 342 TestPOP3_SSLClass] 343 thread_info = test_support.threading_setup() 344 try: 345 test_support.run_unittest(*tests) 346 finally: 347 test_support.threading_cleanup(*thread_info) 348 349 350 if __name__ == '__main__': 351 test_main() 352
