Changeset 391 for python/trunk/Lib/test/test_imaplib.py

Timestamp:

Mar 19, 2014, 11:31:01 PM (11 years ago)

Author:

dmik

Message:

python: Merge vendor 2.7.6 to trunk.

Location:

python/trunk

Files:

• . (modified) (1 prop)
• Lib/test/test_imaplib.py (modified) (10 diffs)

python/trunk/Lib/test/test_imaplib.py

@@ -8 +8 @@
 import imaplib
 import os.path
-import select
-import socket
 import SocketServer
-import sys
 import time
 
-from test_support import verbose
+from test_support import reap_threads, verbose, transient_internet
 import unittest
 
@@ -58 +55 @@
     IMAP4_SSL = None
 
-class TimeoutStreamRequestHandler(SocketServer.StreamRequestHandler):
+
+class SimpleIMAPHandler(SocketServer.StreamRequestHandler):
 
     timeout = 1
-
-    def setup(self):
-        self.connection = self.request
-        if self.timeout is not None:
-            self.connection.settimeout(self.timeout)
-        self.rfile = self.connection.makefile('rb', self.rbufsize)
-        self.wfile = self.connection.makefile('wb', self.wbufsize)
-
-
-class SimpleIMAPHandler(TimeoutStreamRequestHandler):
 
     def _send(self, message):
@@ -92 +80 @@
                     line += part
                 except IOError:
-                    # ..but SSLSockets throw exceptions.
+                    # ..but SSLSockets raise exceptions.
                     return
                 if line.endswith('\r\n'):
@@ -112 +100 @@
         self._send('%s OK CAPABILITY completed\r\n' % (tag,))
 
+
 class BaseThreadedNetworkedTests(unittest.TestCase):
 
@@ -124 +113 @@
         if verbose: print "creating server"
         server = MyServer(addr, hdlr)
-        self.assertEquals(server.server_address, server.socket.getsockname())
+        self.assertEqual(server.server_address, server.socket.getsockname())
 
         if verbose:
@@ -158 +147 @@
             self.reap_server(server, thread)
 
+    @reap_threads
     def test_connect(self):
         with self.reaped_server(SimpleIMAPHandler) as server:
@@ -163 +153 @@
             client.shutdown()
 
+    @reap_threads
     def test_issue5949(self):
 
-        class EOFHandler(TimeoutStreamRequestHandler):
+        class EOFHandler(SocketServer.StreamRequestHandler):
             def handle(self):
                 # EOF without sending a complete welcome message.
                 self.wfile.write('* OK')
-                # explicitly shutdown.  socket.close() merely releases
-                # the socket and waits for GC to perform the actual close.
-                self.request.shutdown(socket.SHUT_WR)
 
         with self.reaped_server(EOFHandler) as server:
@@ -177 +165 @@
                               self.imap_class, *server.server_address)
 
+
 class ThreadedNetworkedTests(BaseThreadedNetworkedTests):
 
@@ -183 +172 @@
 
 
+@unittest.skipUnless(ssl, "SSL not available")
 class ThreadedNetworkedTestsSSL(BaseThreadedNetworkedTests):
 
     server_class = SecureTCPServer
     imap_class = IMAP4_SSL
+
+
+class RemoteIMAPTest(unittest.TestCase):
+    host = 'cyrus.andrew.cmu.edu'
+    port = 143
+    username = 'anonymous'
+    password = 'pass'
+    imap_class = imaplib.IMAP4
+
+    def setUp(self):
+        with transient_internet(self.host):
+            self.server = self.imap_class(self.host, self.port)
+
+    def tearDown(self):
+        if self.server is not None:
+            self.server.logout()
+
+    def test_logincapa(self):
+        self.assertTrue('LOGINDISABLED' in self.server.capabilities)
+
+    def test_anonlogin(self):
+        self.assertTrue('AUTH=ANONYMOUS' in self.server.capabilities)
+        rs = self.server.login(self.username, self.password)
+        self.assertEqual(rs[0], 'OK')
+
+    def test_logout(self):
+        rs = self.server.logout()
+        self.server = None
+        self.assertEqual(rs[0], 'BYE')
+
+
+@unittest.skipUnless(ssl, "SSL not available")
+class RemoteIMAP_SSLTest(RemoteIMAPTest):
+    port = 993
+    imap_class = IMAP4_SSL
+
+    def test_logincapa(self):
+        self.assertFalse('LOGINDISABLED' in self.server.capabilities)
+        self.assertTrue('AUTH=PLAIN' in self.server.capabilities)
 
 
@@ -199 +228 @@
             if not os.path.exists(CERTFILE):
                 raise support.TestFailed("Can't read certificate files!")
-            tests.append(ThreadedNetworkedTestsSSL)
-        tests.append(ThreadedNetworkedTests)
-
-    threadinfo = support.threading_setup()
+        tests.extend([
+            ThreadedNetworkedTests, ThreadedNetworkedTestsSSL,
+            RemoteIMAPTest, RemoteIMAP_SSLTest,
+        ])
 
     support.run_unittest(*tests)
 
-    support.threading_cleanup(*threadinfo)
 
 if __name__ == "__main__":