Home | History | Annotate | Download | only in test
      1 import asyncore
      2 import email.utils
      3 import socket
      4 import smtpd
      5 import smtplib
      6 import StringIO
      7 import sys
      8 import time
      9 import select
     10 
     11 import unittest
     12 from test import test_support
     13 
     14 try:
     15     import threading
     16 except ImportError:
     17     threading = None
     18 
     19 HOST = test_support.HOST
     20 
     21 def server(evt, buf, serv):
     22     serv.listen(5)
     23     evt.set()
     24     try:
     25         conn, addr = serv.accept()
     26     except socket.timeout:
     27         pass
     28     else:
     29         n = 500
     30         while buf and n > 0:
     31             r, w, e = select.select([], [conn], [])
     32             if w:
     33                 sent = conn.send(buf)
     34                 buf = buf[sent:]
     35 
     36             n -= 1
     37 
     38         conn.close()
     39     finally:
     40         serv.close()
     41         evt.set()
     42 
     43 @unittest.skipUnless(threading, 'Threading required for this test.')
     44 class GeneralTests(unittest.TestCase):
     45 
     46     def setUp(self):
     47         self._threads = test_support.threading_setup()
     48         self.evt = threading.Event()
     49         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
     50         self.sock.settimeout(15)
     51         self.port = test_support.bind_port(self.sock)
     52         servargs = (self.evt, "220 Hola mundo\n", self.sock)
     53         self.thread = threading.Thread(target=server, args=servargs)
     54         self.thread.start()
     55         self.evt.wait()
     56         self.evt.clear()
     57 
     58     def tearDown(self):
     59         self.evt.wait()
     60         self.thread.join()
     61         test_support.threading_cleanup(*self._threads)
     62 
     63     def testBasic1(self):
     64         # connects
     65         smtp = smtplib.SMTP(HOST, self.port)
     66         smtp.close()
     67 
     68     def testBasic2(self):
     69         # connects, include port in host name
     70         smtp = smtplib.SMTP("%s:%s" % (HOST, self.port))
     71         smtp.close()
     72 
     73     def testLocalHostName(self):
     74         # check that supplied local_hostname is used
     75         smtp = smtplib.SMTP(HOST, self.port, local_hostname="testhost")
     76         self.assertEqual(smtp.local_hostname, "testhost")
     77         smtp.close()
     78 
     79     def testTimeoutDefault(self):
     80         self.assertIsNone(socket.getdefaulttimeout())
     81         socket.setdefaulttimeout(30)
     82         try:
     83             smtp = smtplib.SMTP(HOST, self.port)
     84         finally:
     85             socket.setdefaulttimeout(None)
     86         self.assertEqual(smtp.sock.gettimeout(), 30)
     87         smtp.close()
     88 
     89     def testTimeoutNone(self):
     90         self.assertIsNone(socket.getdefaulttimeout())
     91         socket.setdefaulttimeout(30)
     92         try:
     93             smtp = smtplib.SMTP(HOST, self.port, timeout=None)
     94         finally:
     95             socket.setdefaulttimeout(None)
     96         self.assertIsNone(smtp.sock.gettimeout())
     97         smtp.close()
     98 
     99     def testTimeoutValue(self):
    100         smtp = smtplib.SMTP(HOST, self.port, timeout=30)
    101         self.assertEqual(smtp.sock.gettimeout(), 30)
    102         smtp.close()
    103 
    104 
    105 # Test server thread using the specified SMTP server class
    106 def debugging_server(serv, serv_evt, client_evt):
    107     serv_evt.set()
    108 
    109     try:
    110         if hasattr(select, 'poll'):
    111             poll_fun = asyncore.poll2
    112         else:
    113             poll_fun = asyncore.poll
    114 
    115         n = 1000
    116         while asyncore.socket_map and n > 0:
    117             poll_fun(0.01, asyncore.socket_map)
    118 
    119             # when the client conversation is finished, it will
    120             # set client_evt, and it's then ok to kill the server
    121             if client_evt.is_set():
    122                 serv.close()
    123                 break
    124 
    125             n -= 1
    126 
    127     except socket.timeout:
    128         pass
    129     finally:
    130         if not client_evt.is_set():
    131             # allow some time for the client to read the result
    132             time.sleep(0.5)
    133             serv.close()
    134         asyncore.close_all()
    135         serv_evt.set()
    136 
    137 MSG_BEGIN = '---------- MESSAGE FOLLOWS ----------\n'
    138 MSG_END = '------------ END MESSAGE ------------\n'
    139 
    140 # NOTE: Some SMTP objects in the tests below are created with a non-default
    141 # local_hostname argument to the constructor, since (on some systems) the FQDN
    142 # lookup caused by the default local_hostname sometimes takes so long that the
    143 # test server times out, causing the test to fail.
    144 
    145 # Test behavior of smtpd.DebuggingServer
    146 @unittest.skipUnless(threading, 'Threading required for this test.')
    147 class DebuggingServerTests(unittest.TestCase):
    148 
    149     def setUp(self):
    150         # temporarily replace sys.stdout to capture DebuggingServer output
    151         self.old_stdout = sys.stdout
    152         self.output = StringIO.StringIO()
    153         sys.stdout = self.output
    154 
    155         self._threads = test_support.threading_setup()
    156         self.serv_evt = threading.Event()
    157         self.client_evt = threading.Event()
    158         # Pick a random unused port by passing 0 for the port number
    159         self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1))
    160         # Keep a note of what port was assigned
    161         self.port = self.serv.socket.getsockname()[1]
    162         serv_args = (self.serv, self.serv_evt, self.client_evt)
    163         self.thread = threading.Thread(target=debugging_server, args=serv_args)
    164         self.thread.start()
    165 
    166         # wait until server thread has assigned a port number
    167         self.serv_evt.wait()
    168         self.serv_evt.clear()
    169 
    170     def tearDown(self):
    171         # indicate that the client is finished
    172         self.client_evt.set()
    173         # wait for the server thread to terminate
    174         self.serv_evt.wait()
    175         self.thread.join()
    176         test_support.threading_cleanup(*self._threads)
    177         # restore sys.stdout
    178         sys.stdout = self.old_stdout
    179 
    180     def testBasic(self):
    181         # connect
    182         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    183         smtp.quit()
    184 
    185     def testNOOP(self):
    186         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    187         expected = (250, 'Ok')
    188         self.assertEqual(smtp.noop(), expected)
    189         smtp.quit()
    190 
    191     def testRSET(self):
    192         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    193         expected = (250, 'Ok')
    194         self.assertEqual(smtp.rset(), expected)
    195         smtp.quit()
    196 
    197     def testNotImplemented(self):
    198         # EHLO isn't implemented in DebuggingServer
    199         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    200         expected = (502, 'Error: command "EHLO" not implemented')
    201         self.assertEqual(smtp.ehlo(), expected)
    202         smtp.quit()
    203 
    204     def testVRFY(self):
    205         # VRFY isn't implemented in DebuggingServer
    206         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    207         expected = (502, 'Error: command "VRFY" not implemented')
    208         self.assertEqual(smtp.vrfy('nobody (at] nowhere.com'), expected)
    209         self.assertEqual(smtp.verify('nobody (at] nowhere.com'), expected)
    210         smtp.quit()
    211 
    212     def testSecondHELO(self):
    213         # check that a second HELO returns a message that it's a duplicate
    214         # (this behavior is specific to smtpd.SMTPChannel)
    215         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    216         smtp.helo()
    217         expected = (503, 'Duplicate HELO/EHLO')
    218         self.assertEqual(smtp.helo(), expected)
    219         smtp.quit()
    220 
    221     def testHELP(self):
    222         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    223         self.assertEqual(smtp.help(), 'Error: command "HELP" not implemented')
    224         smtp.quit()
    225 
    226     def testSend(self):
    227         # connect and send mail
    228         m = 'A test message'
    229         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    230         smtp.sendmail('John', 'Sally', m)
    231         # XXX(nnorwitz): this test is flaky and dies with a bad file descriptor
    232         # in asyncore.  This sleep might help, but should really be fixed
    233         # properly by using an Event variable.
    234         time.sleep(0.01)
    235         smtp.quit()
    236 
    237         self.client_evt.set()
    238         self.serv_evt.wait()
    239         self.output.flush()
    240         mexpect = '%s%s\n%s' % (MSG_BEGIN, m, MSG_END)
    241         self.assertEqual(self.output.getvalue(), mexpect)
    242 
    243 
    244 class NonConnectingTests(unittest.TestCase):
    245 
    246     def testNotConnected(self):
    247         # Test various operations on an unconnected SMTP object that
    248         # should raise exceptions (at present the attempt in SMTP.send
    249         # to reference the nonexistent 'sock' attribute of the SMTP object
    250         # causes an AttributeError)
    251         smtp = smtplib.SMTP()
    252         self.assertRaises(smtplib.SMTPServerDisconnected, smtp.ehlo)
    253         self.assertRaises(smtplib.SMTPServerDisconnected,
    254                           smtp.send, 'test msg')
    255 
    256     def testNonnumericPort(self):
    257         # check that non-numeric port raises socket.error
    258         self.assertRaises(socket.error, smtplib.SMTP,
    259                           "localhost", "bogus")
    260         self.assertRaises(socket.error, smtplib.SMTP,
    261                           "localhost:bogus")
    262 
    263 
    264 # test response of client to a non-successful HELO message
    265 @unittest.skipUnless(threading, 'Threading required for this test.')
    266 class BadHELOServerTests(unittest.TestCase):
    267 
    268     def setUp(self):
    269         self.old_stdout = sys.stdout
    270         self.output = StringIO.StringIO()
    271         sys.stdout = self.output
    272 
    273         self._threads = test_support.threading_setup()
    274         self.evt = threading.Event()
    275         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    276         self.sock.settimeout(15)
    277         self.port = test_support.bind_port(self.sock)
    278         servargs = (self.evt, "199 no hello for you!\n", self.sock)
    279         self.thread = threading.Thread(target=server, args=servargs)
    280         self.thread.start()
    281         self.evt.wait()
    282         self.evt.clear()
    283 
    284     def tearDown(self):
    285         self.evt.wait()
    286         self.thread.join()
    287         test_support.threading_cleanup(*self._threads)
    288         sys.stdout = self.old_stdout
    289 
    290     def testFailingHELO(self):
    291         self.assertRaises(smtplib.SMTPConnectError, smtplib.SMTP,
    292                             HOST, self.port, 'localhost', 3)
    293 
    294 
    295 @unittest.skipUnless(threading, 'Threading required for this test.')
    296 class TooLongLineTests(unittest.TestCase):
    297     respdata = '250 OK' + ('.' * smtplib._MAXLINE * 2) + '\n'
    298 
    299     def setUp(self):
    300         self.old_stdout = sys.stdout
    301         self.output = StringIO.StringIO()
    302         sys.stdout = self.output
    303 
    304         self.evt = threading.Event()
    305         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    306         self.sock.settimeout(15)
    307         self.port = test_support.bind_port(self.sock)
    308         servargs = (self.evt, self.respdata, self.sock)
    309         self.thread = threading.Thread(target=server, args=servargs)
    310         self.thread.start()
    311         self.evt.wait()
    312         self.evt.clear()
    313 
    314     def tearDown(self):
    315         self.evt.wait()
    316         self.thread.join()
    317         sys.stdout = self.old_stdout
    318 
    319     def testLineTooLong(self):
    320         self.assertRaises(smtplib.SMTPResponseException, smtplib.SMTP,
    321                           HOST, self.port, 'localhost', 3)
    322 
    323 
    324 sim_users = {'Mr.A (at] somewhere.com':'John A',
    325              'Ms.B (at] somewhere.com':'Sally B',
    326              'Mrs.C (at] somewhereesle.com':'Ruth C',
    327             }
    328 
    329 sim_auth = ('Mr.A (at] somewhere.com', 'somepassword')
    330 sim_cram_md5_challenge = ('PENCeUxFREJoU0NnbmhNWitOMjNGNn'
    331                           'dAZWx3b29kLmlubm9zb2Z0LmNvbT4=')
    332 sim_auth_credentials = {
    333     'login': 'TXIuQUBzb21ld2hlcmUuY29t',
    334     'plain': 'AE1yLkFAc29tZXdoZXJlLmNvbQBzb21lcGFzc3dvcmQ=',
    335     'cram-md5': ('TXIUQUBZB21LD2HLCMUUY29TIDG4OWQ0MJ'
    336                  'KWZGQ4ODNMNDA4NTGXMDRLZWMYZJDMODG1'),
    337     }
    338 sim_auth_login_password = 'C29TZXBHC3N3B3JK'
    339 
    340 sim_lists = {'list-1':['Mr.A (at] somewhere.com','Mrs.C (at] somewhereesle.com'],
    341              'list-2':['Ms.B (at] somewhere.com',],
    342             }
    343 
    344 # Simulated SMTP channel & server
    345 class SimSMTPChannel(smtpd.SMTPChannel):
    346 
    347     def __init__(self, extra_features, *args, **kw):
    348         self._extrafeatures = ''.join(
    349             [ "250-{0}\r\n".format(x) for x in extra_features ])
    350         smtpd.SMTPChannel.__init__(self, *args, **kw)
    351 
    352     def smtp_EHLO(self, arg):
    353         resp = ('250-testhost\r\n'
    354                 '250-EXPN\r\n'
    355                 '250-SIZE 20000000\r\n'
    356                 '250-STARTTLS\r\n'
    357                 '250-DELIVERBY\r\n')
    358         resp = resp + self._extrafeatures + '250 HELP'
    359         self.push(resp)
    360 
    361     def smtp_VRFY(self, arg):
    362         # For max compatibility smtplib should be sending the raw address.
    363         if arg in sim_users:
    364             self.push('250 %s %s' % (sim_users[arg], smtplib.quoteaddr(arg)))
    365         else:
    366             self.push('550 No such user: %s' % arg)
    367 
    368     def smtp_EXPN(self, arg):
    369         list_name = arg.lower()
    370         if list_name in sim_lists:
    371             user_list = sim_lists[list_name]
    372             for n, user_email in enumerate(user_list):
    373                 quoted_addr = smtplib.quoteaddr(user_email)
    374                 if n < len(user_list) - 1:
    375                     self.push('250-%s %s' % (sim_users[user_email], quoted_addr))
    376                 else:
    377                     self.push('250 %s %s' % (sim_users[user_email], quoted_addr))
    378         else:
    379             self.push('550 No access for you!')
    380 
    381     def smtp_AUTH(self, arg):
    382         if arg.strip().lower()=='cram-md5':
    383             self.push('334 {0}'.format(sim_cram_md5_challenge))
    384             return
    385         mech, auth = arg.split()
    386         mech = mech.lower()
    387         if mech not in sim_auth_credentials:
    388             self.push('504 auth type unimplemented')
    389             return
    390         if mech == 'plain' and auth==sim_auth_credentials['plain']:
    391             self.push('235 plain auth ok')
    392         elif mech=='login' and auth==sim_auth_credentials['login']:
    393             self.push('334 Password:')
    394         else:
    395             self.push('550 No access for you!')
    396 
    397     def handle_error(self):
    398         raise
    399 
    400 
    401 class SimSMTPServer(smtpd.SMTPServer):
    402 
    403     def __init__(self, *args, **kw):
    404         self._extra_features = []
    405         smtpd.SMTPServer.__init__(self, *args, **kw)
    406 
    407     def handle_accept(self):
    408         conn, addr = self.accept()
    409         self._SMTPchannel = SimSMTPChannel(self._extra_features,
    410                                            self, conn, addr)
    411 
    412     def process_message(self, peer, mailfrom, rcpttos, data):
    413         pass
    414 
    415     def add_feature(self, feature):
    416         self._extra_features.append(feature)
    417 
    418     def handle_error(self):
    419         raise
    420 
    421 
    422 # Test various SMTP & ESMTP commands/behaviors that require a simulated server
    423 # (i.e., something with more features than DebuggingServer)
    424 @unittest.skipUnless(threading, 'Threading required for this test.')
    425 class SMTPSimTests(unittest.TestCase):
    426 
    427     def setUp(self):
    428         self._threads = test_support.threading_setup()
    429         self.serv_evt = threading.Event()
    430         self.client_evt = threading.Event()
    431         # Pick a random unused port by passing 0 for the port number
    432         self.serv = SimSMTPServer((HOST, 0), ('nowhere', -1))
    433         # Keep a note of what port was assigned
    434         self.port = self.serv.socket.getsockname()[1]
    435         serv_args = (self.serv, self.serv_evt, self.client_evt)
    436         self.thread = threading.Thread(target=debugging_server, args=serv_args)
    437         self.thread.start()
    438 
    439         # wait until server thread has assigned a port number
    440         self.serv_evt.wait()
    441         self.serv_evt.clear()
    442 
    443     def tearDown(self):
    444         # indicate that the client is finished
    445         self.client_evt.set()
    446         # wait for the server thread to terminate
    447         self.serv_evt.wait()
    448         self.thread.join()
    449         test_support.threading_cleanup(*self._threads)
    450 
    451     def testBasic(self):
    452         # smoke test
    453         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    454         smtp.quit()
    455 
    456     def testEHLO(self):
    457         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    458 
    459         # no features should be present before the EHLO
    460         self.assertEqual(smtp.esmtp_features, {})
    461 
    462         # features expected from the test server
    463         expected_features = {'expn':'',
    464                              'size': '20000000',
    465                              'starttls': '',
    466                              'deliverby': '',
    467                              'help': '',
    468                              }
    469 
    470         smtp.ehlo()
    471         self.assertEqual(smtp.esmtp_features, expected_features)
    472         for k in expected_features:
    473             self.assertTrue(smtp.has_extn(k))
    474         self.assertFalse(smtp.has_extn('unsupported-feature'))
    475         smtp.quit()
    476 
    477     def testVRFY(self):
    478         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    479 
    480         for email, name in sim_users.items():
    481             expected_known = (250, '%s %s' % (name, smtplib.quoteaddr(email)))
    482             self.assertEqual(smtp.vrfy(email), expected_known)
    483 
    484         u = 'nobody (at] nowhere.com'
    485         expected_unknown = (550, 'No such user: %s' % u)
    486         self.assertEqual(smtp.vrfy(u), expected_unknown)
    487         smtp.quit()
    488 
    489     def testEXPN(self):
    490         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    491 
    492         for listname, members in sim_lists.items():
    493             users = []
    494             for m in members:
    495                 users.append('%s %s' % (sim_users[m], smtplib.quoteaddr(m)))
    496             expected_known = (250, '\n'.join(users))
    497             self.assertEqual(smtp.expn(listname), expected_known)
    498 
    499         u = 'PSU-Members-List'
    500         expected_unknown = (550, 'No access for you!')
    501         self.assertEqual(smtp.expn(u), expected_unknown)
    502         smtp.quit()
    503 
    504     def testAUTH_PLAIN(self):
    505         self.serv.add_feature("AUTH PLAIN")
    506         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    507 
    508         expected_auth_ok = (235, b'plain auth ok')
    509         self.assertEqual(smtp.login(sim_auth[0], sim_auth[1]), expected_auth_ok)
    510 
    511     # SimSMTPChannel doesn't fully support LOGIN or CRAM-MD5 auth because they
    512     # require a synchronous read to obtain the credentials...so instead smtpd
    513     # sees the credential sent by smtplib's login method as an unknown command,
    514     # which results in smtplib raising an auth error.  Fortunately the error
    515     # message contains the encoded credential, so we can partially check that it
    516     # was generated correctly (partially, because the 'word' is uppercased in
    517     # the error message).
    518 
    519     def testAUTH_LOGIN(self):
    520         self.serv.add_feature("AUTH LOGIN")
    521         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    522         try: smtp.login(sim_auth[0], sim_auth[1])
    523         except smtplib.SMTPAuthenticationError as err:
    524             if sim_auth_login_password not in str(err):
    525                 raise "expected encoded password not found in error message"
    526 
    527     def testAUTH_CRAM_MD5(self):
    528         self.serv.add_feature("AUTH CRAM-MD5")
    529         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
    530 
    531         try: smtp.login(sim_auth[0], sim_auth[1])
    532         except smtplib.SMTPAuthenticationError as err:
    533             if sim_auth_credentials['cram-md5'] not in str(err):
    534                 raise "expected encoded credentials not found in error message"
    535 
    536     #TODO: add tests for correct AUTH method fallback now that the
    537     #test infrastructure can support it.
    538 
    539     def test_quit_resets_greeting(self):
    540         smtp = smtplib.SMTP(HOST, self.port,
    541                             local_hostname='localhost',
    542                             timeout=15)
    543         code, message = smtp.ehlo()
    544         self.assertEqual(code, 250)
    545         self.assertIn('size', smtp.esmtp_features)
    546         smtp.quit()
    547         self.assertNotIn('size', smtp.esmtp_features)
    548         smtp.connect(HOST, self.port)
    549         self.assertNotIn('size', smtp.esmtp_features)
    550         smtp.ehlo_or_helo_if_needed()
    551         self.assertIn('size', smtp.esmtp_features)
    552         smtp.quit()
    553 
    554 
    555 def test_main(verbose=None):
    556     test_support.run_unittest(GeneralTests, DebuggingServerTests,
    557                               NonConnectingTests,
    558                               BadHELOServerTests, SMTPSimTests,
    559                               TooLongLineTests)
    560 
    561 if __name__ == '__main__':
    562     test_main()
    563