1 """Test script for poplib module."""
3 # Modified by Giampaolo Rodola' to give poplib.POP3 and poplib.POP3_SSL
14 from unittest
import TestCase
15 from test
import support
as test_support
17 HOST
= test_support
.HOST
20 # the dummy data returned by server when LIST and RETR commands are issued
21 LIST_RESP
= b
'1 1\r\n2 2\r\n3 3\r\n4 4\r\n5 5\r\n.\r\n'
22 RETR_RESP
= b
"""From: postmaster@python.org\
23 \r\nContent-Type: text/plain\r\n\
24 MIME-Version: 1.0\r\n\
33 class DummyPOP3Handler(asynchat
.async_chat
):
35 def __init__(self
, conn
):
36 asynchat
.async_chat
.__init
__(self
, conn
)
37 self
.set_terminator(b
"\r\n")
39 self
.push('+OK dummy pop3 server ready.')
41 def collect_incoming_data(self
, data
):
42 self
.in_buffer
.append(data
)
44 def found_terminator(self
):
45 line
= b
''.join(self
.in_buffer
)
46 line
= str(line
, 'ISO-8859-1')
48 cmd
= line
.split(' ')[0].lower()
49 space
= line
.find(' ')
51 arg
= line
[space
+ 1:]
54 if hasattr(self
, 'cmd_' + cmd
):
55 method
= getattr(self
, 'cmd_' + cmd
)
58 self
.push('-ERR unrecognized POP3 command "%s".' %cmd
)
60 def handle_error(self
):
64 asynchat
.async_chat
.push(self
, data
.encode("ISO-8859-1") + b
'\r\n')
66 def cmd_echo(self
, arg
):
67 # sends back the received string (used by the test suite)
70 def cmd_user(self
, arg
):
72 self
.push("-ERR no such user")
73 self
.push('+OK password required')
75 def cmd_pass(self
, arg
):
77 self
.push("-ERR wrong password")
78 self
.push('+OK 10 messages')
80 def cmd_stat(self
, arg
):
81 self
.push('+OK 10 100')
83 def cmd_list(self
, arg
):
85 self
.push('+OK %s %s' %(arg
, arg
))
88 asynchat
.async_chat
.push(self
, LIST_RESP
)
92 def cmd_retr(self
, arg
):
93 self
.push('+OK %s bytes' %len(RETR_RESP
))
94 asynchat
.async_chat
.push(self
, RETR_RESP
)
98 def cmd_dele(self
, arg
):
99 self
.push('+OK message marked for deletion.')
101 def cmd_noop(self
, arg
):
102 self
.push('+OK done nothing.')
104 def cmd_rpop(self
, arg
):
105 self
.push('+OK done nothing.')
108 class DummyPOP3Server(asyncore
.dispatcher
, threading
.Thread
):
110 handler
= DummyPOP3Handler
112 def __init__(self
, address
, af
=socket
.AF_INET
):
113 threading
.Thread
.__init
__(self
)
114 asyncore
.dispatcher
.__init
__(self
)
115 self
.create_socket(af
, socket
.SOCK_STREAM
)
119 self
.active_lock
= threading
.Lock()
120 self
.host
, self
.port
= self
.socket
.getsockname()[:2]
123 assert not self
.active
124 self
.__flag
= threading
.Event()
125 threading
.Thread
.start(self
)
131 while self
.active
and asyncore
.socket_map
:
132 self
.active_lock
.acquire()
133 asyncore
.loop(timeout
=0.1, count
=1)
134 self
.active_lock
.release()
135 asyncore
.close_all(ignore_all
=True)
142 def handle_accept(self
):
143 conn
, addr
= self
.accept()
144 self
.handler
= self
.handler(conn
)
147 def handle_connect(self
):
149 handle_read
= handle_connect
154 def handle_error(self
):
158 class TestPOP3Class(TestCase
):
159 def assertOK(self
, resp
):
160 self
.assertTrue(resp
.startswith(b
"+OK"))
163 self
.server
= DummyPOP3Server((HOST
, PORT
))
165 self
.client
= poplib
.POP3(self
.server
.host
, self
.server
.port
)
171 def test_getwelcome(self
):
172 self
.assertEqual(self
.client
.getwelcome(), b
'+OK dummy pop3 server ready.')
174 def test_exceptions(self
):
175 self
.assertRaises(poplib
.error_proto
, self
.client
._shortcmd
, 'echo -err')
178 self
.assertOK(self
.client
.user('guido'))
179 self
.assertRaises(poplib
.error_proto
, self
.client
.user
, 'invalid')
181 def test_pass_(self
):
182 self
.assertOK(self
.client
.pass_('python'))
183 self
.assertRaises(poplib
.error_proto
, self
.client
.user
, 'invalid')
186 self
.assertEqual(self
.client
.stat(), (10, 100))
189 self
.assertEqual(self
.client
.list()[1:],
190 ([b
'1 1', b
'2 2', b
'3 3', b
'4 4', b
'5 5'],
192 self
.assertTrue(self
.client
.list('1').endswith(b
"OK 1 1"))
195 expected
= (b
'+OK 116 bytes',
196 [b
'From: postmaster@python.org', b
'Content-Type: text/plain',
197 b
'MIME-Version: 1.0', b
'Subject: Dummy',
198 b
'', b
'line1', b
'line2', b
'line3'],
200 foo
= self
.client
.retr('foo')
201 self
.assertEqual(foo
, expected
)
204 self
.assertOK(self
.client
.dele('foo'))
207 self
.assertOK(self
.client
.noop())
210 self
.assertOK(self
.client
.rpop('foo'))
213 expected
= (b
'+OK 116 bytes',
214 [b
'From: postmaster@python.org', b
'Content-Type: text/plain',
215 b
'MIME-Version: 1.0', b
'Subject: Dummy', b
'',
216 b
'line1', b
'line2', b
'line3'],
218 self
.assertEqual(self
.client
.top(1, 1), expected
)
222 self
.client
.uidl('foo')
226 if hasattr(poplib
, 'POP3_SSL'):
230 CERTFILE
= os
.path
.join(os
.path
.dirname(__file__
) or os
.curdir
, "keycert.pem")
232 class DummyPOP3_SSLHandler(DummyPOP3Handler
):
234 def __init__(self
, conn
):
235 asynchat
.async_chat
.__init
__(self
, conn
)
236 ssl_socket
= ssl
.wrap_socket(self
.socket
, certfile
=CERTFILE
,
239 self
.set_socket(ssl_socket
)
240 self
.set_terminator(b
"\r\n")
242 self
.push('+OK dummy pop3 server ready.')
244 class TestPOP3_SSLClass(TestPOP3Class
):
245 # repeat previous tests by using poplib.POP3_SSL
248 self
.server
= DummyPOP3Server((HOST
, PORT
))
249 self
.server
.handler
= DummyPOP3_SSLHandler
251 self
.client
= poplib
.POP3_SSL(self
.server
.host
, self
.server
.port
)
253 def test__all__(self
):
254 self
.assert_('POP3_SSL' in poplib
.__all
__)
257 class TestTimeouts(TestCase
):
260 self
.evt
= threading
.Event()
261 self
.sock
= socket
.socket(socket
.AF_INET
, socket
.SOCK_STREAM
)
262 self
.sock
.settimeout(3)
263 self
.port
= test_support
.bind_port(self
.sock
)
264 threading
.Thread(target
=self
.server
, args
=(self
.evt
,self
.sock
)).start()
270 def server(self
, evt
, serv
):
273 conn
, addr
= serv
.accept()
274 except socket
.timeout
:
277 conn
.send(b
"+ Hola mundo\n")
283 def testTimeoutDefault(self
):
284 self
.assertTrue(socket
.getdefaulttimeout() is None)
285 socket
.setdefaulttimeout(30)
287 pop
= poplib
.POP3("localhost", self
.port
)
289 socket
.setdefaulttimeout(None)
290 self
.assertEqual(pop
.sock
.gettimeout(), 30)
293 def testTimeoutNone(self
):
294 self
.assertTrue(socket
.getdefaulttimeout() is None)
295 socket
.setdefaulttimeout(30)
297 pop
= poplib
.POP3(HOST
, self
.port
, timeout
=None)
299 socket
.setdefaulttimeout(None)
300 self
.assertTrue(pop
.sock
.gettimeout() is None)
303 def testTimeoutValue(self
):
304 pop
= poplib
.POP3("localhost", self
.port
, timeout
=30)
305 self
.assertEqual(pop
.sock
.gettimeout(), 30)
310 tests
= [TestPOP3Class
, TestTimeouts
]
312 tests
.append(TestPOP3_SSLClass
)
313 thread_info
= test_support
.threading_setup()
315 test_support
.run_unittest(*tests
)
317 test_support
.threading_cleanup(*thread_info
)
320 if __name__
== '__main__':