<s> import asyncore <EOL> import email . mime . text <EOL> import email . utils <EOL> import socket <EOL> import smtpd <EOL> import smtplib <EOL> import io <EOL> import re <EOL> import sys <EOL> import time <EOL> import select <EOL> import errno <EOL> import unittest <EOL> from test import support , mock_socket <EOL> try : <EOL> import threading <EOL> except ImportError : <EOL> threading = None <EOL> HOST = support . HOST <EOL> if sys . platform == '<STR_LIT>' : <EOL> def handle_expt ( self ) : <EOL> pass <EOL> smtpd . SMTPChannel . handle_expt = handle_expt <EOL> def server ( evt , buf , serv ) : <EOL> serv . listen ( <NUM_LIT:5> ) <EOL> evt . set ( ) <EOL> try : <EOL> conn , addr = serv . accept ( ) <EOL> except socket . timeout : <EOL> pass <EOL> else : <EOL> n = <NUM_LIT> <EOL> while buf and n > <NUM_LIT:0> : <EOL> r , w , e = select . select ( [ ] , [ conn ] , [ ] ) <EOL> if w : <EOL> sent = conn . send ( buf ) <EOL> buf = buf [ sent : ] <EOL> n -= <NUM_LIT:1> <EOL> conn . close ( ) <EOL> finally : <EOL> serv . close ( ) <EOL> evt . set ( ) <EOL> class GeneralTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> smtplib . socket = mock_socket <EOL> self . port = <NUM_LIT> <EOL> def tearDown ( self ) : <EOL> smtplib . socket = socket <EOL> def testQuoteData ( self ) : <EOL> teststr = "<STR_LIT>" <EOL> expected = "<STR_LIT>" <EOL> self . assertEqual ( expected , smtplib . quotedata ( teststr ) ) <EOL> def testBasic1 ( self ) : <EOL> mock_socket . reply_with ( b"<STR_LIT>" ) <EOL> smtp = smtplib . SMTP ( HOST , self . port ) <EOL> smtp . close ( ) <EOL> def testSourceAddress ( self ) : <EOL> mock_socket . reply_with ( b"<STR_LIT>" ) <EOL> smtp = smtplib . SMTP ( HOST , self . port , <EOL> source_address = ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) ) <EOL> self . assertEqual ( smtp . source_address , ( '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) ) <EOL> smtp . close ( ) <EOL> def testBasic2 ( self ) : <EOL> mock_socket . reply_with ( b"<STR_LIT>" ) <EOL> smtp = smtplib . SMTP ( "<STR_LIT>" % ( HOST , self . port ) ) <EOL> smtp . close ( ) <EOL> def testLocalHostName ( self ) : <EOL> mock_socket . reply_with ( b"<STR_LIT>" ) <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = "<STR_LIT>" ) <EOL> self . assertEqual ( smtp . local_hostname , "<STR_LIT>" ) <EOL> smtp . close ( ) <EOL> def testTimeoutDefault ( self ) : <EOL> mock_socket . reply_with ( b"<STR_LIT>" ) <EOL> self . assertTrue ( mock_socket . getdefaulttimeout ( ) is None ) <EOL> mock_socket . setdefaulttimeout ( <NUM_LIT:30> ) <EOL> self . assertEqual ( mock_socket . getdefaulttimeout ( ) , <NUM_LIT:30> ) <EOL> try : <EOL> smtp = smtplib . SMTP ( HOST , self . port ) <EOL> finally : <EOL> mock_socket . setdefaulttimeout ( None ) <EOL> self . assertEqual ( smtp . sock . gettimeout ( ) , <NUM_LIT:30> ) <EOL> smtp . close ( ) <EOL> def testTimeoutNone ( self ) : <EOL> mock_socket . reply_with ( b"<STR_LIT>" ) <EOL> self . assertTrue ( socket . getdefaulttimeout ( ) is None ) <EOL> socket . setdefaulttimeout ( <NUM_LIT:30> ) <EOL> try : <EOL> smtp = smtplib . SMTP ( HOST , self . port , timeout = None ) <EOL> finally : <EOL> socket . setdefaulttimeout ( None ) <EOL> self . assertTrue ( smtp . sock . gettimeout ( ) is None ) <EOL> smtp . close ( ) <EOL> def testTimeoutValue ( self ) : <EOL> mock_socket . reply_with ( b"<STR_LIT>" ) <EOL> smtp = smtplib . SMTP ( HOST , self . port , timeout = <NUM_LIT:30> ) <EOL> self . assertEqual ( smtp . sock . gettimeout ( ) , <NUM_LIT:30> ) <EOL> smtp . close ( ) <EOL> def debugging_server ( serv , serv_evt , client_evt ) : <EOL> serv_evt . set ( ) <EOL> try : <EOL> if hasattr ( select , '<STR_LIT>' ) : <EOL> poll_fun = asyncore . poll2 <EOL> else : <EOL> poll_fun = asyncore . poll <EOL> n = <NUM_LIT:1000> <EOL> while asyncore . socket_map and n > <NUM_LIT:0> : <EOL> poll_fun ( <NUM_LIT> , asyncore . socket_map ) <EOL> if client_evt . is_set ( ) : <EOL> serv . close ( ) <EOL> break <EOL> n -= <NUM_LIT:1> <EOL> except socket . timeout : <EOL> pass <EOL> finally : <EOL> if not client_evt . is_set ( ) : <EOL> time . sleep ( <NUM_LIT:0.5> ) <EOL> serv . close ( ) <EOL> asyncore . close_all ( ) <EOL> serv_evt . set ( ) <EOL> MSG_BEGIN = '<STR_LIT>' <EOL> MSG_END = '<STR_LIT>' <EOL> @ unittest . skipUnless ( threading , '<STR_LIT>' ) <EOL> class DebuggingServerTests ( unittest . TestCase ) : <EOL> maxDiff = None <EOL> def setUp ( self ) : <EOL> self . real_getfqdn = socket . getfqdn <EOL> socket . getfqdn = mock_socket . getfqdn <EOL> self . old_stdout = sys . stdout <EOL> self . output = io . StringIO ( ) <EOL> sys . stdout = self . output <EOL> self . serv_evt = threading . Event ( ) <EOL> self . client_evt = threading . Event ( ) <EOL> self . old_DEBUGSTREAM = smtpd . DEBUGSTREAM <EOL> smtpd . DEBUGSTREAM = io . StringIO ( ) <EOL> self . serv = smtpd . DebuggingServer ( ( HOST , <NUM_LIT:0> ) , ( '<STR_LIT>' , - <NUM_LIT:1> ) ) <EOL> self . port = self . serv . socket . getsockname ( ) [ <NUM_LIT:1> ] <EOL> serv_args = ( self . serv , self . serv_evt , self . client_evt ) <EOL> self . thread = threading . Thread ( target = debugging_server , args = serv_args ) <EOL> self . thread . start ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . serv_evt . clear ( ) <EOL> def tearDown ( self ) : <EOL> socket . getfqdn = self . real_getfqdn <EOL> self . client_evt . set ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . thread . join ( ) <EOL> sys . stdout = self . old_stdout <EOL> smtpd . DEBUGSTREAM . close ( ) <EOL> smtpd . DEBUGSTREAM = self . old_DEBUGSTREAM <EOL> def testBasic ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> smtp . quit ( ) <EOL> def testSourceAddress ( self ) : <EOL> port = support . find_unused_port ( ) <EOL> try : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , <EOL> timeout = <NUM_LIT:3> , source_address = ( '<STR_LIT:127.0.0.1>' , port ) ) <EOL> self . assertEqual ( smtp . source_address , ( '<STR_LIT:127.0.0.1>' , port ) ) <EOL> self . assertEqual ( smtp . local_hostname , '<STR_LIT:localhost>' ) <EOL> smtp . quit ( ) <EOL> except IOError as e : <EOL> if e . errno == errno . EADDRINUSE : <EOL> self . skipTest ( "<STR_LIT>" % port ) <EOL> raise <EOL> def testNOOP ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> expected = ( <NUM_LIT> , b'<STR_LIT:OK>' ) <EOL> self . assertEqual ( smtp . noop ( ) , expected ) <EOL> smtp . quit ( ) <EOL> def testRSET ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> expected = ( <NUM_LIT> , b'<STR_LIT:OK>' ) <EOL> self . assertEqual ( smtp . rset ( ) , expected ) <EOL> smtp . quit ( ) <EOL> def testELHO ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> expected = ( <NUM_LIT> , b'<STR_LIT>' ) <EOL> self . assertEqual ( smtp . ehlo ( ) , expected ) <EOL> smtp . quit ( ) <EOL> def testEXPNNotImplemented ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> expected = ( <NUM_LIT> , b'<STR_LIT>' ) <EOL> smtp . putcmd ( '<STR_LIT>' ) <EOL> self . assertEqual ( smtp . getreply ( ) , expected ) <EOL> smtp . quit ( ) <EOL> def testVRFY ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> expected = ( <NUM_LIT> , b'<STR_LIT>' + b'<STR_LIT>' ) <EOL> self . assertEqual ( smtp . vrfy ( '<STR_LIT>' ) , expected ) <EOL> self . assertEqual ( smtp . verify ( '<STR_LIT>' ) , expected ) <EOL> smtp . quit ( ) <EOL> def testSecondHELO ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> smtp . helo ( ) <EOL> expected = ( <NUM_LIT> , b'<STR_LIT>' ) <EOL> self . assertEqual ( smtp . helo ( ) , expected ) <EOL> smtp . quit ( ) <EOL> def testHELP ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> self . assertEqual ( smtp . help ( ) , b'<STR_LIT>' + b'<STR_LIT>' ) <EOL> smtp . quit ( ) <EOL> def testSend ( self ) : <EOL> m = '<STR_LIT>' <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> smtp . sendmail ( '<STR_LIT>' , '<STR_LIT>' , m ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> smtp . quit ( ) <EOL> self . client_evt . set ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . output . flush ( ) <EOL> mexpect = '<STR_LIT>' % ( MSG_BEGIN , m , MSG_END ) <EOL> self . assertEqual ( self . output . getvalue ( ) , mexpect ) <EOL> def testSendBinary ( self ) : <EOL> m = b'<STR_LIT>' <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> smtp . sendmail ( '<STR_LIT>' , '<STR_LIT>' , m ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> smtp . quit ( ) <EOL> self . client_evt . set ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . output . flush ( ) <EOL> mexpect = '<STR_LIT>' % ( MSG_BEGIN , m . decode ( '<STR_LIT:ascii>' ) , MSG_END ) <EOL> self . assertEqual ( self . output . getvalue ( ) , mexpect ) <EOL> def testSendNeedingDotQuote ( self ) : <EOL> m = '<STR_LIT>' <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> smtp . sendmail ( '<STR_LIT>' , '<STR_LIT>' , m ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> smtp . quit ( ) <EOL> self . client_evt . set ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . output . flush ( ) <EOL> mexpect = '<STR_LIT>' % ( MSG_BEGIN , m , MSG_END ) <EOL> self . assertEqual ( self . output . getvalue ( ) , mexpect ) <EOL> def testSendNullSender ( self ) : <EOL> m = '<STR_LIT>' <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> smtp . sendmail ( '<STR_LIT>' , '<STR_LIT>' , m ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> smtp . quit ( ) <EOL> self . client_evt . set ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . output . flush ( ) <EOL> mexpect = '<STR_LIT>' % ( MSG_BEGIN , m , MSG_END ) <EOL> self . assertEqual ( self . output . getvalue ( ) , mexpect ) <EOL> debugout = smtpd . DEBUGSTREAM . getvalue ( ) <EOL> sender = re . compile ( "<STR_LIT>" , re . MULTILINE ) <EOL> self . assertRegex ( debugout , sender ) <EOL> def testSendMessage ( self ) : <EOL> m = email . mime . text . MIMEText ( '<STR_LIT>' ) <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> smtp . send_message ( m , from_addr = '<STR_LIT>' , to_addrs = '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> smtp . quit ( ) <EOL> self . client_evt . set ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . output . flush ( ) <EOL> m [ '<STR_LIT>' ] = socket . gethostbyname ( '<STR_LIT:localhost>' ) <EOL> mexpect = '<STR_LIT>' % ( MSG_BEGIN , m . as_string ( ) , MSG_END ) <EOL> self . assertEqual ( self . output . getvalue ( ) , mexpect ) <EOL> def testSendMessageWithAddresses ( self ) : <EOL> m = email . mime . text . MIMEText ( '<STR_LIT>' ) <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> smtp . send_message ( m ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> smtp . quit ( ) <EOL> self . assertEqual ( m [ '<STR_LIT>' ] , '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . client_evt . set ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . output . flush ( ) <EOL> m [ '<STR_LIT>' ] = socket . gethostbyname ( '<STR_LIT:localhost>' ) <EOL> del m [ '<STR_LIT>' ] <EOL> mexpect = '<STR_LIT>' % ( MSG_BEGIN , m . as_string ( ) , MSG_END ) <EOL> self . assertEqual ( self . output . getvalue ( ) , mexpect ) <EOL> debugout = smtpd . DEBUGSTREAM . getvalue ( ) <EOL> sender = re . compile ( "<STR_LIT>" , re . MULTILINE ) <EOL> self . assertRegex ( debugout , sender ) <EOL> for addr in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ) : <EOL> to_addr = re . compile ( r"<STR_LIT>" . format ( addr ) , <EOL> re . MULTILINE ) <EOL> self . assertRegex ( debugout , to_addr ) <EOL> def testSendMessageWithSomeAddresses ( self ) : <EOL> m = email . mime . text . MIMEText ( '<STR_LIT>' ) <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> smtp . send_message ( m ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> smtp . quit ( ) <EOL> self . client_evt . set ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . output . flush ( ) <EOL> m [ '<STR_LIT>' ] = socket . gethostbyname ( '<STR_LIT:localhost>' ) <EOL> mexpect = '<STR_LIT>' % ( MSG_BEGIN , m . as_string ( ) , MSG_END ) <EOL> self . assertEqual ( self . output . getvalue ( ) , mexpect ) <EOL> debugout = smtpd . DEBUGSTREAM . getvalue ( ) <EOL> sender = re . compile ( "<STR_LIT>" , re . MULTILINE ) <EOL> self . assertRegex ( debugout , sender ) <EOL> for addr in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> to_addr = re . compile ( r"<STR_LIT>" . format ( addr ) , <EOL> re . MULTILINE ) <EOL> self . assertRegex ( debugout , to_addr ) <EOL> def testSendMessageWithSpecifiedAddresses ( self ) : <EOL> m = email . mime . text . MIMEText ( '<STR_LIT>' ) <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> smtp . send_message ( m , from_addr = '<STR_LIT>' , to_addrs = '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> smtp . quit ( ) <EOL> self . client_evt . set ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . output . flush ( ) <EOL> m [ '<STR_LIT>' ] = socket . gethostbyname ( '<STR_LIT:localhost>' ) <EOL> mexpect = '<STR_LIT>' % ( MSG_BEGIN , m . as_string ( ) , MSG_END ) <EOL> self . assertEqual ( self . output . getvalue ( ) , mexpect ) <EOL> debugout = smtpd . DEBUGSTREAM . getvalue ( ) <EOL> sender = re . compile ( "<STR_LIT>" , re . MULTILINE ) <EOL> self . assertRegex ( debugout , sender ) <EOL> for addr in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> to_addr = re . compile ( r"<STR_LIT>" . format ( addr ) , <EOL> re . MULTILINE ) <EOL> self . assertNotRegex ( debugout , to_addr ) <EOL> recip = re . compile ( r"<STR_LIT>" , re . MULTILINE ) <EOL> self . assertRegex ( debugout , recip ) <EOL> def testSendMessageWithMultipleFrom ( self ) : <EOL> m = email . mime . text . MIMEText ( '<STR_LIT>' ) <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> smtp . send_message ( m ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> smtp . quit ( ) <EOL> self . client_evt . set ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . output . flush ( ) <EOL> m [ '<STR_LIT>' ] = socket . gethostbyname ( '<STR_LIT:localhost>' ) <EOL> mexpect = '<STR_LIT>' % ( MSG_BEGIN , m . as_string ( ) , MSG_END ) <EOL> self . assertEqual ( self . output . getvalue ( ) , mexpect ) <EOL> debugout = smtpd . DEBUGSTREAM . getvalue ( ) <EOL> sender = re . compile ( "<STR_LIT>" , re . MULTILINE ) <EOL> self . assertRegex ( debugout , sender ) <EOL> for addr in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> to_addr = re . compile ( r"<STR_LIT>" . format ( addr ) , <EOL> re . MULTILINE ) <EOL> self . assertRegex ( debugout , to_addr ) <EOL> def testSendMessageResent ( self ) : <EOL> m = email . mime . text . MIMEText ( '<STR_LIT>' ) <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> smtp . send_message ( m ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> smtp . quit ( ) <EOL> self . client_evt . set ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . output . flush ( ) <EOL> del m [ '<STR_LIT>' ] <EOL> del m [ '<STR_LIT>' ] <EOL> m [ '<STR_LIT>' ] = socket . gethostbyname ( '<STR_LIT:localhost>' ) <EOL> mexpect = '<STR_LIT>' % ( MSG_BEGIN , m . as_string ( ) , MSG_END ) <EOL> self . assertEqual ( self . output . getvalue ( ) , mexpect ) <EOL> debugout = smtpd . DEBUGSTREAM . getvalue ( ) <EOL> sender = re . compile ( "<STR_LIT>" , re . MULTILINE ) <EOL> self . assertRegex ( debugout , sender ) <EOL> for addr in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> to_addr = re . compile ( r"<STR_LIT>" . format ( addr ) , <EOL> re . MULTILINE ) <EOL> self . assertRegex ( debugout , to_addr ) <EOL> def testSendMessageMultipleResentRaises ( self ) : <EOL> m = email . mime . text . MIMEText ( '<STR_LIT>' ) <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> m [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:3> ) <EOL> with self . assertRaises ( ValueError ) : <EOL> smtp . send_message ( m ) <EOL> smtp . close ( ) <EOL> class NonConnectingTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> smtplib . socket = mock_socket <EOL> def tearDown ( self ) : <EOL> smtplib . socket = socket <EOL> def testNotConnected ( self ) : <EOL> smtp = smtplib . SMTP ( ) <EOL> self . assertRaises ( smtplib . SMTPServerDisconnected , smtp . ehlo ) <EOL> self . assertRaises ( smtplib . SMTPServerDisconnected , <EOL> smtp . send , '<STR_LIT>' ) <EOL> def testNonnumericPort ( self ) : <EOL> self . assertRaises ( mock_socket . error , smtplib . SMTP , <EOL> "<STR_LIT:localhost>" , "<STR_LIT>" ) <EOL> self . assertRaises ( mock_socket . error , smtplib . SMTP , <EOL> "<STR_LIT>" ) <EOL> @ unittest . skipUnless ( threading , '<STR_LIT>' ) <EOL> class BadHELOServerTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> smtplib . socket = mock_socket <EOL> mock_socket . reply_with ( b"<STR_LIT>" ) <EOL> self . old_stdout = sys . stdout <EOL> self . output = io . StringIO ( ) <EOL> sys . stdout = self . output <EOL> self . port = <NUM_LIT> <EOL> def tearDown ( self ) : <EOL> smtplib . socket = socket <EOL> sys . stdout = self . old_stdout <EOL> def testFailingHELO ( self ) : <EOL> self . assertRaises ( smtplib . SMTPConnectError , smtplib . SMTP , <EOL> HOST , self . port , '<STR_LIT:localhost>' , <NUM_LIT:3> ) <EOL> sim_users = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> sim_auth = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> sim_cram_md5_challenge = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> sim_auth_credentials = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> } <EOL> sim_auth_login_password = '<STR_LIT>' <EOL> sim_lists = { '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , ] , <EOL> } <EOL> class SimSMTPChannel ( smtpd . SMTPChannel ) : <EOL> quit_response = None <EOL> mail_response = None <EOL> rcpt_response = None <EOL> data_response = None <EOL> rcpt_count = <NUM_LIT:0> <EOL> rset_count = <NUM_LIT:0> <EOL> def __init__ ( self , extra_features , * args , ** kw ) : <EOL> self . _extrafeatures = '<STR_LIT>' . join ( <EOL> [ "<STR_LIT>" . format ( x ) for x in extra_features ] ) <EOL> super ( SimSMTPChannel , self ) . __init__ ( * args , ** kw ) <EOL> def smtp_EHLO ( self , arg ) : <EOL> resp = ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> resp = resp + self . _extrafeatures + '<STR_LIT>' <EOL> self . push ( resp ) <EOL> self . seen_greeting = arg <EOL> self . extended_smtp = True <EOL> def smtp_VRFY ( self , arg ) : <EOL> if arg in sim_users : <EOL> self . push ( '<STR_LIT>' % ( sim_users [ arg ] , smtplib . quoteaddr ( arg ) ) ) <EOL> else : <EOL> self . push ( '<STR_LIT>' % arg ) <EOL> def smtp_EXPN ( self , arg ) : <EOL> list_name = arg . lower ( ) <EOL> if list_name in sim_lists : <EOL> user_list = sim_lists [ list_name ] <EOL> for n , user_email in enumerate ( user_list ) : <EOL> quoted_addr = smtplib . quoteaddr ( user_email ) <EOL> if n < len ( user_list ) - <NUM_LIT:1> : <EOL> self . push ( '<STR_LIT>' % ( sim_users [ user_email ] , quoted_addr ) ) <EOL> else : <EOL> self . push ( '<STR_LIT>' % ( sim_users [ user_email ] , quoted_addr ) ) <EOL> else : <EOL> self . push ( '<STR_LIT>' ) <EOL> def smtp_AUTH ( self , arg ) : <EOL> if arg . strip ( ) . lower ( ) == '<STR_LIT>' : <EOL> self . push ( '<STR_LIT>' . format ( sim_cram_md5_challenge ) ) <EOL> return <EOL> mech , auth = arg . split ( ) <EOL> mech = mech . lower ( ) <EOL> if mech not in sim_auth_credentials : <EOL> self . push ( '<STR_LIT>' ) <EOL> return <EOL> if mech == '<STR_LIT>' and auth == sim_auth_credentials [ '<STR_LIT>' ] : <EOL> self . push ( '<STR_LIT>' ) <EOL> elif mech == '<STR_LIT>' and auth == sim_auth_credentials [ '<STR_LIT>' ] : <EOL> self . push ( '<STR_LIT>' ) <EOL> else : <EOL> self . push ( '<STR_LIT>' ) <EOL> def smtp_QUIT ( self , arg ) : <EOL> if self . quit_response is None : <EOL> super ( SimSMTPChannel , self ) . smtp_QUIT ( arg ) <EOL> else : <EOL> self . push ( self . quit_response ) <EOL> self . close_when_done ( ) <EOL> def smtp_MAIL ( self , arg ) : <EOL> if self . mail_response is None : <EOL> super ( ) . smtp_MAIL ( arg ) <EOL> else : <EOL> self . push ( self . mail_response ) <EOL> def smtp_RCPT ( self , arg ) : <EOL> if self . rcpt_response is None : <EOL> super ( ) . smtp_RCPT ( arg ) <EOL> return <EOL> self . rcpt_count += <NUM_LIT:1> <EOL> self . push ( self . rcpt_response [ self . rcpt_count - <NUM_LIT:1> ] ) <EOL> def smtp_RSET ( self , arg ) : <EOL> self . rset_count += <NUM_LIT:1> <EOL> super ( ) . smtp_RSET ( arg ) <EOL> def smtp_DATA ( self , arg ) : <EOL> if self . data_response is None : <EOL> super ( ) . smtp_DATA ( arg ) <EOL> else : <EOL> self . push ( self . data_response ) <EOL> def handle_error ( self ) : <EOL> raise <EOL> class SimSMTPServer ( smtpd . SMTPServer ) : <EOL> channel_class = SimSMTPChannel <EOL> def __init__ ( self , * args , ** kw ) : <EOL> self . _extra_features = [ ] <EOL> smtpd . SMTPServer . __init__ ( self , * args , ** kw ) <EOL> def handle_accepted ( self , conn , addr ) : <EOL> self . _SMTPchannel = self . channel_class ( <EOL> self . _extra_features , self , conn , addr ) <EOL> def process_message ( self , peer , mailfrom , rcpttos , data ) : <EOL> pass <EOL> def add_feature ( self , feature ) : <EOL> self . _extra_features . append ( feature ) <EOL> def handle_error ( self ) : <EOL> raise <EOL> @ unittest . skipUnless ( threading , '<STR_LIT>' ) <EOL> class SMTPSimTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . real_getfqdn = socket . getfqdn <EOL> socket . getfqdn = mock_socket . getfqdn <EOL> self . serv_evt = threading . Event ( ) <EOL> self . client_evt = threading . Event ( ) <EOL> self . serv = SimSMTPServer ( ( HOST , <NUM_LIT:0> ) , ( '<STR_LIT>' , - <NUM_LIT:1> ) ) <EOL> self . port = self . serv . socket . getsockname ( ) [ <NUM_LIT:1> ] <EOL> serv_args = ( self . serv , self . serv_evt , self . client_evt ) <EOL> self . thread = threading . Thread ( target = debugging_server , args = serv_args ) <EOL> self . thread . start ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . serv_evt . clear ( ) <EOL> def tearDown ( self ) : <EOL> socket . getfqdn = self . real_getfqdn <EOL> self . client_evt . set ( ) <EOL> self . serv_evt . wait ( ) <EOL> self . thread . join ( ) <EOL> def testBasic ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:15> ) <EOL> smtp . quit ( ) <EOL> def testEHLO ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:15> ) <EOL> self . assertEqual ( smtp . esmtp_features , { } ) <EOL> expected_features = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:size>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> smtp . ehlo ( ) <EOL> self . assertEqual ( smtp . esmtp_features , expected_features ) <EOL> for k in expected_features : <EOL> self . assertTrue ( smtp . has_extn ( k ) ) <EOL> self . assertFalse ( smtp . has_extn ( '<STR_LIT>' ) ) <EOL> smtp . quit ( ) <EOL> def testVRFY ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:15> ) <EOL> for email , name in sim_users . items ( ) : <EOL> expected_known = ( <NUM_LIT> , bytes ( '<STR_LIT>' % <EOL> ( name , smtplib . quoteaddr ( email ) ) , <EOL> "<STR_LIT:ascii>" ) ) <EOL> self . assertEqual ( smtp . vrfy ( email ) , expected_known ) <EOL> u = '<STR_LIT>' <EOL> expected_unknown = ( <NUM_LIT> , ( '<STR_LIT>' % u ) . encode ( '<STR_LIT:ascii>' ) ) <EOL> self . assertEqual ( smtp . vrfy ( u ) , expected_unknown ) <EOL> smtp . quit ( ) <EOL> def testEXPN ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:15> ) <EOL> for listname , members in sim_lists . items ( ) : <EOL> users = [ ] <EOL> for m in members : <EOL> users . append ( '<STR_LIT>' % ( sim_users [ m ] , smtplib . quoteaddr ( m ) ) ) <EOL> expected_known = ( <NUM_LIT> , bytes ( '<STR_LIT:\n>' . join ( users ) , "<STR_LIT:ascii>" ) ) <EOL> self . assertEqual ( smtp . expn ( listname ) , expected_known ) <EOL> u = '<STR_LIT>' <EOL> expected_unknown = ( <NUM_LIT> , b'<STR_LIT>' ) <EOL> self . assertEqual ( smtp . expn ( u ) , expected_unknown ) <EOL> smtp . quit ( ) <EOL> def testAUTH_PLAIN ( self ) : <EOL> self . serv . add_feature ( "<STR_LIT>" ) <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:15> ) <EOL> expected_auth_ok = ( <NUM_LIT> , b'<STR_LIT>' ) <EOL> self . assertEqual ( smtp . login ( sim_auth [ <NUM_LIT:0> ] , sim_auth [ <NUM_LIT:1> ] ) , expected_auth_ok ) <EOL> smtp . close ( ) <EOL> def testAUTH_LOGIN ( self ) : <EOL> self . serv . add_feature ( "<STR_LIT>" ) <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:15> ) <EOL> try : smtp . login ( sim_auth [ <NUM_LIT:0> ] , sim_auth [ <NUM_LIT:1> ] ) <EOL> except smtplib . SMTPAuthenticationError as err : <EOL> self . assertIn ( sim_auth_login_password , str ( err ) ) <EOL> smtp . close ( ) <EOL> def testAUTH_CRAM_MD5 ( self ) : <EOL> self . serv . add_feature ( "<STR_LIT>" ) <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:15> ) <EOL> try : smtp . login ( sim_auth [ <NUM_LIT:0> ] , sim_auth [ <NUM_LIT:1> ] ) <EOL> except smtplib . SMTPAuthenticationError as err : <EOL> self . assertIn ( sim_auth_credentials [ '<STR_LIT>' ] , str ( err ) ) <EOL> smtp . close ( ) <EOL> def test_with_statement ( self ) : <EOL> with smtplib . SMTP ( HOST , self . port ) as smtp : <EOL> code , message = smtp . noop ( ) <EOL> self . assertEqual ( code , <NUM_LIT> ) <EOL> self . assertRaises ( smtplib . SMTPServerDisconnected , smtp . send , b'<STR_LIT:foo>' ) <EOL> with smtplib . SMTP ( HOST , self . port ) as smtp : <EOL> smtp . close ( ) <EOL> self . assertRaises ( smtplib . SMTPServerDisconnected , smtp . send , b'<STR_LIT:foo>' ) <EOL> def test_with_statement_QUIT_failure ( self ) : <EOL> with self . assertRaises ( smtplib . SMTPResponseException ) as error : <EOL> with smtplib . SMTP ( HOST , self . port ) as smtp : <EOL> smtp . noop ( ) <EOL> self . serv . _SMTPchannel . quit_response = '<STR_LIT>' <EOL> self . assertEqual ( error . exception . smtp_code , <NUM_LIT> ) <EOL> self . assertEqual ( error . exception . smtp_error , b'<STR_LIT>' ) <EOL> def test_421_from_mail_cmd ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:15> ) <EOL> smtp . noop ( ) <EOL> self . serv . _SMTPchannel . mail_response = '<STR_LIT>' <EOL> with self . assertRaises ( smtplib . SMTPSenderRefused ) : <EOL> smtp . sendmail ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertIsNone ( smtp . sock ) <EOL> self . assertEqual ( self . serv . _SMTPchannel . rset_count , <NUM_LIT:0> ) <EOL> def test_421_from_rcpt_cmd ( self ) : <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:15> ) <EOL> smtp . noop ( ) <EOL> self . serv . _SMTPchannel . rcpt_response = [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> with self . assertRaises ( smtplib . SMTPRecipientsRefused ) as r : <EOL> smtp . sendmail ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertIsNone ( smtp . sock ) <EOL> self . assertEqual ( self . serv . _SMTPchannel . rset_count , <NUM_LIT:0> ) <EOL> self . assertDictEqual ( r . exception . args [ <NUM_LIT:0> ] , { '<STR_LIT>' : ( <NUM_LIT> , b'<STR_LIT>' ) } ) <EOL> def test_421_from_data_cmd ( self ) : <EOL> class MySimSMTPChannel ( SimSMTPChannel ) : <EOL> def found_terminator ( self ) : <EOL> if self . smtp_state == self . DATA : <EOL> self . push ( '<STR_LIT>' ) <EOL> else : <EOL> super ( ) . found_terminator ( ) <EOL> self . serv . channel_class = MySimSMTPChannel <EOL> smtp = smtplib . SMTP ( HOST , self . port , local_hostname = '<STR_LIT:localhost>' , timeout = <NUM_LIT:15> ) <EOL> smtp . noop ( ) <EOL> with self . assertRaises ( smtplib . SMTPDataError ) : <EOL> smtp . sendmail ( '<STR_LIT>' , [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> self . assertIsNone ( smtp . sock ) <EOL> self . assertEqual ( self . serv . _SMTPchannel . rcpt_count , <NUM_LIT:0> ) <EOL> @ support . reap_threads <EOL> def test_main ( verbose = None ) : <EOL> support . run_unittest ( GeneralTests , DebuggingServerTests , <EOL> NonConnectingTests , <EOL> BadHELOServerTests , SMTPSimTests ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> test_main ( ) </s>
