R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 1 | from 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. |
| 5 | threading = support.import_module('threading') |
| 6 | |
| 7 | from contextlib import contextmanager |
Martin v. Löwis | ea752fb | 2002-01-05 11:31:49 +0000 | [diff] [blame] | 8 | import imaplib |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 9 | import os.path |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 10 | import socketserver |
Tim Peters | 108b791 | 2002-07-31 16:42:33 +0000 | [diff] [blame] | 11 | import time |
Alexander Belopolsky | 7dabf16 | 2011-01-29 19:49:40 +0000 | [diff] [blame] | 12 | import calendar |
Martin v. Löwis | ea752fb | 2002-01-05 11:31:49 +0000 | [diff] [blame] | 13 | |
Alexander Belopolsky | 8141cc7 | 2012-06-22 21:03:39 -0400 | [diff] [blame] | 14 | from test.support import reap_threads, verbose, transient_internet, run_with_tz, run_with_locale |
Christian Heimes | f6cd967 | 2008-03-26 13:45:42 +0000 | [diff] [blame] | 15 | import unittest |
Alexander Belopolsky | 8141cc7 | 2012-06-22 21:03:39 -0400 | [diff] [blame] | 16 | from datetime import datetime, timezone, timedelta |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 17 | try: |
| 18 | import ssl |
| 19 | except ImportError: |
| 20 | ssl = None |
| 21 | |
| 22 | CERTFILE = None |
| 23 | |
Martin v. Löwis | ea752fb | 2002-01-05 11:31:49 +0000 | [diff] [blame] | 24 | |
Christian Heimes | f6cd967 | 2008-03-26 13:45:42 +0000 | [diff] [blame] | 25 | class TestImaplib(unittest.TestCase): |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 26 | |
Alexander Belopolsky | 19e0a9e | 2011-01-29 17:19:08 +0000 | [diff] [blame] | 27 | def test_Internaldate2tuple(self): |
Alexander Belopolsky | 7dabf16 | 2011-01-29 19:49:40 +0000 | [diff] [blame] | 28 | t0 = calendar.timegm((2000, 1, 1, 0, 0, 0, -1, -1, -1)) |
Alexander Belopolsky | 19e0a9e | 2011-01-29 17:19:08 +0000 | [diff] [blame] | 29 | tt = imaplib.Internaldate2tuple( |
Alexander Belopolsky | 7dabf16 | 2011-01-29 19:49:40 +0000 | [diff] [blame] | 30 | b'25 (INTERNALDATE "01-Jan-2000 00:00:00 +0000")') |
| 31 | self.assertEqual(time.mktime(tt), t0) |
Alexander Belopolsky | 19e0a9e | 2011-01-29 17:19:08 +0000 | [diff] [blame] | 32 | tt = imaplib.Internaldate2tuple( |
Alexander Belopolsky | 7dabf16 | 2011-01-29 19:49:40 +0000 | [diff] [blame] | 33 | b'25 (INTERNALDATE "01-Jan-2000 11:30:00 +1130")') |
| 34 | self.assertEqual(time.mktime(tt), t0) |
Alexander Belopolsky | 19e0a9e | 2011-01-29 17:19:08 +0000 | [diff] [blame] | 35 | tt = imaplib.Internaldate2tuple( |
Alexander Belopolsky | 7dabf16 | 2011-01-29 19:49:40 +0000 | [diff] [blame] | 36 | b'25 (INTERNALDATE "31-Dec-1999 12:30:00 -1130")') |
| 37 | self.assertEqual(time.mktime(tt), t0) |
Alexander Belopolsky | 19e0a9e | 2011-01-29 17:19:08 +0000 | [diff] [blame] | 38 | |
Alexander Belopolsky | 2420d83 | 2012-04-29 15:56:49 -0400 | [diff] [blame] | 39 | @run_with_tz('MST+07MDT,M4.1.0,M10.5.0') |
| 40 | def test_Internaldate2tuple_issue10941(self): |
| 41 | self.assertNotEqual(imaplib.Internaldate2tuple( |
| 42 | b'25 (INTERNALDATE "02-Apr-2000 02:30:00 +0000")'), |
| 43 | imaplib.Internaldate2tuple( |
| 44 | b'25 (INTERNALDATE "02-Apr-2000 03:30:00 +0000")')) |
| 45 | |
Christian Heimes | f6cd967 | 2008-03-26 13:45:42 +0000 | [diff] [blame] | 46 | |
Alexander Belopolsky | 8141cc7 | 2012-06-22 21:03:39 -0400 | [diff] [blame] | 47 | |
| 48 | def timevalues(self): |
| 49 | return [2000000000, 2000000000.0, time.localtime(2000000000), |
| 50 | (2033, 5, 18, 5, 33, 20, -1, -1, -1), |
| 51 | (2033, 5, 18, 5, 33, 20, -1, -1, 1), |
Alexander Belopolsky | 6489213 | 2012-06-22 21:10:50 -0400 | [diff] [blame] | 52 | datetime.fromtimestamp(2000000000, |
Alexander Belopolsky | 8141cc7 | 2012-06-22 21:03:39 -0400 | [diff] [blame] | 53 | timezone(timedelta(0, 2*60*60))), |
| 54 | '"18-May-2033 05:33:20 +0200"'] |
| 55 | |
| 56 | @run_with_locale('LC_ALL', 'de_DE', 'fr_FR') |
| 57 | @run_with_tz('STD-1DST') |
| 58 | def test_Time2Internaldate(self): |
| 59 | expected = '"18-May-2033 05:33:20 +0200"' |
| 60 | |
| 61 | for t in self.timevalues(): |
| 62 | internal = imaplib.Time2Internaldate(t) |
| 63 | self.assertEqual(internal, expected) |
| 64 | |
| 65 | def test_that_Time2Internaldate_returns_a_result(self): |
| 66 | # Without tzset, we can check only that it successfully |
| 67 | # produces a result, not the correctness of the result itself, |
| 68 | # since the result depends on the timezone the machine is in. |
| 69 | for t in self.timevalues(): |
Christian Heimes | f6cd967 | 2008-03-26 13:45:42 +0000 | [diff] [blame] | 70 | imaplib.Time2Internaldate(t) |
| 71 | |
| 72 | |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 73 | if ssl: |
| 74 | |
| 75 | class SecureTCPServer(socketserver.TCPServer): |
| 76 | |
| 77 | def get_request(self): |
| 78 | newsocket, fromaddr = self.socket.accept() |
| 79 | connstream = ssl.wrap_socket(newsocket, |
| 80 | server_side=True, |
| 81 | certfile=CERTFILE) |
| 82 | return connstream, fromaddr |
| 83 | |
| 84 | IMAP4_SSL = imaplib.IMAP4_SSL |
| 85 | |
| 86 | else: |
| 87 | |
| 88 | class SecureTCPServer: |
| 89 | pass |
| 90 | |
| 91 | IMAP4_SSL = None |
| 92 | |
| 93 | |
| 94 | class SimpleIMAPHandler(socketserver.StreamRequestHandler): |
| 95 | |
| 96 | timeout = 1 |
R David Murray | 774a39f | 2013-02-19 12:17:31 -0500 | [diff] [blame] | 97 | continuation = None |
| 98 | capabilities = '' |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 99 | |
| 100 | def _send(self, message): |
Antoine Pitrou | 36c0dbc | 2010-11-16 17:49:46 +0000 | [diff] [blame] | 101 | if verbose: print("SENT: %r" % message.strip()) |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 102 | self.wfile.write(message) |
| 103 | |
R David Murray | 774a39f | 2013-02-19 12:17:31 -0500 | [diff] [blame] | 104 | def _send_line(self, message): |
| 105 | self._send(message + b'\r\n') |
| 106 | |
| 107 | def _send_textline(self, message): |
| 108 | self._send_line(message.encode('ASCII')) |
| 109 | |
| 110 | def _send_tagged(self, tag, code, message): |
| 111 | self._send_textline(' '.join((tag, code, message))) |
| 112 | |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 113 | def handle(self): |
| 114 | # Send a welcome message. |
R David Murray | 774a39f | 2013-02-19 12:17:31 -0500 | [diff] [blame] | 115 | self._send_textline('* OK IMAP4rev1') |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 116 | while 1: |
| 117 | # Gather up input until we receive a line terminator or we timeout. |
| 118 | # Accumulate read(1) because it's simpler to handle the differences |
| 119 | # between naked sockets and SSL sockets. |
| 120 | line = b'' |
| 121 | while 1: |
| 122 | try: |
| 123 | part = self.rfile.read(1) |
| 124 | if part == b'': |
| 125 | # Naked sockets return empty strings.. |
| 126 | return |
| 127 | line += part |
| 128 | except IOError: |
Andrew Svetlov | 737fb89 | 2012-12-18 21:14:22 +0200 | [diff] [blame] | 129 | # ..but SSLSockets raise exceptions. |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 130 | return |
| 131 | if line.endswith(b'\r\n'): |
| 132 | break |
| 133 | |
Antoine Pitrou | 36c0dbc | 2010-11-16 17:49:46 +0000 | [diff] [blame] | 134 | if verbose: print('GOT: %r' % line.strip()) |
R David Murray | 774a39f | 2013-02-19 12:17:31 -0500 | [diff] [blame] | 135 | if self.continuation: |
| 136 | try: |
| 137 | self.continuation.send(line) |
| 138 | except StopIteration: |
| 139 | self.continuation = None |
| 140 | continue |
| 141 | splitline = line.decode('ASCII').split() |
| 142 | tag = splitline[0] |
| 143 | cmd = splitline[1] |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 144 | args = splitline[2:] |
| 145 | |
| 146 | if hasattr(self, 'cmd_'+cmd): |
R David Murray | 774a39f | 2013-02-19 12:17:31 -0500 | [diff] [blame] | 147 | continuation = getattr(self, 'cmd_'+cmd)(tag, args) |
| 148 | if continuation: |
| 149 | self.continuation = continuation |
| 150 | next(continuation) |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 151 | else: |
R David Murray | 774a39f | 2013-02-19 12:17:31 -0500 | [diff] [blame] | 152 | self._send_tagged(tag, 'BAD', cmd + ' unknown') |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 153 | |
| 154 | def cmd_CAPABILITY(self, tag, args): |
R David Murray | 774a39f | 2013-02-19 12:17:31 -0500 | [diff] [blame] | 155 | caps = 'IMAP4rev1 ' + self.capabilities if self.capabilities else 'IMAP4rev1' |
| 156 | self._send_textline('* CAPABILITY ' + caps) |
| 157 | self._send_tagged(tag, 'OK', 'CAPABILITY completed') |
| 158 | |
| 159 | def cmd_LOGOUT(self, tag, args): |
| 160 | self._send_textline('* BYE IMAP4ref1 Server logging out') |
| 161 | self._send_tagged(tag, 'OK', 'LOGOUT completed') |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 162 | |
| 163 | |
| 164 | class BaseThreadedNetworkedTests(unittest.TestCase): |
| 165 | |
| 166 | def make_server(self, addr, hdlr): |
| 167 | |
| 168 | class MyServer(self.server_class): |
| 169 | def handle_error(self, request, client_address): |
| 170 | self.close_request(request) |
| 171 | self.server_close() |
| 172 | raise |
| 173 | |
| 174 | if verbose: print("creating server") |
| 175 | server = MyServer(addr, hdlr) |
Ezio Melotti | b3aedd4 | 2010-11-20 19:04:17 +0000 | [diff] [blame] | 176 | self.assertEqual(server.server_address, server.socket.getsockname()) |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 177 | |
| 178 | if verbose: |
| 179 | print("server created") |
| 180 | print("ADDR =", addr) |
| 181 | print("CLASS =", self.server_class) |
| 182 | print("HDLR =", server.RequestHandlerClass) |
| 183 | |
| 184 | t = threading.Thread( |
| 185 | name='%s serving' % self.server_class, |
| 186 | target=server.serve_forever, |
| 187 | # Short poll interval to make the test finish quickly. |
| 188 | # Time between requests is short enough that we won't wake |
| 189 | # up spuriously too many times. |
| 190 | kwargs={'poll_interval':0.01}) |
| 191 | t.daemon = True # In case this function raises. |
| 192 | t.start() |
| 193 | if verbose: print("server running") |
| 194 | return server, t |
| 195 | |
| 196 | def reap_server(self, server, thread): |
| 197 | if verbose: print("waiting for server") |
| 198 | server.shutdown() |
Victor Stinner | 73efd62 | 2011-01-05 23:01:38 +0000 | [diff] [blame] | 199 | server.server_close() |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 200 | thread.join() |
| 201 | if verbose: print("done") |
| 202 | |
| 203 | @contextmanager |
| 204 | def reaped_server(self, hdlr): |
| 205 | server, thread = self.make_server((support.HOST, 0), hdlr) |
| 206 | try: |
| 207 | yield server |
| 208 | finally: |
| 209 | self.reap_server(server, thread) |
| 210 | |
R David Murray | 774a39f | 2013-02-19 12:17:31 -0500 | [diff] [blame] | 211 | @contextmanager |
| 212 | def reaped_pair(self, hdlr): |
| 213 | server, thread = self.make_server((support.HOST, 0), hdlr) |
| 214 | client = self.imap_class(*server.server_address) |
| 215 | try: |
| 216 | yield server, client |
| 217 | finally: |
| 218 | client.logout() |
| 219 | self.reap_server(server, thread) |
| 220 | |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 221 | @reap_threads |
| 222 | def test_connect(self): |
| 223 | with self.reaped_server(SimpleIMAPHandler) as server: |
| 224 | client = self.imap_class(*server.server_address) |
| 225 | client.shutdown() |
| 226 | |
| 227 | @reap_threads |
| 228 | def test_issue5949(self): |
| 229 | |
| 230 | class EOFHandler(socketserver.StreamRequestHandler): |
| 231 | def handle(self): |
| 232 | # EOF without sending a complete welcome message. |
| 233 | self.wfile.write(b'* OK') |
| 234 | |
| 235 | with self.reaped_server(EOFHandler) as server: |
| 236 | self.assertRaises(imaplib.IMAP4.abort, |
| 237 | self.imap_class, *server.server_address) |
| 238 | |
| 239 | @reap_threads |
| 240 | def test_line_termination(self): |
| 241 | |
| 242 | class BadNewlineHandler(SimpleIMAPHandler): |
| 243 | |
| 244 | def cmd_CAPABILITY(self, tag, args): |
| 245 | self._send(b'* CAPABILITY IMAP4rev1 AUTH\n') |
R David Murray | 774a39f | 2013-02-19 12:17:31 -0500 | [diff] [blame] | 246 | self._send_tagged(tag, 'OK', 'CAPABILITY completed') |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 247 | |
| 248 | with self.reaped_server(BadNewlineHandler) as server: |
| 249 | self.assertRaises(imaplib.IMAP4.abort, |
| 250 | self.imap_class, *server.server_address) |
| 251 | |
R David Murray | 774a39f | 2013-02-19 12:17:31 -0500 | [diff] [blame] | 252 | @reap_threads |
| 253 | def test_bad_auth_name(self): |
| 254 | |
| 255 | class MyServer(SimpleIMAPHandler): |
| 256 | |
| 257 | def cmd_AUTHENTICATE(self, tag, args): |
| 258 | self._send_tagged(tag, 'NO', 'unrecognized authentication ' |
| 259 | 'type {}'.format(args[0])) |
| 260 | |
| 261 | with self.reaped_pair(MyServer) as (server, client): |
| 262 | with self.assertRaises(imaplib.IMAP4.error): |
| 263 | client.authenticate('METHOD', lambda: 1) |
| 264 | |
| 265 | @reap_threads |
| 266 | def test_invalid_authentication(self): |
| 267 | |
| 268 | class MyServer(SimpleIMAPHandler): |
| 269 | |
| 270 | def cmd_AUTHENTICATE(self, tag, args): |
| 271 | self._send_textline('+') |
| 272 | self.response = yield |
| 273 | self._send_tagged(tag, 'NO', '[AUTHENTICATIONFAILED] invalid') |
| 274 | |
| 275 | with self.reaped_pair(MyServer) as (server, client): |
| 276 | with self.assertRaises(imaplib.IMAP4.error): |
| 277 | code, data = client.authenticate('MYAUTH', lambda x: b'fake') |
| 278 | |
| 279 | @reap_threads |
| 280 | def test_valid_authentication(self): |
| 281 | |
| 282 | class MyServer(SimpleIMAPHandler): |
| 283 | |
| 284 | def cmd_AUTHENTICATE(self, tag, args): |
| 285 | self._send_textline('+') |
| 286 | self.server.response = yield |
| 287 | self._send_tagged(tag, 'OK', 'FAKEAUTH successful') |
| 288 | |
| 289 | with self.reaped_pair(MyServer) as (server, client): |
| 290 | code, data = client.authenticate('MYAUTH', lambda x: b'fake') |
| 291 | self.assertEqual(code, 'OK') |
| 292 | self.assertEqual(server.response, |
| 293 | b'ZmFrZQ==\r\n') #b64 encoded 'fake' |
| 294 | |
| 295 | with self.reaped_pair(MyServer) as (server, client): |
| 296 | code, data = client.authenticate('MYAUTH', lambda x: 'fake') |
| 297 | self.assertEqual(code, 'OK') |
| 298 | self.assertEqual(server.response, |
| 299 | b'ZmFrZQ==\r\n') #b64 encoded 'fake' |
| 300 | |
| 301 | @reap_threads |
| 302 | def test_login_cram_md5(self): |
| 303 | |
| 304 | class AuthHandler(SimpleIMAPHandler): |
| 305 | |
| 306 | capabilities = 'LOGINDISABLED AUTH=CRAM-MD5' |
| 307 | |
| 308 | def cmd_AUTHENTICATE(self, tag, args): |
| 309 | self._send_textline('+ PDE4OTYuNjk3MTcwOTUyQHBvc3RvZmZpY2Uucm' |
| 310 | 'VzdG9uLm1jaS5uZXQ=') |
| 311 | r = yield |
| 312 | if r == b'dGltIGYxY2E2YmU0NjRiOWVmYTFjY2E2ZmZkNmNmMmQ5ZjMy\r\n': |
| 313 | self._send_tagged(tag, 'OK', 'CRAM-MD5 successful') |
| 314 | else: |
| 315 | self._send_tagged(tag, 'NO', 'No access') |
| 316 | |
| 317 | with self.reaped_pair(AuthHandler) as (server, client): |
| 318 | self.assertTrue('AUTH=CRAM-MD5' in client.capabilities) |
| 319 | ret, data = client.login_cram_md5("tim", "tanstaaftanstaaf") |
| 320 | self.assertEqual(ret, "OK") |
| 321 | |
| 322 | with self.reaped_pair(AuthHandler) as (server, client): |
| 323 | self.assertTrue('AUTH=CRAM-MD5' in client.capabilities) |
| 324 | ret, data = client.login_cram_md5("tim", b"tanstaaftanstaaf") |
| 325 | self.assertEqual(ret, "OK") |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 326 | |
| 327 | |
| 328 | class ThreadedNetworkedTests(BaseThreadedNetworkedTests): |
| 329 | |
| 330 | server_class = socketserver.TCPServer |
| 331 | imap_class = imaplib.IMAP4 |
| 332 | |
| 333 | |
| 334 | @unittest.skipUnless(ssl, "SSL not available") |
| 335 | class ThreadedNetworkedTestsSSL(BaseThreadedNetworkedTests): |
| 336 | |
| 337 | server_class = SecureTCPServer |
| 338 | imap_class = IMAP4_SSL |
| 339 | |
| 340 | |
Antoine Pitrou | b1436f1 | 2010-11-09 22:55:55 +0000 | [diff] [blame] | 341 | class RemoteIMAPTest(unittest.TestCase): |
| 342 | host = 'cyrus.andrew.cmu.edu' |
| 343 | port = 143 |
| 344 | username = 'anonymous' |
| 345 | password = 'pass' |
| 346 | imap_class = imaplib.IMAP4 |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 347 | |
Antoine Pitrou | b1436f1 | 2010-11-09 22:55:55 +0000 | [diff] [blame] | 348 | def setUp(self): |
| 349 | with transient_internet(self.host): |
| 350 | self.server = self.imap_class(self.host, self.port) |
| 351 | |
| 352 | def tearDown(self): |
| 353 | if self.server is not None: |
Antoine Pitrou | 924cbea | 2011-03-23 03:10:14 +0100 | [diff] [blame] | 354 | with transient_internet(self.host): |
| 355 | self.server.logout() |
Antoine Pitrou | b1436f1 | 2010-11-09 22:55:55 +0000 | [diff] [blame] | 356 | |
| 357 | def test_logincapa(self): |
Antoine Pitrou | 924cbea | 2011-03-23 03:10:14 +0100 | [diff] [blame] | 358 | with transient_internet(self.host): |
| 359 | for cap in self.server.capabilities: |
| 360 | self.assertIsInstance(cap, str) |
Nick Coghlan | e6ef462 | 2012-06-17 21:10:21 +1000 | [diff] [blame] | 361 | self.assertIn('LOGINDISABLED', self.server.capabilities) |
| 362 | self.assertIn('AUTH=ANONYMOUS', self.server.capabilities) |
Antoine Pitrou | 924cbea | 2011-03-23 03:10:14 +0100 | [diff] [blame] | 363 | rs = self.server.login(self.username, self.password) |
| 364 | self.assertEqual(rs[0], 'OK') |
Antoine Pitrou | b1436f1 | 2010-11-09 22:55:55 +0000 | [diff] [blame] | 365 | |
| 366 | def test_logout(self): |
Antoine Pitrou | 924cbea | 2011-03-23 03:10:14 +0100 | [diff] [blame] | 367 | with transient_internet(self.host): |
| 368 | rs = self.server.logout() |
| 369 | self.server = None |
| 370 | self.assertEqual(rs[0], 'BYE') |
Antoine Pitrou | b1436f1 | 2010-11-09 22:55:55 +0000 | [diff] [blame] | 371 | |
| 372 | |
| 373 | @unittest.skipUnless(ssl, "SSL not available") |
Antoine Pitrou | f3b001f | 2010-11-12 18:49:16 +0000 | [diff] [blame] | 374 | class RemoteIMAP_STARTTLSTest(RemoteIMAPTest): |
| 375 | |
| 376 | def setUp(self): |
| 377 | super().setUp() |
Antoine Pitrou | 924cbea | 2011-03-23 03:10:14 +0100 | [diff] [blame] | 378 | with transient_internet(self.host): |
| 379 | rs = self.server.starttls() |
| 380 | self.assertEqual(rs[0], 'OK') |
Antoine Pitrou | f3b001f | 2010-11-12 18:49:16 +0000 | [diff] [blame] | 381 | |
| 382 | def test_logincapa(self): |
Antoine Pitrou | dbe7519 | 2010-11-16 17:55:26 +0000 | [diff] [blame] | 383 | for cap in self.server.capabilities: |
| 384 | self.assertIsInstance(cap, str) |
Nick Coghlan | e6ef462 | 2012-06-17 21:10:21 +1000 | [diff] [blame] | 385 | self.assertNotIn('LOGINDISABLED', self.server.capabilities) |
Antoine Pitrou | f3b001f | 2010-11-12 18:49:16 +0000 | [diff] [blame] | 386 | |
| 387 | |
| 388 | @unittest.skipUnless(ssl, "SSL not available") |
Antoine Pitrou | b1436f1 | 2010-11-09 22:55:55 +0000 | [diff] [blame] | 389 | class RemoteIMAP_SSLTest(RemoteIMAPTest): |
| 390 | port = 993 |
| 391 | imap_class = IMAP4_SSL |
| 392 | |
Antoine Pitrou | 0872816 | 2011-05-06 18:49:52 +0200 | [diff] [blame] | 393 | def setUp(self): |
| 394 | pass |
| 395 | |
| 396 | def tearDown(self): |
| 397 | pass |
| 398 | |
| 399 | def create_ssl_context(self): |
| 400 | ssl_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23) |
| 401 | ssl_context.load_cert_chain(CERTFILE) |
| 402 | return ssl_context |
| 403 | |
| 404 | def check_logincapa(self, server): |
| 405 | try: |
| 406 | for cap in server.capabilities: |
| 407 | self.assertIsInstance(cap, str) |
Nick Coghlan | e51e25a | 2012-06-17 21:15:45 +1000 | [diff] [blame] | 408 | self.assertNotIn('LOGINDISABLED', server.capabilities) |
| 409 | self.assertIn('AUTH=PLAIN', server.capabilities) |
Antoine Pitrou | 0872816 | 2011-05-06 18:49:52 +0200 | [diff] [blame] | 410 | rs = server.login(self.username, self.password) |
| 411 | self.assertEqual(rs[0], 'OK') |
| 412 | finally: |
| 413 | server.logout() |
| 414 | |
Antoine Pitrou | b1436f1 | 2010-11-09 22:55:55 +0000 | [diff] [blame] | 415 | def test_logincapa(self): |
Antoine Pitrou | 0872816 | 2011-05-06 18:49:52 +0200 | [diff] [blame] | 416 | with transient_internet(self.host): |
| 417 | _server = self.imap_class(self.host, self.port) |
| 418 | self.check_logincapa(_server) |
| 419 | |
| 420 | def test_logincapa_with_client_certfile(self): |
| 421 | with transient_internet(self.host): |
| 422 | _server = self.imap_class(self.host, self.port, certfile=CERTFILE) |
| 423 | self.check_logincapa(_server) |
| 424 | |
| 425 | def test_logincapa_with_client_ssl_context(self): |
| 426 | with transient_internet(self.host): |
| 427 | _server = self.imap_class(self.host, self.port, ssl_context=self.create_ssl_context()) |
| 428 | self.check_logincapa(_server) |
| 429 | |
| 430 | def test_logout(self): |
| 431 | with transient_internet(self.host): |
| 432 | _server = self.imap_class(self.host, self.port) |
| 433 | rs = _server.logout() |
| 434 | self.assertEqual(rs[0], 'BYE') |
| 435 | |
| 436 | def test_ssl_context_certfile_exclusive(self): |
| 437 | with transient_internet(self.host): |
| 438 | self.assertRaises(ValueError, self.imap_class, self.host, self.port, |
| 439 | certfile=CERTFILE, ssl_context=self.create_ssl_context()) |
| 440 | |
| 441 | def test_ssl_context_keyfile_exclusive(self): |
| 442 | with transient_internet(self.host): |
| 443 | self.assertRaises(ValueError, self.imap_class, self.host, self.port, |
| 444 | keyfile=CERTFILE, ssl_context=self.create_ssl_context()) |
Antoine Pitrou | b1436f1 | 2010-11-09 22:55:55 +0000 | [diff] [blame] | 445 | |
| 446 | |
Ezio Melotti | 02bf701 | 2013-03-02 14:25:56 +0200 | [diff] [blame] | 447 | def load_tests(*args): |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 448 | tests = [TestImaplib] |
| 449 | |
| 450 | if support.is_resource_enabled('network'): |
| 451 | if ssl: |
| 452 | global CERTFILE |
| 453 | CERTFILE = os.path.join(os.path.dirname(__file__) or os.curdir, |
| 454 | "keycert.pem") |
| 455 | if not os.path.exists(CERTFILE): |
| 456 | raise support.TestFailed("Can't read certificate files!") |
Antoine Pitrou | b1436f1 | 2010-11-09 22:55:55 +0000 | [diff] [blame] | 457 | tests.extend([ |
| 458 | ThreadedNetworkedTests, ThreadedNetworkedTestsSSL, |
Antoine Pitrou | f3b001f | 2010-11-12 18:49:16 +0000 | [diff] [blame] | 459 | RemoteIMAPTest, RemoteIMAP_SSLTest, RemoteIMAP_STARTTLSTest, |
Antoine Pitrou | b1436f1 | 2010-11-09 22:55:55 +0000 | [diff] [blame] | 460 | ]) |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 461 | |
Ezio Melotti | 02bf701 | 2013-03-02 14:25:56 +0200 | [diff] [blame] | 462 | return unittest.TestSuite([unittest.makeSuite(test) for test in tests]) |
Christian Heimes | f6cd967 | 2008-03-26 13:45:42 +0000 | [diff] [blame] | 463 | |
| 464 | |
| 465 | if __name__ == "__main__": |
R. David Murray | e8dc258 | 2009-12-10 02:08:06 +0000 | [diff] [blame] | 466 | support.use_resources = ['network'] |
Ezio Melotti | 02bf701 | 2013-03-02 14:25:56 +0200 | [diff] [blame] | 467 | unittest.main() |