Changeset 391 for python/trunk/Lib/test/test_smtplib.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_smtplib.py (modified) (18 diffs)

python/trunk/Lib/test/test_smtplib.py

@@ -2 +2 @@
 import email.utils
 import socket
-import threading
 import smtpd
 import smtplib
@@ -10 +9 @@
 import select
 
-from unittest import TestCase
+import unittest
 from test import test_support
+
+try:
+    import threading
+except ImportError:
+    threading = None
 
 HOST = test_support.HOST
@@ -37 +41 @@
         evt.set()
 
-class GeneralTests(TestCase):
+@unittest.skipUnless(threading, 'Threading required for this test.')
+class GeneralTests(unittest.TestCase):
 
     def setUp(self):
+        self._threads = test_support.threading_setup()
         self.evt = threading.Event()
         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -45 +51 @@
         self.port = test_support.bind_port(self.sock)
         servargs = (self.evt, "220 Hola mundo\n", self.sock)
-        threading.Thread(target=server, args=servargs).start()
+        self.thread = threading.Thread(target=server, args=servargs)
+        self.thread.start()
         self.evt.wait()
         self.evt.clear()
@@ -51 +58 @@
     def tearDown(self):
         self.evt.wait()
+        self.thread.join()
+        test_support.threading_cleanup(*self._threads)
 
     def testBasic1(self):
@@ -135 +144 @@
 
 # Test behavior of smtpd.DebuggingServer
-class DebuggingServerTests(TestCase):
+@unittest.skipUnless(threading, 'Threading required for this test.')
+class DebuggingServerTests(unittest.TestCase):
 
     def setUp(self):
@@ -143 +153 @@
         sys.stdout = self.output
 
+        self._threads = test_support.threading_setup()
         self.serv_evt = threading.Event()
         self.client_evt = threading.Event()
-        self.port = test_support.find_unused_port()
-        self.serv = smtpd.DebuggingServer((HOST, self.port), ('nowhere', -1))
+        # Pick a random unused port by passing 0 for the port number
+        self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1))
+        # Keep a note of what port was assigned
+        self.port = self.serv.socket.getsockname()[1]
         serv_args = (self.serv, self.serv_evt, self.client_evt)
-        threading.Thread(target=debugging_server, args=serv_args).start()
+        self.thread = threading.Thread(target=debugging_server, args=serv_args)
+        self.thread.start()
 
         # wait until server thread has assigned a port number
@@ -159 +173 @@
         # wait for the server thread to terminate
         self.serv_evt.wait()
+        self.thread.join()
+        test_support.threading_cleanup(*self._threads)
         # restore sys.stdout
         sys.stdout = self.old_stdout
@@ -226 +242 @@
 
 
-class NonConnectingTests(TestCase):
+class NonConnectingTests(unittest.TestCase):
 
     def testNotConnected(self):
@@ -247 +263 @@
 
 # test response of client to a non-successful HELO message
-class BadHELOServerTests(TestCase):
+@unittest.skipUnless(threading, 'Threading required for this test.')
+class BadHELOServerTests(unittest.TestCase):
 
     def setUp(self):
@@ -254 +271 @@
         sys.stdout = self.output
 
+        self._threads = test_support.threading_setup()
         self.evt = threading.Event()
         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -259 +277 @@
         self.port = test_support.bind_port(self.sock)
         servargs = (self.evt, "199 no hello for you!\n", self.sock)
-        threading.Thread(target=server, args=servargs).start()
+        self.thread = threading.Thread(target=server, args=servargs)
+        self.thread.start()
         self.evt.wait()
         self.evt.clear()
@@ -265 +284 @@
     def tearDown(self):
         self.evt.wait()
+        self.thread.join()
+        test_support.threading_cleanup(*self._threads)
         sys.stdout = self.old_stdout
 
@@ -310 +331 @@
 
     def smtp_VRFY(self, arg):
-        raw_addr = email.utils.parseaddr(arg)[1]
-        quoted_addr = smtplib.quoteaddr(arg)
-        if raw_addr in sim_users:
-            self.push('250 %s %s' % (sim_users[raw_addr], quoted_addr))
+        # For max compatibility smtplib should be sending the raw address.
+        if arg in sim_users:
+            self.push('250 %s %s' % (sim_users[arg], smtplib.quoteaddr(arg)))
         else:
             self.push('550 No such user: %s' % arg)
 
     def smtp_EXPN(self, arg):
-        list_name = email.utils.parseaddr(arg)[1].lower()
+        list_name = arg.lower()
         if list_name in sim_lists:
             user_list = sim_lists[list_name]
@@ -346 +366 @@
             self.push('550 No access for you!')
 
+    def handle_error(self):
+        raise
+
 
 class SimSMTPServer(smtpd.SMTPServer):
@@ -364 +387 @@
         self._extra_features.append(feature)
 
+    def handle_error(self):
+        raise
+
 
 # Test various SMTP & ESMTP commands/behaviors that require a simulated server
 # (i.e., something with more features than DebuggingServer)
-class SMTPSimTests(TestCase):
+@unittest.skipUnless(threading, 'Threading required for this test.')
+class SMTPSimTests(unittest.TestCase):
 
     def setUp(self):
+        self._threads = test_support.threading_setup()
         self.serv_evt = threading.Event()
         self.client_evt = threading.Event()
-        self.port = test_support.find_unused_port()
-        self.serv = SimSMTPServer((HOST, self.port), ('nowhere', -1))
+        # Pick a random unused port by passing 0 for the port number
+        self.serv = SimSMTPServer((HOST, 0), ('nowhere', -1))
+        # Keep a note of what port was assigned
+        self.port = self.serv.socket.getsockname()[1]
         serv_args = (self.serv, self.serv_evt, self.client_evt)
-        threading.Thread(target=debugging_server, args=serv_args).start()
+        self.thread = threading.Thread(target=debugging_server, args=serv_args)
+        self.thread.start()
 
         # wait until server thread has assigned a port number
@@ -386 +417 @@
         # wait for the server thread to terminate
         self.serv_evt.wait()
+        self.thread.join()
+        test_support.threading_cleanup(*self._threads)
 
     def testBasic(self):
@@ -421 +454 @@
 
         u = 'nobody@nowhere.com'
-        expected_unknown = (550, 'No such user: %s' % smtplib.quoteaddr(u))
+        expected_unknown = (550, 'No such user: %s' % u)
         self.assertEqual(smtp.vrfy(u), expected_unknown)
         smtp.quit()