AppPkg/Applications/Python/Python-2.7.2/Lib/test/test_smtplib.py

   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.assertTrue(socket.getdefaulttimeout() is None)
  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.assertTrue(socket.getdefaulttimeout() is None)
  91         socket.setdefaulttimeout(30)
  92         try:
  93             smtp = smtplib.SMTP(HOST, self.port, timeout=None)
  94         finally:
  95             socket.setdefaulttimeout(None)
  96         self.assertTrue(smtp.sock.gettimeout() is None)
  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=3)
 183         smtp.quit()
 184
 185     def testNOOP(self):
 186         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
 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=3)
 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=3)
 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=3)
 207         expected = (502, 'Error: command "VRFY" not implemented')
 208         self.assertEqual(smtp.vrfy('nobody@nowhere.com'), expected)
 209         self.assertEqual(smtp.verify('nobody@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=3)
 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=3)
 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=3)
 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 sim_users = {'Mr.A@somewhere.com':'John A',
 296              'Ms.B@somewhere.com':'Sally B',
 297              'Mrs.C@somewhereesle.com':'Ruth C',
 298             }
 299
 300 sim_auth = ('Mr.A@somewhere.com', 'somepassword')
 301 sim_cram_md5_challenge = ('PENCeUxFREJoU0NnbmhNWitOMjNGNn'
 302                           'dAZWx3b29kLmlubm9zb2Z0LmNvbT4=')
 303 sim_auth_credentials = {
 304     'login': 'TXIuQUBzb21ld2hlcmUuY29t',
 305     'plain': 'AE1yLkFAc29tZXdoZXJlLmNvbQBzb21lcGFzc3dvcmQ=',
 306     'cram-md5': ('TXIUQUBZB21LD2HLCMUUY29TIDG4OWQ0MJ'
 307                  'KWZGQ4ODNMNDA4NTGXMDRLZWMYZJDMODG1'),
 308     }
 309 sim_auth_login_password = 'C29TZXBHC3N3B3JK'
 310
 311 sim_lists = {'list-1':['Mr.A@somewhere.com','Mrs.C@somewhereesle.com'],
 312              'list-2':['Ms.B@somewhere.com',],
 313             }
 314
 315 # Simulated SMTP channel & server
 316 class SimSMTPChannel(smtpd.SMTPChannel):
 317
 318     def __init__(self, extra_features, *args, **kw):
 319         self._extrafeatures = ''.join(
 320             [ "250-{0}\r\n".format(x) for x in extra_features ])
 321         smtpd.SMTPChannel.__init__(self, *args, **kw)
 322
 323     def smtp_EHLO(self, arg):
 324         resp = ('250-testhost\r\n'
 325                 '250-EXPN\r\n'
 326                 '250-SIZE 20000000\r\n'
 327                 '250-STARTTLS\r\n'
 328                 '250-DELIVERBY\r\n')
 329         resp = resp + self._extrafeatures + '250 HELP'
 330         self.push(resp)
 331
 332     def smtp_VRFY(self, arg):
 333         raw_addr = email.utils.parseaddr(arg)[1]
 334         quoted_addr = smtplib.quoteaddr(arg)
 335         if raw_addr in sim_users:
 336             self.push('250 %s %s' % (sim_users[raw_addr], quoted_addr))
 337         else:
 338             self.push('550 No such user: %s' % arg)
 339
 340     def smtp_EXPN(self, arg):
 341         list_name = email.utils.parseaddr(arg)[1].lower()
 342         if list_name in sim_lists:
 343             user_list = sim_lists[list_name]
 344             for n, user_email in enumerate(user_list):
 345                 quoted_addr = smtplib.quoteaddr(user_email)
 346                 if n < len(user_list) - 1:
 347                     self.push('250-%s %s' % (sim_users[user_email], quoted_addr))
 348                 else:
 349                     self.push('250 %s %s' % (sim_users[user_email], quoted_addr))
 350         else:
 351             self.push('550 No access for you!')
 352
 353     def smtp_AUTH(self, arg):
 354         if arg.strip().lower()=='cram-md5':
 355             self.push('334 {0}'.format(sim_cram_md5_challenge))
 356             return
 357         mech, auth = arg.split()
 358         mech = mech.lower()
 359         if mech not in sim_auth_credentials:
 360             self.push('504 auth type unimplemented')
 361             return
 362         if mech == 'plain' and auth==sim_auth_credentials['plain']:
 363             self.push('235 plain auth ok')
 364         elif mech=='login' and auth==sim_auth_credentials['login']:
 365             self.push('334 Password:')
 366         else:
 367             self.push('550 No access for you!')
 368
 369     def handle_error(self):
 370         raise
 371
 372
 373 class SimSMTPServer(smtpd.SMTPServer):
 374
 375     def __init__(self, *args, **kw):
 376         self._extra_features = []
 377         smtpd.SMTPServer.__init__(self, *args, **kw)
 378
 379     def handle_accept(self):
 380         conn, addr = self.accept()
 381         self._SMTPchannel = SimSMTPChannel(self._extra_features,
 382                                            self, conn, addr)
 383
 384     def process_message(self, peer, mailfrom, rcpttos, data):
 385         pass
 386
 387     def add_feature(self, feature):
 388         self._extra_features.append(feature)
 389
 390     def handle_error(self):
 391         raise
 392
 393
 394 # Test various SMTP & ESMTP commands/behaviors that require a simulated server
 395 # (i.e., something with more features than DebuggingServer)
 396 @unittest.skipUnless(threading, 'Threading required for this test.')
 397 class SMTPSimTests(unittest.TestCase):
 398
 399     def setUp(self):
 400         self._threads = test_support.threading_setup()
 401         self.serv_evt = threading.Event()
 402         self.client_evt = threading.Event()
 403         # Pick a random unused port by passing 0 for the port number
 404         self.serv = SimSMTPServer((HOST, 0), ('nowhere', -1))
 405         # Keep a note of what port was assigned
 406         self.port = self.serv.socket.getsockname()[1]
 407         serv_args = (self.serv, self.serv_evt, self.client_evt)
 408         self.thread = threading.Thread(target=debugging_server, args=serv_args)
 409         self.thread.start()
 410
 411         # wait until server thread has assigned a port number
 412         self.serv_evt.wait()
 413         self.serv_evt.clear()
 414
 415     def tearDown(self):
 416         # indicate that the client is finished
 417         self.client_evt.set()
 418         # wait for the server thread to terminate
 419         self.serv_evt.wait()
 420         self.thread.join()
 421         test_support.threading_cleanup(*self._threads)
 422
 423     def testBasic(self):
 424         # smoke test
 425         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
 426         smtp.quit()
 427
 428     def testEHLO(self):
 429         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
 430
 431         # no features should be present before the EHLO
 432         self.assertEqual(smtp.esmtp_features, {})
 433
 434         # features expected from the test server
 435         expected_features = {'expn':'',
 436                              'size': '20000000',
 437                              'starttls': '',
 438                              'deliverby': '',
 439                              'help': '',
 440                              }
 441
 442         smtp.ehlo()
 443         self.assertEqual(smtp.esmtp_features, expected_features)
 444         for k in expected_features:
 445             self.assertTrue(smtp.has_extn(k))
 446         self.assertFalse(smtp.has_extn('unsupported-feature'))
 447         smtp.quit()
 448
 449     def testVRFY(self):
 450         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
 451
 452         for email, name in sim_users.items():
 453             expected_known = (250, '%s %s' % (name, smtplib.quoteaddr(email)))
 454             self.assertEqual(smtp.vrfy(email), expected_known)
 455
 456         u = 'nobody@nowhere.com'
 457         expected_unknown = (550, 'No such user: %s' % smtplib.quoteaddr(u))
 458         self.assertEqual(smtp.vrfy(u), expected_unknown)
 459         smtp.quit()
 460
 461     def testEXPN(self):
 462         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
 463
 464         for listname, members in sim_lists.items():
 465             users = []
 466             for m in members:
 467                 users.append('%s %s' % (sim_users[m], smtplib.quoteaddr(m)))
 468             expected_known = (250, '\n'.join(users))
 469             self.assertEqual(smtp.expn(listname), expected_known)
 470
 471         u = 'PSU-Members-List'
 472         expected_unknown = (550, 'No access for you!')
 473         self.assertEqual(smtp.expn(u), expected_unknown)
 474         smtp.quit()
 475
 476     def testAUTH_PLAIN(self):
 477         self.serv.add_feature("AUTH PLAIN")
 478         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
 479
 480         expected_auth_ok = (235, b'plain auth ok')
 481         self.assertEqual(smtp.login(sim_auth[0], sim_auth[1]), expected_auth_ok)
 482
 483     # SimSMTPChannel doesn't fully support LOGIN or CRAM-MD5 auth because they
 484     # require a synchronous read to obtain the credentials...so instead smtpd
 485     # sees the credential sent by smtplib's login method as an unknown command,
 486     # which results in smtplib raising an auth error.  Fortunately the error
 487     # message contains the encoded credential, so we can partially check that it
 488     # was generated correctly (partially, because the 'word' is uppercased in
 489     # the error message).
 490
 491     def testAUTH_LOGIN(self):
 492         self.serv.add_feature("AUTH LOGIN")
 493         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
 494         try: smtp.login(sim_auth[0], sim_auth[1])
 495         except smtplib.SMTPAuthenticationError as err:
 496             if sim_auth_login_password not in str(err):
 497                 raise "expected encoded password not found in error message"
 498
 499     def testAUTH_CRAM_MD5(self):
 500         self.serv.add_feature("AUTH CRAM-MD5")
 501         smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
 502
 503         try: smtp.login(sim_auth[0], sim_auth[1])
 504         except smtplib.SMTPAuthenticationError as err:
 505             if sim_auth_credentials['cram-md5'] not in str(err):
 506                 raise "expected encoded credentials not found in error message"
 507
 508     #TODO: add tests for correct AUTH method fallback now that the
 509     #test infrastructure can support it.
 510
 511
 512 def test_main(verbose=None):
 513     test_support.run_unittest(GeneralTests, DebuggingServerTests,
 514                               NonConnectingTests,
 515                               BadHELOServerTests, SMTPSimTests)
 516
 517 if __name__ == '__main__':
 518     test_main()