aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/Lib/test/test_smtplib.py
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test/test_smtplib.py')
-rw-r--r--Lib/test/test_smtplib.py375
1 files changed, 312 insertions, 63 deletions
diff --git a/Lib/test/test_smtplib.py b/Lib/test/test_smtplib.py
index 81806c9898d..2cb0d1ab52b 100644
--- a/Lib/test/test_smtplib.py
+++ b/Lib/test/test_smtplib.py
@@ -1,22 +1,32 @@
import asyncore
+import email.mime.text
import email.utils
import socket
import smtpd
import smtplib
-import StringIO
+import io
+import re
import sys
import time
import select
import unittest
-from test import test_support
+from test import support, mock_socket
try:
import threading
except ImportError:
threading = None
-HOST = test_support.HOST
+HOST = support.HOST
+
+if sys.platform == 'darwin':
+ # select.poll returns a select.POLLHUP at the end of the tests
+ # on darwin, so just ignore it
+ def handle_expt(self):
+ pass
+ smtpd.SMTPChannel.handle_expt = handle_expt
+
def server(evt, buf, serv):
serv.listen(5)
@@ -40,53 +50,55 @@ def server(evt, buf, serv):
serv.close()
evt.set()
-@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)
- self.sock.settimeout(15)
- self.port = test_support.bind_port(self.sock)
- servargs = (self.evt, "220 Hola mundo\n", self.sock)
- self.thread = threading.Thread(target=server, args=servargs)
- self.thread.start()
- self.evt.wait()
- self.evt.clear()
+ smtplib.socket = mock_socket
+ self.port = 25
def tearDown(self):
- self.evt.wait()
- self.thread.join()
- test_support.threading_cleanup(*self._threads)
+ smtplib.socket = socket
+
+ # This method is no longer used but is retained for backward compatibility,
+ # so test to make sure it still works.
+ def testQuoteData(self):
+ teststr = "abc\n.jkl\rfoo\r\n..blue"
+ expected = "abc\r\n..jkl\r\nfoo\r\n...blue"
+ self.assertEqual(expected, smtplib.quotedata(teststr))
def testBasic1(self):
+ mock_socket.reply_with(b"220 Hola mundo")
# connects
smtp = smtplib.SMTP(HOST, self.port)
smtp.close()
def testBasic2(self):
+ mock_socket.reply_with(b"220 Hola mundo")
# connects, include port in host name
smtp = smtplib.SMTP("%s:%s" % (HOST, self.port))
smtp.close()
def testLocalHostName(self):
+ mock_socket.reply_with(b"220 Hola mundo")
# check that supplied local_hostname is used
smtp = smtplib.SMTP(HOST, self.port, local_hostname="testhost")
self.assertEqual(smtp.local_hostname, "testhost")
smtp.close()
def testTimeoutDefault(self):
- self.assertTrue(socket.getdefaulttimeout() is None)
- socket.setdefaulttimeout(30)
+ mock_socket.reply_with(b"220 Hola mundo")
+ self.assertTrue(mock_socket.getdefaulttimeout() is None)
+ mock_socket.setdefaulttimeout(30)
+ self.assertEqual(mock_socket.getdefaulttimeout(), 30)
try:
smtp = smtplib.SMTP(HOST, self.port)
finally:
- socket.setdefaulttimeout(None)
+ mock_socket.setdefaulttimeout(None)
self.assertEqual(smtp.sock.gettimeout(), 30)
smtp.close()
def testTimeoutNone(self):
+ mock_socket.reply_with(b"220 Hola mundo")
self.assertTrue(socket.getdefaulttimeout() is None)
socket.setdefaulttimeout(30)
try:
@@ -97,6 +109,7 @@ class GeneralTests(unittest.TestCase):
smtp.close()
def testTimeoutValue(self):
+ mock_socket.reply_with(b"220 Hola mundo")
smtp = smtplib.SMTP(HOST, self.port, timeout=30)
self.assertEqual(smtp.sock.gettimeout(), 30)
smtp.close()
@@ -146,15 +159,21 @@ MSG_END = '------------ END MESSAGE ------------\n'
@unittest.skipUnless(threading, 'Threading required for this test.')
class DebuggingServerTests(unittest.TestCase):
+ maxDiff = None
+
def setUp(self):
+ self.real_getfqdn = socket.getfqdn
+ socket.getfqdn = mock_socket.getfqdn
# temporarily replace sys.stdout to capture DebuggingServer output
self.old_stdout = sys.stdout
- self.output = StringIO.StringIO()
+ self.output = io.StringIO()
sys.stdout = self.output
- self._threads = test_support.threading_setup()
self.serv_evt = threading.Event()
self.client_evt = threading.Event()
+ # Capture SMTPChannel debug output
+ self.old_DEBUGSTREAM = smtpd.DEBUGSTREAM
+ smtpd.DEBUGSTREAM = io.StringIO()
# 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
@@ -168,14 +187,17 @@ class DebuggingServerTests(unittest.TestCase):
self.serv_evt.clear()
def tearDown(self):
+ socket.getfqdn = self.real_getfqdn
# indicate that the client is finished
self.client_evt.set()
# 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
+ # restore DEBUGSTREAM
+ smtpd.DEBUGSTREAM.close()
+ smtpd.DEBUGSTREAM = self.old_DEBUGSTREAM
def testBasic(self):
# connect
@@ -184,27 +206,27 @@ class DebuggingServerTests(unittest.TestCase):
def testNOOP(self):
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
- expected = (250, 'Ok')
+ expected = (250, b'Ok')
self.assertEqual(smtp.noop(), expected)
smtp.quit()
def testRSET(self):
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
- expected = (250, 'Ok')
+ expected = (250, b'Ok')
self.assertEqual(smtp.rset(), expected)
smtp.quit()
def testNotImplemented(self):
# EHLO isn't implemented in DebuggingServer
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
- expected = (502, 'Error: command "EHLO" not implemented')
+ expected = (502, b'Error: command "EHLO" not implemented')
self.assertEqual(smtp.ehlo(), expected)
smtp.quit()
def testVRFY(self):
# VRFY isn't implemented in DebuggingServer
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
- expected = (502, 'Error: command "VRFY" not implemented')
+ expected = (502, b'Error: command "VRFY" not implemented')
self.assertEqual(smtp.vrfy('nobody@nowhere.com'), expected)
self.assertEqual(smtp.verify('nobody@nowhere.com'), expected)
smtp.quit()
@@ -214,13 +236,13 @@ class DebuggingServerTests(unittest.TestCase):
# (this behavior is specific to smtpd.SMTPChannel)
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
smtp.helo()
- expected = (503, 'Duplicate HELO/EHLO')
+ expected = (503, b'Duplicate HELO/EHLO')
self.assertEqual(smtp.helo(), expected)
smtp.quit()
def testHELP(self):
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
- self.assertEqual(smtp.help(), 'Error: command "HELP" not implemented')
+ self.assertEqual(smtp.help(), b'Error: command "HELP" not implemented')
smtp.quit()
def testSend(self):
@@ -240,9 +262,242 @@ class DebuggingServerTests(unittest.TestCase):
mexpect = '%s%s\n%s' % (MSG_BEGIN, m, MSG_END)
self.assertEqual(self.output.getvalue(), mexpect)
+ def testSendBinary(self):
+ m = b'A test message'
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ smtp.sendmail('John', 'Sally', m)
+ # XXX (see comment in testSend)
+ time.sleep(0.01)
+ smtp.quit()
+
+ self.client_evt.set()
+ self.serv_evt.wait()
+ self.output.flush()
+ mexpect = '%s%s\n%s' % (MSG_BEGIN, m.decode('ascii'), MSG_END)
+ self.assertEqual(self.output.getvalue(), mexpect)
+
+ def testSendNeedingDotQuote(self):
+ # Issue 12283
+ m = '.A test\n.mes.sage.'
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ smtp.sendmail('John', 'Sally', m)
+ # XXX (see comment in testSend)
+ time.sleep(0.01)
+ smtp.quit()
+
+ self.client_evt.set()
+ self.serv_evt.wait()
+ self.output.flush()
+ mexpect = '%s%s\n%s' % (MSG_BEGIN, m, MSG_END)
+ self.assertEqual(self.output.getvalue(), mexpect)
+
+ def testSendNullSender(self):
+ m = 'A test message'
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ smtp.sendmail('<>', 'Sally', m)
+ # XXX (see comment in testSend)
+ time.sleep(0.01)
+ smtp.quit()
+
+ self.client_evt.set()
+ self.serv_evt.wait()
+ self.output.flush()
+ mexpect = '%s%s\n%s' % (MSG_BEGIN, m, MSG_END)
+ self.assertEqual(self.output.getvalue(), mexpect)
+ debugout = smtpd.DEBUGSTREAM.getvalue()
+ sender = re.compile("^sender: <>$", re.MULTILINE)
+ self.assertRegex(debugout, sender)
+
+ def testSendMessage(self):
+ m = email.mime.text.MIMEText('A test message')
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ smtp.send_message(m, from_addr='John', to_addrs='Sally')
+ # XXX (see comment in testSend)
+ time.sleep(0.01)
+ smtp.quit()
+
+ self.client_evt.set()
+ self.serv_evt.wait()
+ self.output.flush()
+ # Add the X-Peer header that DebuggingServer adds
+ m['X-Peer'] = socket.gethostbyname('localhost')
+ mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
+ self.assertEqual(self.output.getvalue(), mexpect)
+
+ def testSendMessageWithAddresses(self):
+ m = email.mime.text.MIMEText('A test message')
+ m['From'] = 'foo@bar.com'
+ m['To'] = 'John'
+ m['CC'] = 'Sally, Fred'
+ m['Bcc'] = 'John Root <root@localhost>, "Dinsdale" <warped@silly.walks.com>'
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ smtp.send_message(m)
+ # XXX (see comment in testSend)
+ time.sleep(0.01)
+ smtp.quit()
+ # make sure the Bcc header is still in the message.
+ self.assertEqual(m['Bcc'], 'John Root <root@localhost>, "Dinsdale" '
+ '<warped@silly.walks.com>')
+
+ self.client_evt.set()
+ self.serv_evt.wait()
+ self.output.flush()
+ # Add the X-Peer header that DebuggingServer adds
+ m['X-Peer'] = socket.gethostbyname('localhost')
+ # The Bcc header should not be transmitted.
+ del m['Bcc']
+ mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
+ self.assertEqual(self.output.getvalue(), mexpect)
+ debugout = smtpd.DEBUGSTREAM.getvalue()
+ sender = re.compile("^sender: foo@bar.com$", re.MULTILINE)
+ self.assertRegex(debugout, sender)
+ for addr in ('John', 'Sally', 'Fred', 'root@localhost',
+ 'warped@silly.walks.com'):
+ to_addr = re.compile(r"^recips: .*'{}'.*$".format(addr),
+ re.MULTILINE)
+ self.assertRegex(debugout, to_addr)
+
+ def testSendMessageWithSomeAddresses(self):
+ # Make sure nothing breaks if not all of the three 'to' headers exist
+ m = email.mime.text.MIMEText('A test message')
+ m['From'] = 'foo@bar.com'
+ m['To'] = 'John, Dinsdale'
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ smtp.send_message(m)
+ # XXX (see comment in testSend)
+ time.sleep(0.01)
+ smtp.quit()
+
+ self.client_evt.set()
+ self.serv_evt.wait()
+ self.output.flush()
+ # Add the X-Peer header that DebuggingServer adds
+ m['X-Peer'] = socket.gethostbyname('localhost')
+ mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
+ self.assertEqual(self.output.getvalue(), mexpect)
+ debugout = smtpd.DEBUGSTREAM.getvalue()
+ sender = re.compile("^sender: foo@bar.com$", re.MULTILINE)
+ self.assertRegex(debugout, sender)
+ for addr in ('John', 'Dinsdale'):
+ to_addr = re.compile(r"^recips: .*'{}'.*$".format(addr),
+ re.MULTILINE)
+ self.assertRegex(debugout, to_addr)
+
+ def testSendMessageWithSpecifiedAddresses(self):
+ # Make sure addresses specified in call override those in message.
+ m = email.mime.text.MIMEText('A test message')
+ m['From'] = 'foo@bar.com'
+ m['To'] = 'John, Dinsdale'
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ smtp.send_message(m, from_addr='joe@example.com', to_addrs='foo@example.net')
+ # XXX (see comment in testSend)
+ time.sleep(0.01)
+ smtp.quit()
+
+ self.client_evt.set()
+ self.serv_evt.wait()
+ self.output.flush()
+ # Add the X-Peer header that DebuggingServer adds
+ m['X-Peer'] = socket.gethostbyname('localhost')
+ mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
+ self.assertEqual(self.output.getvalue(), mexpect)
+ debugout = smtpd.DEBUGSTREAM.getvalue()
+ sender = re.compile("^sender: joe@example.com$", re.MULTILINE)
+ self.assertRegex(debugout, sender)
+ for addr in ('John', 'Dinsdale'):
+ to_addr = re.compile(r"^recips: .*'{}'.*$".format(addr),
+ re.MULTILINE)
+ self.assertNotRegex(debugout, to_addr)
+ recip = re.compile(r"^recips: .*'foo@example.net'.*$", re.MULTILINE)
+ self.assertRegex(debugout, recip)
+
+ def testSendMessageWithMultipleFrom(self):
+ # Sender overrides To
+ m = email.mime.text.MIMEText('A test message')
+ m['From'] = 'Bernard, Bianca'
+ m['Sender'] = 'the_rescuers@Rescue-Aid-Society.com'
+ m['To'] = 'John, Dinsdale'
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ smtp.send_message(m)
+ # XXX (see comment in testSend)
+ time.sleep(0.01)
+ smtp.quit()
+
+ self.client_evt.set()
+ self.serv_evt.wait()
+ self.output.flush()
+ # Add the X-Peer header that DebuggingServer adds
+ m['X-Peer'] = socket.gethostbyname('localhost')
+ mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
+ self.assertEqual(self.output.getvalue(), mexpect)
+ debugout = smtpd.DEBUGSTREAM.getvalue()
+ sender = re.compile("^sender: the_rescuers@Rescue-Aid-Society.com$", re.MULTILINE)
+ self.assertRegex(debugout, sender)
+ for addr in ('John', 'Dinsdale'):
+ to_addr = re.compile(r"^recips: .*'{}'.*$".format(addr),
+ re.MULTILINE)
+ self.assertRegex(debugout, to_addr)
+
+ def testSendMessageResent(self):
+ m = email.mime.text.MIMEText('A test message')
+ m['From'] = 'foo@bar.com'
+ m['To'] = 'John'
+ m['CC'] = 'Sally, Fred'
+ m['Bcc'] = 'John Root <root@localhost>, "Dinsdale" <warped@silly.walks.com>'
+ m['Resent-Date'] = 'Thu, 1 Jan 1970 17:42:00 +0000'
+ m['Resent-From'] = 'holy@grail.net'
+ m['Resent-To'] = 'Martha <my_mom@great.cooker.com>, Jeff'
+ m['Resent-Bcc'] = 'doe@losthope.net'
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ smtp.send_message(m)
+ # XXX (see comment in testSend)
+ time.sleep(0.01)
+ smtp.quit()
+
+ self.client_evt.set()
+ self.serv_evt.wait()
+ self.output.flush()
+ # The Resent-Bcc headers are deleted before serialization.
+ del m['Bcc']
+ del m['Resent-Bcc']
+ # Add the X-Peer header that DebuggingServer adds
+ m['X-Peer'] = socket.gethostbyname('localhost')
+ mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
+ self.assertEqual(self.output.getvalue(), mexpect)
+ debugout = smtpd.DEBUGSTREAM.getvalue()
+ sender = re.compile("^sender: holy@grail.net$", re.MULTILINE)
+ self.assertRegex(debugout, sender)
+ for addr in ('my_mom@great.cooker.com', 'Jeff', 'doe@losthope.net'):
+ to_addr = re.compile(r"^recips: .*'{}'.*$".format(addr),
+ re.MULTILINE)
+ self.assertRegex(debugout, to_addr)
+
+ def testSendMessageMultipleResentRaises(self):
+ m = email.mime.text.MIMEText('A test message')
+ m['From'] = 'foo@bar.com'
+ m['To'] = 'John'
+ m['CC'] = 'Sally, Fred'
+ m['Bcc'] = 'John Root <root@localhost>, "Dinsdale" <warped@silly.walks.com>'
+ m['Resent-Date'] = 'Thu, 1 Jan 1970 17:42:00 +0000'
+ m['Resent-From'] = 'holy@grail.net'
+ m['Resent-To'] = 'Martha <my_mom@great.cooker.com>, Jeff'
+ m['Resent-Bcc'] = 'doe@losthope.net'
+ m['Resent-Date'] = 'Thu, 2 Jan 1970 17:42:00 +0000'
+ m['Resent-To'] = 'holy@grail.net'
+ m['Resent-From'] = 'Martha <my_mom@great.cooker.com>, Jeff'
+ smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
+ with self.assertRaises(ValueError):
+ smtp.send_message(m)
+ smtp.close()
class NonConnectingTests(unittest.TestCase):
+ def setUp(self):
+ smtplib.socket = mock_socket
+
+ def tearDown(self):
+ smtplib.socket = socket
+
def testNotConnected(self):
# Test various operations on an unconnected SMTP object that
# should raise exceptions (at present the attempt in SMTP.send
@@ -255,9 +510,9 @@ class NonConnectingTests(unittest.TestCase):
def testNonnumericPort(self):
# check that non-numeric port raises socket.error
- self.assertRaises(socket.error, smtplib.SMTP,
+ self.assertRaises(mock_socket.error, smtplib.SMTP,
"localhost", "bogus")
- self.assertRaises(socket.error, smtplib.SMTP,
+ self.assertRaises(mock_socket.error, smtplib.SMTP,
"localhost:bogus")
@@ -266,25 +521,15 @@ class NonConnectingTests(unittest.TestCase):
class BadHELOServerTests(unittest.TestCase):
def setUp(self):
+ smtplib.socket = mock_socket
+ mock_socket.reply_with(b"199 no hello for you!")
self.old_stdout = sys.stdout
- self.output = StringIO.StringIO()
+ self.output = io.StringIO()
sys.stdout = self.output
-
- self._threads = test_support.threading_setup()
- self.evt = threading.Event()
- self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self.sock.settimeout(15)
- self.port = test_support.bind_port(self.sock)
- servargs = (self.evt, "199 no hello for you!\n", self.sock)
- self.thread = threading.Thread(target=server, args=servargs)
- self.thread.start()
- self.evt.wait()
- self.evt.clear()
+ self.port = 25
def tearDown(self):
- self.evt.wait()
- self.thread.join()
- test_support.threading_cleanup(*self._threads)
+ smtplib.socket = socket
sys.stdout = self.old_stdout
def testFailingHELO(self):
@@ -293,7 +538,7 @@ class BadHELOServerTests(unittest.TestCase):
sim_users = {'Mr.A@somewhere.com':'John A',
- 'Ms.B@somewhere.com':'Sally B',
+ 'Ms.B@xn--fo-fka.com':'Sally B',
'Mrs.C@somewhereesle.com':'Ruth C',
}
@@ -309,7 +554,7 @@ sim_auth_credentials = {
sim_auth_login_password = 'C29TZXBHC3N3B3JK'
sim_lists = {'list-1':['Mr.A@somewhere.com','Mrs.C@somewhereesle.com'],
- 'list-2':['Ms.B@somewhere.com',],
+ 'list-2':['Ms.B@xn--fo-fka.com',],
}
# Simulated SMTP channel & server
@@ -318,7 +563,7 @@ class SimSMTPChannel(smtpd.SMTPChannel):
def __init__(self, extra_features, *args, **kw):
self._extrafeatures = ''.join(
[ "250-{0}\r\n".format(x) for x in extra_features ])
- smtpd.SMTPChannel.__init__(self, *args, **kw)
+ super(SimSMTPChannel, self).__init__(*args, **kw)
def smtp_EHLO(self, arg):
resp = ('250-testhost\r\n'
@@ -351,7 +596,7 @@ class SimSMTPChannel(smtpd.SMTPChannel):
def smtp_AUTH(self, arg):
if arg.strip().lower()=='cram-md5':
- self.push('334 {0}'.format(sim_cram_md5_challenge))
+ self.push('334 {}'.format(sim_cram_md5_challenge))
return
mech, auth = arg.split()
mech = mech.lower()
@@ -375,8 +620,7 @@ class SimSMTPServer(smtpd.SMTPServer):
self._extra_features = []
smtpd.SMTPServer.__init__(self, *args, **kw)
- def handle_accept(self):
- conn, addr = self.accept()
+ def handle_accepted(self, conn, addr):
self._SMTPchannel = SimSMTPChannel(self._extra_features,
self, conn, addr)
@@ -396,7 +640,8 @@ class SimSMTPServer(smtpd.SMTPServer):
class SMTPSimTests(unittest.TestCase):
def setUp(self):
- self._threads = test_support.threading_setup()
+ self.real_getfqdn = socket.getfqdn
+ socket.getfqdn = mock_socket.getfqdn
self.serv_evt = threading.Event()
self.client_evt = threading.Event()
# Pick a random unused port by passing 0 for the port number
@@ -412,12 +657,12 @@ class SMTPSimTests(unittest.TestCase):
self.serv_evt.clear()
def tearDown(self):
+ socket.getfqdn = self.real_getfqdn
# indicate that the client is finished
self.client_evt.set()
# wait for the server thread to terminate
self.serv_evt.wait()
self.thread.join()
- test_support.threading_cleanup(*self._threads)
def testBasic(self):
# smoke test
@@ -449,11 +694,13 @@ class SMTPSimTests(unittest.TestCase):
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
for email, name in sim_users.items():
- expected_known = (250, '%s %s' % (name, smtplib.quoteaddr(email)))
+ expected_known = (250, bytes('%s %s' %
+ (name, smtplib.quoteaddr(email)),
+ "ascii"))
self.assertEqual(smtp.vrfy(email), expected_known)
u = 'nobody@nowhere.com'
- expected_unknown = (550, 'No such user: %s' % u)
+ expected_unknown = (550, ('No such user: %s' % u).encode('ascii'))
self.assertEqual(smtp.vrfy(u), expected_unknown)
smtp.quit()
@@ -464,11 +711,11 @@ class SMTPSimTests(unittest.TestCase):
users = []
for m in members:
users.append('%s %s' % (sim_users[m], smtplib.quoteaddr(m)))
- expected_known = (250, '\n'.join(users))
+ expected_known = (250, bytes('\n'.join(users), "ascii"))
self.assertEqual(smtp.expn(listname), expected_known)
u = 'PSU-Members-List'
- expected_unknown = (550, 'No access for you!')
+ expected_unknown = (550, b'No access for you!')
self.assertEqual(smtp.expn(u), expected_unknown)
smtp.quit()
@@ -478,6 +725,7 @@ class SMTPSimTests(unittest.TestCase):
expected_auth_ok = (235, b'plain auth ok')
self.assertEqual(smtp.login(sim_auth[0], sim_auth[1]), expected_auth_ok)
+ smtp.close()
# SimSMTPChannel doesn't fully support LOGIN or CRAM-MD5 auth because they
# require a synchronous read to obtain the credentials...so instead smtpd
@@ -492,8 +740,8 @@ class SMTPSimTests(unittest.TestCase):
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
try: smtp.login(sim_auth[0], sim_auth[1])
except smtplib.SMTPAuthenticationError as err:
- if sim_auth_login_password not in str(err):
- raise "expected encoded password not found in error message"
+ self.assertIn(sim_auth_login_password, str(err))
+ smtp.close()
def testAUTH_CRAM_MD5(self):
self.serv.add_feature("AUTH CRAM-MD5")
@@ -501,15 +749,16 @@ class SMTPSimTests(unittest.TestCase):
try: smtp.login(sim_auth[0], sim_auth[1])
except smtplib.SMTPAuthenticationError as err:
- if sim_auth_credentials['cram-md5'] not in str(err):
- raise "expected encoded credentials not found in error message"
+ self.assertIn(sim_auth_credentials['cram-md5'], str(err))
+ smtp.close()
#TODO: add tests for correct AUTH method fallback now that the
#test infrastructure can support it.
+@support.reap_threads
def test_main(verbose=None):
- test_support.run_unittest(GeneralTests, DebuggingServerTests,
+ support.run_unittest(GeneralTests, DebuggingServerTests,
NonConnectingTests,
BadHELOServerTests, SMTPSimTests)