diff options
Diffstat (limited to 'Lib/test/test_smtplib.py')
| -rw-r--r-- | Lib/test/test_smtplib.py | 210 |
1 files changed, 195 insertions, 15 deletions
diff --git a/Lib/test/test_smtplib.py b/Lib/test/test_smtplib.py index 3542ddb..cf92662 100644 --- a/Lib/test/test_smtplib.py +++ b/Lib/test/test_smtplib.py @@ -1,53 +1,100 @@ +import asyncore import socket import threading +import smtpd import smtplib +import io +import sys import time +import select from unittest import TestCase from test import test_support +# PORT is used to communicate the port number assigned to the server +# to the test client +HOST = "localhost" +PORT = None -def server(evt): - serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) - serv.settimeout(3) - serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) - serv.bind(("", 9091)) - serv.listen(5) +def server(evt, buf): try: + serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + serv.settimeout(3) + serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + serv.bind(("", 0)) + global PORT + PORT = serv.getsockname()[1] + serv.listen(5) conn, addr = serv.accept() except socket.timeout: pass else: - conn.send("220 Hola mundo\n") + n = 500 + while buf and n > 0: + r, w, e = select.select([], [conn], []) + if w: + sent = conn.send(buf) + buf = buf[sent:] + + n -= 1 + time.sleep(0.01) + conn.close() finally: serv.close() + PORT = None evt.set() class GeneralTests(TestCase): def setUp(self): self.evt = threading.Event() - threading.Thread(target=server, args=(self.evt,)).start() - time.sleep(.1) + servargs = (self.evt, "220 Hola mundo\n") + threading.Thread(target=server, args=servargs).start() + + # wait until server thread has assigned a port number + n = 500 + while PORT is None and n > 0: + time.sleep(0.01) + n -= 1 + + # wait a little longer (sometimes connections are refused + # on slow machines without this additional wait) + time.sleep(0.5) def tearDown(self): self.evt.wait() - def testBasic(self): + def testBasic1(self): # connects - smtp = smtplib.SMTP("localhost", 9091) + smtp = smtplib.SMTP(HOST, PORT) + smtp.sock.close() + + def testBasic2(self): + # connects, include port in host name + smtp = smtplib.SMTP("%s:%s" % (HOST, PORT)) + smtp.sock.close() + + def testLocalHostName(self): + # check that supplied local_hostname is used + smtp = smtplib.SMTP(HOST, PORT, local_hostname="testhost") + self.assertEqual(smtp.local_hostname, "testhost") smtp.sock.close() + def testNonnumericPort(self): + # check that non-numeric port raises ValueError + self.assertRaises(socket.error, smtplib.SMTP, + "localhost", "bogus") + def testTimeoutDefault(self): # default - smtp = smtplib.SMTP("localhost", 9091) + smtp = smtplib.SMTP(HOST, PORT) self.assertTrue(smtp.sock.gettimeout() is None) smtp.sock.close() def testTimeoutValue(self): # a value - smtp = smtplib.SMTP("localhost", 9091, timeout=30) + smtp = smtplib.SMTP(HOST, PORT, timeout=30) self.assertEqual(smtp.sock.gettimeout(), 30) smtp.sock.close() @@ -56,16 +103,149 @@ class GeneralTests(TestCase): previous = socket.getdefaulttimeout() socket.setdefaulttimeout(30) try: - smtp = smtplib.SMTP("localhost", 9091, timeout=None) + smtp = smtplib.SMTP(HOST, PORT, timeout=None) finally: socket.setdefaulttimeout(previous) self.assertEqual(smtp.sock.gettimeout(), 30) smtp.sock.close() +# Test server using smtpd.DebuggingServer +def debugging_server(serv_evt, client_evt): + serv = smtpd.DebuggingServer(("", 0), ('nowhere', -1)) + global PORT + PORT = serv.getsockname()[1] + + try: + if hasattr(select, 'poll'): + poll_fun = asyncore.poll2 + else: + poll_fun = asyncore.poll + + n = 1000 + while asyncore.socket_map and n > 0: + poll_fun(0.01, asyncore.socket_map) + + # when the client conversation is finished, it will + # set client_evt, and it's then ok to kill the server + if client_evt.isSet(): + serv.close() + break + + n -= 1 + + except socket.timeout: + pass + finally: + # allow some time for the client to read the result + time.sleep(0.5) + serv.close() + asyncore.close_all() + PORT = None + time.sleep(0.5) + serv_evt.set() + +MSG_BEGIN = '---------- MESSAGE FOLLOWS ----------\n' +MSG_END = '------------ END MESSAGE ------------\n' + +# Test behavior of smtpd.DebuggingServer +# NOTE: the SMTP objects are created with a non-default local_hostname +# argument to the constructor, since (on some systems) the FQDN lookup +# caused by the default local_hostname sometimes takes so long that the +# test server times out, causing the test to fail. +class DebuggingServerTests(TestCase): + + def setUp(self): + # temporarily replace sys.stdout to capture DebuggingServer output + self.old_stdout = sys.stdout + self.output = io.StringIO() + sys.stdout = self.output + + self.serv_evt = threading.Event() + self.client_evt = threading.Event() + serv_args = (self.serv_evt, self.client_evt) + threading.Thread(target=debugging_server, args=serv_args).start() + + # wait until server thread has assigned a port number + n = 500 + while PORT is None and n > 0: + time.sleep(0.01) + n -= 1 + + # wait a little longer (sometimes connections are refused + # on slow machines without this additional wait) + time.sleep(0.5) + + def tearDown(self): + # indicate that the client is finished + self.client_evt.set() + # wait for the server thread to terminate + self.serv_evt.wait() + # restore sys.stdout + sys.stdout = self.old_stdout + + def testBasic(self): + # connect + smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) + smtp.quit() + + def testEHLO(self): + smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) + expected = (502, b'Error: command "EHLO" not implemented') + self.assertEqual(smtp.ehlo(), expected) + smtp.quit() + + def testHELP(self): + smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) + self.assertEqual(smtp.help(), b'Error: command "HELP" not implemented') + smtp.quit() + + def testSend(self): + # connect and send mail + m = 'A test message' + smtp = smtplib.SMTP(HOST, PORT, local_hostname='localhost', timeout=3) + smtp.sendmail('John', 'Sally', m) + 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) + + +class BadHELOServerTests(TestCase): + + def setUp(self): + self.old_stdout = sys.stdout + self.output = io.StringIO() + sys.stdout = self.output + + self.evt = threading.Event() + servargs = (self.evt, b"199 no hello for you!\n") + threading.Thread(target=server, args=servargs).start() + + # wait until server thread has assigned a port number + n = 500 + while PORT is None and n > 0: + time.sleep(0.01) + n -= 1 + + # wait a little longer (sometimes connections are refused + # on slow machines without this additional wait) + time.sleep(0.5) + + def tearDown(self): + self.evt.wait() + sys.stdout = self.old_stdout + + def testFailingHELO(self): + self.assertRaises(smtplib.SMTPConnectError, smtplib.SMTP, + HOST, PORT, 'localhost', 3) def test_main(verbose=None): - test_support.run_unittest(GeneralTests) + test_support.run_unittest(GeneralTests, DebuggingServerTests, + BadHELOServerTests) if __name__ == '__main__': test_main() |