diff options
Diffstat (limited to 'Lib/test/test_poplib.py')
| -rw-r--r-- | Lib/test/test_poplib.py | 96 |
1 files changed, 62 insertions, 34 deletions
diff --git a/Lib/test/test_poplib.py b/Lib/test/test_poplib.py index 9505c2250a3..e3901b886ca 100644 --- a/Lib/test/test_poplib.py +++ b/Lib/test/test_poplib.py @@ -12,14 +12,15 @@ import time import errno from unittest import TestCase -from test import test_support -from test.test_support import HOST +from test import support as test_support threading = test_support.import_module('threading') +HOST = test_support.HOST +PORT = 0 # the dummy data returned by server when LIST and RETR commands are issued -LIST_RESP = '1 1\r\n2 2\r\n3 3\r\n4 4\r\n5 5\r\n.\r\n' -RETR_RESP = """From: postmaster@python.org\ +LIST_RESP = b'1 1\r\n2 2\r\n3 3\r\n4 4\r\n5 5\r\n.\r\n' +RETR_RESP = b"""From: postmaster@python.org\ \r\nContent-Type: text/plain\r\n\ MIME-Version: 1.0\r\n\ Subject: Dummy\r\n\ @@ -34,15 +35,16 @@ class DummyPOP3Handler(asynchat.async_chat): def __init__(self, conn): asynchat.async_chat.__init__(self, conn) - self.set_terminator("\r\n") + self.set_terminator(b"\r\n") self.in_buffer = [] - self.push('+OK dummy pop3 server ready.') + self.push('+OK dummy pop3 server ready. <timestamp>') def collect_incoming_data(self, data): self.in_buffer.append(data) def found_terminator(self): - line = ''.join(self.in_buffer) + line = b''.join(self.in_buffer) + line = str(line, 'ISO-8859-1') self.in_buffer = [] cmd = line.split(' ')[0].lower() space = line.find(' ') @@ -60,7 +62,7 @@ class DummyPOP3Handler(asynchat.async_chat): raise def push(self, data): - asynchat.async_chat.push(self, data + '\r\n') + asynchat.async_chat.push(self, data.encode("ISO-8859-1") + b'\r\n') def cmd_echo(self, arg): # sends back the received string (used by the test suite) @@ -103,6 +105,9 @@ class DummyPOP3Handler(asynchat.async_chat): def cmd_rpop(self, arg): self.push('+OK done nothing.') + def cmd_apop(self, arg): + self.push('+OK done nothing.') + class DummyPOP3Server(asyncore.dispatcher, threading.Thread): @@ -117,6 +122,7 @@ class DummyPOP3Server(asyncore.dispatcher, threading.Thread): self.active = False self.active_lock = threading.Lock() self.host, self.port = self.socket.getsockname()[:2] + self.handler_instance = None def start(self): assert not self.active @@ -138,10 +144,8 @@ class DummyPOP3Server(asyncore.dispatcher, threading.Thread): self.active = False self.join() - def handle_accept(self): - conn, addr = self.accept() - self.handler = self.handler(conn) - self.close() + def handle_accepted(self, conn, addr): + self.handler_instance = self.handler(conn) def handle_connect(self): self.close() @@ -155,12 +159,11 @@ class DummyPOP3Server(asyncore.dispatcher, threading.Thread): class TestPOP3Class(TestCase): - def assertOK(self, resp): - self.assertTrue(resp.startswith("+OK")) + self.assertTrue(resp.startswith(b"+OK")) def setUp(self): - self.server = DummyPOP3Server((HOST, 0)) + self.server = DummyPOP3Server((HOST, PORT)) self.server.start() self.client = poplib.POP3(self.server.host, self.server.port) @@ -169,7 +172,8 @@ class TestPOP3Class(TestCase): self.server.stop() def test_getwelcome(self): - self.assertEqual(self.client.getwelcome(), '+OK dummy pop3 server ready.') + self.assertEqual(self.client.getwelcome(), + b'+OK dummy pop3 server ready. <timestamp>') def test_exceptions(self): self.assertRaises(poplib.error_proto, self.client._shortcmd, 'echo -err') @@ -187,16 +191,18 @@ class TestPOP3Class(TestCase): def test_list(self): self.assertEqual(self.client.list()[1:], - (['1 1', '2 2', '3 3', '4 4', '5 5'], 25)) - self.assertTrue(self.client.list('1').endswith("OK 1 1")) + ([b'1 1', b'2 2', b'3 3', b'4 4', b'5 5'], + 25)) + self.assertTrue(self.client.list('1').endswith(b"OK 1 1")) def test_retr(self): - expected = ('+OK 116 bytes', - ['From: postmaster@python.org', 'Content-Type: text/plain', - 'MIME-Version: 1.0', 'Subject: Dummy', - '', 'line1', 'line2', 'line3'], + expected = (b'+OK 116 bytes', + [b'From: postmaster@python.org', b'Content-Type: text/plain', + b'MIME-Version: 1.0', b'Subject: Dummy', + b'', b'line1', b'line2', b'line3'], 113) - self.assertEqual(self.client.retr('foo'), expected) + foo = self.client.retr('foo') + self.assertEqual(foo, expected) def test_dele(self): self.assertOK(self.client.dele('foo')) @@ -207,11 +213,14 @@ class TestPOP3Class(TestCase): def test_rpop(self): self.assertOK(self.client.rpop('foo')) + def test_apop(self): + self.assertOK(self.client.apop('foo', 'dummypassword')) + def test_top(self): - expected = ('+OK 116 bytes', - ['From: postmaster@python.org', 'Content-Type: text/plain', - 'MIME-Version: 1.0', 'Subject: Dummy', '', - 'line1', 'line2', 'line3'], + expected = (b'+OK 116 bytes', + [b'From: postmaster@python.org', b'Content-Type: text/plain', + b'MIME-Version: 1.0', b'Subject: Dummy', b'', + b'line1', b'line2', b'line3'], 113) self.assertEqual(self.client.top(1, 1), expected) @@ -231,27 +240,29 @@ if hasattr(poplib, 'POP3_SSL'): def __init__(self, conn): asynchat.async_chat.__init__(self, conn) - self.socket = ssl.wrap_socket(self.socket, certfile=CERTFILE, + ssl_socket = ssl.wrap_socket(self.socket, certfile=CERTFILE, server_side=True, do_handshake_on_connect=False) + self.del_channel() + self.set_socket(ssl_socket) # Must try handshake before calling push() self._ssl_accepting = True self._do_ssl_handshake() - self.set_terminator("\r\n") + self.set_terminator(b"\r\n") self.in_buffer = [] - self.push('+OK dummy pop3 server ready.') + self.push('+OK dummy pop3 server ready. <timestamp>') def _do_ssl_handshake(self): try: self.socket.do_handshake() - except ssl.SSLError, err: + except ssl.SSLError as err: if err.args[0] in (ssl.SSL_ERROR_WANT_READ, ssl.SSL_ERROR_WANT_WRITE): return elif err.args[0] == ssl.SSL_ERROR_EOF: return self.handle_close() raise - except socket.error, err: + except socket.error as err: if err.args[0] == errno.ECONNABORTED: return self.handle_close() else: @@ -267,7 +278,7 @@ if hasattr(poplib, 'POP3_SSL'): # repeat previous tests by using poplib.POP3_SSL def setUp(self): - self.server = DummyPOP3Server((HOST, 0)) + self.server = DummyPOP3Server((HOST, PORT)) self.server.handler = DummyPOP3_SSLHandler self.server.start() self.client = poplib.POP3_SSL(self.server.host, self.server.port) @@ -275,6 +286,23 @@ if hasattr(poplib, 'POP3_SSL'): def test__all__(self): self.assertIn('POP3_SSL', poplib.__all__) + def test_context(self): + ctx = ssl.SSLContext(ssl.PROTOCOL_TLSv1) + self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host, + self.server.port, keyfile=CERTFILE, context=ctx) + self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host, + self.server.port, certfile=CERTFILE, context=ctx) + self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host, + self.server.port, keyfile=CERTFILE, + certfile=CERTFILE, context=ctx) + + self.client.quit() + self.client = poplib.POP3_SSL(self.server.host, self.server.port, + context=ctx) + self.assertIsInstance(self.client.sock, ssl.SSLSocket) + self.assertIs(self.client.sock.context, ctx) + self.assertTrue(self.client.noop().startswith(b'+OK')) + class TestTimeouts(TestCase): @@ -297,7 +325,7 @@ class TestTimeouts(TestCase): evt.set() try: conn, addr = serv.accept() - conn.send("+ Hola mundo\n") + conn.send(b"+ Hola mundo\n") conn.close() except socket.timeout: pass |