blob: 39984c535de564ea046c3db7ea94b5cb9790a21b [file] [log] [blame]
R. David Murraya6cdd832009-12-12 18:38:24 +00001from test import support
2# If we end up with a significant number of tests that don't require
3# threading, this test module should be split. Right now we skip
4# them all if we don't have threading.
5threading = support.import_module('threading')
6
7from contextlib import contextmanager
Martin v. Löwisea752fb2002-01-05 11:31:49 +00008import imaplib
R. David Murraya6cdd832009-12-12 18:38:24 +00009import os.path
10import select
11import socket
12import socketserver
13import sys
Tim Peters108b7912002-07-31 16:42:33 +000014import time
Martin v. Löwisea752fb2002-01-05 11:31:49 +000015
Antoine Pitrou7ec819e2010-11-09 22:57:20 +000016from test.support import reap_threads, verbose, transient_internet
Christian Heimesf6cd9672008-03-26 13:45:42 +000017import unittest
Piers Lauderf0a70f62002-06-17 07:06:24 +000018
R. David Murraya6cdd832009-12-12 18:38:24 +000019try:
20 import ssl
21except ImportError:
22 ssl = None
23
24CERTFILE = None
25
Martin v. Löwisea752fb2002-01-05 11:31:49 +000026
Christian Heimesf6cd9672008-03-26 13:45:42 +000027class TestImaplib(unittest.TestCase):
R. David Murraya6cdd832009-12-12 18:38:24 +000028
Alexander Belopolsky87bb6e92011-01-29 18:43:43 +000029 def test_Internaldate2tuple(self):
30 tt = imaplib.Internaldate2tuple(
31 b'25 (INTERNALDATE "01-Jan-1970 00:00:00 +0000")')
32 self.assertEqual(time.mktime(tt), 0)
33 tt = imaplib.Internaldate2tuple(
34 b'25 (INTERNALDATE "01-Jan-1970 11:30:00 +1130")')
35 self.assertEqual(time.mktime(tt), 0)
36 tt = imaplib.Internaldate2tuple(
37 b'25 (INTERNALDATE "31-Dec-1969 12:30:00 -1130")')
38 self.assertEqual(time.mktime(tt), 0)
39
Christian Heimesf6cd9672008-03-26 13:45:42 +000040 def test_that_Time2Internaldate_returns_a_result(self):
41 # We can check only that it successfully produces a result,
42 # not the correctness of the result itself, since the result
43 # depends on the timezone the machine is in.
44 timevalues = [2000000000, 2000000000.0, time.localtime(2000000000),
45 '"18-May-2033 05:33:20 +0200"']
46
47 for t in timevalues:
48 imaplib.Time2Internaldate(t)
49
50
R. David Murraya6cdd832009-12-12 18:38:24 +000051if ssl:
52
53 class SecureTCPServer(socketserver.TCPServer):
54
55 def get_request(self):
56 newsocket, fromaddr = self.socket.accept()
57 connstream = ssl.wrap_socket(newsocket,
58 server_side=True,
59 certfile=CERTFILE)
60 return connstream, fromaddr
61
62 IMAP4_SSL = imaplib.IMAP4_SSL
63
64else:
65
66 class SecureTCPServer:
67 pass
68
69 IMAP4_SSL = None
70
71
72class SimpleIMAPHandler(socketserver.StreamRequestHandler):
73
74 timeout = 1
75
76 def _send(self, message):
77 if verbose: print("SENT:", message.strip())
78 self.wfile.write(message)
79
80 def handle(self):
81 # Send a welcome message.
82 self._send(b'* OK IMAP4rev1\r\n')
83 while 1:
84 # Gather up input until we receive a line terminator or we timeout.
85 # Accumulate read(1) because it's simpler to handle the differences
86 # between naked sockets and SSL sockets.
87 line = b''
88 while 1:
89 try:
90 part = self.rfile.read(1)
91 if part == b'':
92 # Naked sockets return empty strings..
93 return
94 line += part
95 except IOError:
96 # ..but SSLSockets throw exceptions.
97 return
98 if line.endswith(b'\r\n'):
99 break
100
101 if verbose: print('GOT:', line.strip())
102 splitline = line.split()
103 tag = splitline[0].decode('ASCII')
104 cmd = splitline[1].decode('ASCII')
105 args = splitline[2:]
106
107 if hasattr(self, 'cmd_'+cmd):
108 getattr(self, 'cmd_'+cmd)(tag, args)
109 else:
110 self._send('{} BAD {} unknown\r\n'.format(tag, cmd).encode('ASCII'))
111
112 def cmd_CAPABILITY(self, tag, args):
113 self._send(b'* CAPABILITY IMAP4rev1\r\n')
114 self._send('{} OK CAPABILITY completed\r\n'.format(tag).encode('ASCII'))
115
116
117class BaseThreadedNetworkedTests(unittest.TestCase):
118
119 def make_server(self, addr, hdlr):
120
121 class MyServer(self.server_class):
122 def handle_error(self, request, client_address):
123 self.close_request(request)
124 self.server_close()
125 raise
126
127 if verbose: print("creating server")
128 server = MyServer(addr, hdlr)
Ezio Melotti19f2aeb2010-11-21 01:30:29 +0000129 self.assertEqual(server.server_address, server.socket.getsockname())
R. David Murraya6cdd832009-12-12 18:38:24 +0000130
131 if verbose:
132 print("server created")
133 print("ADDR =", addr)
134 print("CLASS =", self.server_class)
135 print("HDLR =", server.RequestHandlerClass)
136
137 t = threading.Thread(
138 name='%s serving' % self.server_class,
139 target=server.serve_forever,
140 # Short poll interval to make the test finish quickly.
141 # Time between requests is short enough that we won't wake
142 # up spuriously too many times.
143 kwargs={'poll_interval':0.01})
144 t.daemon = True # In case this function raises.
145 t.start()
146 if verbose: print("server running")
147 return server, t
148
149 def reap_server(self, server, thread):
150 if verbose: print("waiting for server")
151 server.shutdown()
152 thread.join()
153 if verbose: print("done")
154
155 @contextmanager
156 def reaped_server(self, hdlr):
157 server, thread = self.make_server((support.HOST, 0), hdlr)
158 try:
159 yield server
160 finally:
161 self.reap_server(server, thread)
162
163 @reap_threads
164 def test_connect(self):
165 with self.reaped_server(SimpleIMAPHandler) as server:
166 client = self.imap_class(*server.server_address)
167 client.shutdown()
168
169 @reap_threads
170 def test_issue5949(self):
171
172 class EOFHandler(socketserver.StreamRequestHandler):
173 def handle(self):
174 # EOF without sending a complete welcome message.
175 self.wfile.write(b'* OK')
176
177 with self.reaped_server(EOFHandler) as server:
178 self.assertRaises(imaplib.IMAP4.abort,
179 self.imap_class, *server.server_address)
180
181 @reap_threads
182 def test_line_termination(self):
183
184 class BadNewlineHandler(SimpleIMAPHandler):
185
186 def cmd_CAPABILITY(self, tag, args):
187 self._send(b'* CAPABILITY IMAP4rev1 AUTH\n')
188 self._send('{} OK CAPABILITY completed\r\n'.format(tag).encode('ASCII'))
189
190 with self.reaped_server(BadNewlineHandler) as server:
191 self.assertRaises(imaplib.IMAP4.abort,
192 self.imap_class, *server.server_address)
193
194
195
196class ThreadedNetworkedTests(BaseThreadedNetworkedTests):
197
198 server_class = socketserver.TCPServer
199 imap_class = imaplib.IMAP4
200
201
202@unittest.skipUnless(ssl, "SSL not available")
203class ThreadedNetworkedTestsSSL(BaseThreadedNetworkedTests):
204
205 server_class = SecureTCPServer
206 imap_class = IMAP4_SSL
207
208
Antoine Pitrou7ec819e2010-11-09 22:57:20 +0000209class RemoteIMAPTest(unittest.TestCase):
210 host = 'cyrus.andrew.cmu.edu'
211 port = 143
212 username = 'anonymous'
213 password = 'pass'
214 imap_class = imaplib.IMAP4
R. David Murraya6cdd832009-12-12 18:38:24 +0000215
Antoine Pitrou7ec819e2010-11-09 22:57:20 +0000216 def setUp(self):
217 with transient_internet(self.host):
218 self.server = self.imap_class(self.host, self.port)
219
220 def tearDown(self):
221 if self.server is not None:
222 self.server.logout()
223
224 def test_logincapa(self):
225 self.assertTrue('LOGINDISABLED' in self.server.capabilities)
226
227 def test_anonlogin(self):
228 self.assertTrue('AUTH=ANONYMOUS' in self.server.capabilities)
229 rs = self.server.login(self.username, self.password)
230 self.assertEqual(rs[0], 'OK')
231
232 def test_logout(self):
233 rs = self.server.logout()
Antoine Pitrou5466e922010-11-09 23:12:32 +0000234 self.server = None
Antoine Pitrou7ec819e2010-11-09 22:57:20 +0000235 self.assertEqual(rs[0], 'BYE')
236
237
238@unittest.skipUnless(ssl, "SSL not available")
239class RemoteIMAP_SSLTest(RemoteIMAPTest):
240 port = 993
241 imap_class = IMAP4_SSL
242
243 def test_logincapa(self):
244 self.assertFalse('LOGINDISABLED' in self.server.capabilities)
245 self.assertTrue('AUTH=PLAIN' in self.server.capabilities)
246
247
248def test_main():
R. David Murraya6cdd832009-12-12 18:38:24 +0000249 tests = [TestImaplib]
250
251 if support.is_resource_enabled('network'):
252 if ssl:
253 global CERTFILE
254 CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir,
255 "keycert.pem")
256 if not os.path.exists(CERTFILE):
257 raise support.TestFailed("Can't read certificate files!")
Antoine Pitrou7ec819e2010-11-09 22:57:20 +0000258 tests.extend([
259 ThreadedNetworkedTests, ThreadedNetworkedTestsSSL,
260 RemoteIMAPTest, RemoteIMAP_SSLTest,
261 ])
R. David Murraya6cdd832009-12-12 18:38:24 +0000262
263 support.run_unittest(*tests)
Christian Heimesf6cd9672008-03-26 13:45:42 +0000264
265
266if __name__ == "__main__":
R. David Murraya6cdd832009-12-12 18:38:24 +0000267 support.use_resources = ['network']
268 test_main()