<s> """<STR_LIT>""" <EOL> from zope . interface import implements <EOL> from twisted . python . util import LineLog <EOL> from twisted . trial import unittest , util <EOL> from twisted . protocols import basic , loopback <EOL> from twisted . mail import smtp <EOL> from twisted . internet import defer , protocol , reactor , interfaces <EOL> from twisted . internet import address , error , task <EOL> from twisted . test . proto_helpers import StringTransport <EOL> from twisted import cred <EOL> import twisted . cred . error <EOL> import twisted . cred . portal <EOL> import twisted . cred . checkers <EOL> import twisted . cred . credentials <EOL> from twisted . cred . portal import IRealm , Portal <EOL> from twisted . cred . checkers import ICredentialsChecker , AllowAnonymousAccess <EOL> from twisted . cred . credentials import IAnonymous <EOL> from twisted . cred . error import UnauthorizedLogin <EOL> from twisted . mail import imap4 <EOL> try : <EOL> from twisted . test . ssl_helpers import ClientTLSContext , ServerTLSContext <EOL> except ImportError : <EOL> ClientTLSContext = ServerTLSContext = None <EOL> import re <EOL> try : <EOL> from cStringIO import StringIO <EOL> except ImportError : <EOL> from StringIO import StringIO <EOL> def spameater ( * spam , ** eggs ) : <EOL> return None <EOL> class BrokenMessage ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( smtp . IMessage ) <EOL> def __init__ ( self , user ) : <EOL> pass <EOL> def lineReceived ( self , line ) : <EOL> pass <EOL> def eomReceived ( self ) : <EOL> raise RuntimeError ( "<STR_LIT>" ) <EOL> def connectionLost ( self ) : <EOL> pass <EOL> class DummyMessage ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , domain , user ) : <EOL> self . domain = domain <EOL> self . user = user <EOL> self . buffer = [ ] <EOL> def lineReceived ( self , line ) : <EOL> if not re . match ( '<STR_LIT>' , line ) : <EOL> self . buffer . append ( line ) <EOL> def eomReceived ( self ) : <EOL> message = '<STR_LIT:\n>' . join ( self . buffer ) + '<STR_LIT:\n>' <EOL> self . domain . messages [ self . user . dest . local ] . append ( message ) <EOL> deferred = defer . Deferred ( ) <EOL> deferred . callback ( "<STR_LIT>" ) <EOL> return deferred <EOL> class DummyDomain ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , names ) : <EOL> self . messages = { } <EOL> for name in names : <EOL> self . messages [ name ] = [ ] <EOL> def exists ( self , user ) : <EOL> if user . dest . local in self . messages : <EOL> return defer . succeed ( lambda : self . startMessage ( user ) ) <EOL> return defer . fail ( smtp . SMTPBadRcpt ( user ) ) <EOL> def startMessage ( self , user ) : <EOL> return DummyMessage ( self , user ) <EOL> class SMTPTestCase ( unittest . TestCase ) : <EOL> messages = [ ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] , '''<STR_LIT>''' ) ] <EOL> mbox = { '<STR_LIT:foo>' : [ '<STR_LIT>' ] } <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . factory = smtp . SMTPFactory ( ) <EOL> self . factory . domains = { } <EOL> self . factory . domains [ '<STR_LIT>' ] = DummyDomain ( [ '<STR_LIT:foo>' ] ) <EOL> self . transport = StringTransport ( ) <EOL> def testMessages ( self ) : <EOL> from twisted . mail import protocols <EOL> protocol = protocols . DomainSMTP ( ) <EOL> protocol . service = self . factory <EOL> protocol . factory = self . factory <EOL> protocol . receivedHeader = spameater <EOL> protocol . makeConnection ( self . transport ) <EOL> protocol . lineReceived ( '<STR_LIT>' ) <EOL> for message in self . messages : <EOL> protocol . lineReceived ( '<STR_LIT>' % message [ <NUM_LIT:0> ] ) <EOL> for target in message [ <NUM_LIT:1> ] : <EOL> protocol . lineReceived ( '<STR_LIT>' % target ) <EOL> protocol . lineReceived ( '<STR_LIT>' ) <EOL> protocol . dataReceived ( message [ <NUM_LIT:2> ] ) <EOL> protocol . lineReceived ( '<STR_LIT:.>' ) <EOL> protocol . lineReceived ( '<STR_LIT>' ) <EOL> if self . mbox != self . factory . domains [ '<STR_LIT>' ] . messages : <EOL> raise AssertionError ( self . factory . domains [ '<STR_LIT>' ] . messages ) <EOL> protocol . setTimeout ( None ) <EOL> testMessages . suppress = [ util . suppress ( message = '<STR_LIT>' , category = DeprecationWarning ) ] <EOL> mail = '''<STR_LIT>''' <EOL> class MyClient : <EOL> def __init__ ( self , messageInfo = None ) : <EOL> if messageInfo is None : <EOL> messageInfo = ( <EOL> '<STR_LIT>' , [ '<STR_LIT>' ] , StringIO ( mail ) ) <EOL> self . _sender = messageInfo [ <NUM_LIT:0> ] <EOL> self . _recipient = messageInfo [ <NUM_LIT:1> ] <EOL> self . _data = messageInfo [ <NUM_LIT:2> ] <EOL> def getMailFrom ( self ) : <EOL> return self . _sender <EOL> def getMailTo ( self ) : <EOL> return self . _recipient <EOL> def getMailData ( self ) : <EOL> return self . _data <EOL> def sendError ( self , exc ) : <EOL> self . _error = exc <EOL> def sentMail ( self , code , resp , numOk , addresses , log ) : <EOL> self . _sender = None <EOL> self . _recipient = None <EOL> self . _data = None <EOL> class MySMTPClient ( MyClient , smtp . SMTPClient ) : <EOL> def __init__ ( self , messageInfo = None ) : <EOL> smtp . SMTPClient . __init__ ( self , '<STR_LIT>' ) <EOL> MyClient . __init__ ( self , messageInfo ) <EOL> class MyESMTPClient ( MyClient , smtp . ESMTPClient ) : <EOL> def __init__ ( self , secret = '<STR_LIT>' , contextFactory = None ) : <EOL> smtp . ESMTPClient . __init__ ( self , secret , contextFactory , '<STR_LIT>' ) <EOL> MyClient . __init__ ( self ) <EOL> class LoopbackMixin : <EOL> def loopback ( self , server , client ) : <EOL> return loopback . loopbackTCP ( server , client ) <EOL> class LoopbackTestCase ( LoopbackMixin ) : <EOL> def testMessages ( self ) : <EOL> factory = smtp . SMTPFactory ( ) <EOL> factory . domains = { } <EOL> factory . domains [ '<STR_LIT>' ] = DummyDomain ( [ '<STR_LIT>' ] ) <EOL> from twisted . mail . protocols import DomainSMTP <EOL> protocol = DomainSMTP ( ) <EOL> protocol . service = factory <EOL> protocol . factory = factory <EOL> clientProtocol = self . clientClass ( ) <EOL> return self . loopback ( protocol , clientProtocol ) <EOL> testMessages . suppress = [ util . suppress ( message = '<STR_LIT>' , category = DeprecationWarning ) ] <EOL> class LoopbackSMTPTestCase ( LoopbackTestCase , unittest . TestCase ) : <EOL> clientClass = MySMTPClient <EOL> class LoopbackESMTPTestCase ( LoopbackTestCase , unittest . TestCase ) : <EOL> clientClass = MyESMTPClient <EOL> class FakeSMTPServer ( basic . LineReceiver ) : <EOL> clientData = [ <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' <EOL> ] <EOL> def connectionMade ( self ) : <EOL> self . buffer = [ ] <EOL> self . clientData = self . clientData [ : ] <EOL> self . clientData . reverse ( ) <EOL> self . sendLine ( self . clientData . pop ( ) ) <EOL> def lineReceived ( self , line ) : <EOL> self . buffer . append ( line ) <EOL> if line == "<STR_LIT>" : <EOL> self . transport . write ( "<STR_LIT>" ) <EOL> self . transport . loseConnection ( ) <EOL> elif line == "<STR_LIT:.>" : <EOL> self . transport . write ( "<STR_LIT>" ) <EOL> elif line == "<STR_LIT>" : <EOL> self . transport . loseConnection ( ) <EOL> if self . clientData : <EOL> self . sendLine ( self . clientData . pop ( ) ) <EOL> class SMTPClientTestCase ( unittest . TestCase , LoopbackMixin ) : <EOL> """<STR_LIT>""" <EOL> def test_timeoutConnection ( self ) : <EOL> """<STR_LIT>""" <EOL> error = [ ] <EOL> client = MySMTPClient ( ) <EOL> client . sendError = error . append <EOL> client . makeConnection ( StringTransport ( ) ) <EOL> client . lineReceived ( "<STR_LIT>" ) <EOL> client . timeoutConnection ( ) <EOL> self . assertIsInstance ( error [ <NUM_LIT:0> ] , smtp . SMTPTimeoutError ) <EOL> self . assertTrue ( error [ <NUM_LIT:0> ] . isFatal ) <EOL> self . assertEqual ( <EOL> str ( error [ <NUM_LIT:0> ] ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> expected_output = [ <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:.>' , '<STR_LIT>' <EOL> ] <EOL> def test_messages ( self ) : <EOL> """<STR_LIT>""" <EOL> client = MySMTPClient ( ) <EOL> server = FakeSMTPServer ( ) <EOL> d = self . loopback ( server , client ) <EOL> d . addCallback ( lambda x : <EOL> self . assertEqual ( server . buffer , self . expected_output ) ) <EOL> return d <EOL> def test_transferError ( self ) : <EOL> """<STR_LIT>""" <EOL> client = MySMTPClient ( <EOL> ( '<STR_LIT>' , [ '<STR_LIT>' ] , StringIO ( "<STR_LIT:foo>" ) ) ) <EOL> transport = StringTransport ( ) <EOL> client . makeConnection ( transport ) <EOL> client . dataReceived ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ) <EOL> self . assertNotIdentical ( transport . producer , None ) <EOL> self . assertFalse ( transport . streaming ) <EOL> transport . producer . stopProducing ( ) <EOL> self . assertIsInstance ( client . _error , Exception ) <EOL> def test_sendFatalError ( self ) : <EOL> """<STR_LIT>""" <EOL> client = smtp . SMTPClient ( None ) <EOL> transport = StringTransport ( ) <EOL> client . makeConnection ( transport ) <EOL> client . sendError ( smtp . SMTPClientError ( <NUM_LIT> , "<STR_LIT:foo>" , isFatal = True ) ) <EOL> self . assertEqual ( transport . value ( ) , "<STR_LIT>" ) <EOL> self . assertTrue ( transport . disconnecting ) <EOL> def test_sendNonFatalError ( self ) : <EOL> """<STR_LIT>""" <EOL> client = smtp . SMTPClient ( None ) <EOL> transport = StringTransport ( ) <EOL> client . makeConnection ( transport ) <EOL> client . sendError ( smtp . SMTPClientError ( <NUM_LIT> , "<STR_LIT:foo>" , isFatal = False ) ) <EOL> self . assertEqual ( transport . value ( ) , "<STR_LIT>" ) <EOL> self . assertFalse ( transport . disconnecting ) <EOL> def test_sendOtherError ( self ) : <EOL> """<STR_LIT>""" <EOL> client = smtp . SMTPClient ( None ) <EOL> transport = StringTransport ( ) <EOL> client . makeConnection ( transport ) <EOL> client . sendError ( Exception ( "<STR_LIT:foo>" ) ) <EOL> self . assertEqual ( transport . value ( ) , "<STR_LIT>" ) <EOL> self . assertTrue ( transport . disconnecting ) <EOL> class DummySMTPMessage : <EOL> def __init__ ( self , protocol , users ) : <EOL> self . protocol = protocol <EOL> self . users = users <EOL> self . buffer = [ ] <EOL> def lineReceived ( self , line ) : <EOL> self . buffer . append ( line ) <EOL> def eomReceived ( self ) : <EOL> message = '<STR_LIT:\n>' . join ( self . buffer ) + '<STR_LIT:\n>' <EOL> helo , origin = self . users [ <NUM_LIT:0> ] . helo [ <NUM_LIT:0> ] , str ( self . users [ <NUM_LIT:0> ] . orig ) <EOL> recipients = [ ] <EOL> for user in self . users : <EOL> recipients . append ( str ( user ) ) <EOL> self . protocol . message [ tuple ( recipients ) ] = ( helo , origin , recipients , message ) <EOL> return defer . succeed ( "<STR_LIT>" ) <EOL> class DummyProto : <EOL> def connectionMade ( self ) : <EOL> self . dummyMixinBase . connectionMade ( self ) <EOL> self . message = { } <EOL> def startMessage ( self , users ) : <EOL> return DummySMTPMessage ( self , users ) <EOL> def receivedHeader ( * spam ) : <EOL> return None <EOL> def validateTo ( self , user ) : <EOL> self . delivery = SimpleDelivery ( None ) <EOL> return lambda : self . startMessage ( [ user ] ) <EOL> def validateFrom ( self , helo , origin ) : <EOL> return origin <EOL> class DummySMTP ( DummyProto , smtp . SMTP ) : <EOL> dummyMixinBase = smtp . SMTP <EOL> class DummyESMTP ( DummyProto , smtp . ESMTP ) : <EOL> dummyMixinBase = smtp . ESMTP <EOL> class AnotherTestCase : <EOL> serverClass = None <EOL> clientClass = None <EOL> messages = [ ( '<STR_LIT>' , '<STR_LIT>' , [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' , [ '<STR_LIT>' ] , '''<STR_LIT>''' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] , '''<STR_LIT>''' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' , [ '<STR_LIT>' ] , '''<STR_LIT>''' ) , <EOL> ] <EOL> data = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' , None , None ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , None , None ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' , None , None ) , <EOL> ] <EOL> for helo_ , from_ , to_ , realfrom , realto , msg in messages : <EOL> data . append ( ( '<STR_LIT>' % from_ , '<STR_LIT>' , <EOL> None , None ) ) <EOL> for rcpt in to_ : <EOL> data . append ( ( '<STR_LIT>' % rcpt , '<STR_LIT>' , <EOL> None , None ) ) <EOL> data . append ( ( '<STR_LIT>' , '<STR_LIT>' , <EOL> msg , ( '<STR_LIT>' , <EOL> ( helo_ , realfrom , realto , msg ) ) ) ) <EOL> def test_buffer ( self ) : <EOL> """<STR_LIT>""" <EOL> transport = StringTransport ( ) <EOL> a = self . serverClass ( ) <EOL> class fooFactory : <EOL> domain = '<STR_LIT>' <EOL> a . factory = fooFactory ( ) <EOL> a . makeConnection ( transport ) <EOL> for ( send , expect , msg , msgexpect ) in self . data : <EOL> if send : <EOL> a . dataReceived ( send ) <EOL> data = transport . value ( ) <EOL> transport . clear ( ) <EOL> if not re . match ( expect , data ) : <EOL> raise AssertionError , ( send , expect , data ) <EOL> if data [ : <NUM_LIT:3> ] == '<STR_LIT>' : <EOL> for line in msg . splitlines ( ) : <EOL> if line and line [ <NUM_LIT:0> ] == '<STR_LIT:.>' : <EOL> line = '<STR_LIT:.>' + line <EOL> a . dataReceived ( line + '<STR_LIT:\r\n>' ) <EOL> a . dataReceived ( '<STR_LIT>' ) <EOL> data = transport . value ( ) <EOL> transport . clear ( ) <EOL> resp , msgdata = msgexpect <EOL> if not re . match ( resp , data ) : <EOL> raise AssertionError , ( resp , data ) <EOL> for recip in msgdata [ <NUM_LIT:2> ] : <EOL> expected = list ( msgdata [ : ] ) <EOL> expected [ <NUM_LIT:2> ] = [ recip ] <EOL> self . assertEqual ( <EOL> a . message [ ( recip , ) ] , <EOL> tuple ( expected ) <EOL> ) <EOL> a . setTimeout ( None ) <EOL> class AnotherESMTPTestCase ( AnotherTestCase , unittest . TestCase ) : <EOL> serverClass = DummyESMTP <EOL> clientClass = MyESMTPClient <EOL> class AnotherSMTPTestCase ( AnotherTestCase , unittest . TestCase ) : <EOL> serverClass = DummySMTP <EOL> clientClass = MySMTPClient <EOL> class DummyChecker : <EOL> implements ( cred . checkers . ICredentialsChecker ) <EOL> users = { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> credentialInterfaces = ( cred . credentials . IUsernamePassword , <EOL> cred . credentials . IUsernameHashedPassword ) <EOL> def requestAvatarId ( self , credentials ) : <EOL> return defer . maybeDeferred ( <EOL> credentials . checkPassword , self . users [ credentials . username ] <EOL> ) . addCallback ( self . _cbCheck , credentials . username ) <EOL> def _cbCheck ( self , result , username ) : <EOL> if result : <EOL> return username <EOL> raise cred . error . UnauthorizedLogin ( ) <EOL> class SimpleDelivery ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( smtp . IMessageDelivery ) <EOL> def __init__ ( self , messageFactory ) : <EOL> self . _messageFactory = messageFactory <EOL> def receivedHeader ( self , helo , origin , recipients ) : <EOL> return None <EOL> def validateFrom ( self , helo , origin ) : <EOL> return origin <EOL> def validateTo ( self , user ) : <EOL> return lambda : self . _messageFactory ( user ) <EOL> class DummyRealm : <EOL> def requestAvatar ( self , avatarId , mind , * interfaces ) : <EOL> return smtp . IMessageDelivery , SimpleDelivery ( None ) , lambda : None <EOL> class AuthTestCase ( unittest . TestCase , LoopbackMixin ) : <EOL> def test_crammd5Auth ( self ) : <EOL> """<STR_LIT>""" <EOL> realm = DummyRealm ( ) <EOL> p = cred . portal . Portal ( realm ) <EOL> p . registerChecker ( DummyChecker ( ) ) <EOL> server = DummyESMTP ( { '<STR_LIT>' : cred . credentials . CramMD5Credentials } ) <EOL> server . portal = p <EOL> client = MyESMTPClient ( '<STR_LIT>' ) <EOL> cAuth = smtp . CramMD5ClientAuthenticator ( '<STR_LIT>' ) <EOL> client . registerAuthenticator ( cAuth ) <EOL> d = self . loopback ( server , client ) <EOL> d . addCallback ( lambda x : self . assertEqual ( server . authenticated , <NUM_LIT:1> ) ) <EOL> return d <EOL> def test_loginAuth ( self ) : <EOL> """<STR_LIT>""" <EOL> realm = DummyRealm ( ) <EOL> p = cred . portal . Portal ( realm ) <EOL> p . registerChecker ( DummyChecker ( ) ) <EOL> server = DummyESMTP ( { '<STR_LIT>' : imap4 . LOGINCredentials } ) <EOL> server . portal = p <EOL> client = MyESMTPClient ( '<STR_LIT>' ) <EOL> cAuth = smtp . LOGINAuthenticator ( '<STR_LIT>' ) <EOL> client . registerAuthenticator ( cAuth ) <EOL> d = self . loopback ( server , client ) <EOL> d . addCallback ( lambda x : self . assertTrue ( server . authenticated ) ) <EOL> return d <EOL> def test_loginAgainstWeirdServer ( self ) : <EOL> """<STR_LIT>""" <EOL> realm = DummyRealm ( ) <EOL> p = cred . portal . Portal ( realm ) <EOL> p . registerChecker ( DummyChecker ( ) ) <EOL> server = DummyESMTP ( { '<STR_LIT>' : smtp . LOGINCredentials } ) <EOL> server . portal = p <EOL> client = MyESMTPClient ( '<STR_LIT>' ) <EOL> cAuth = smtp . LOGINAuthenticator ( '<STR_LIT>' ) <EOL> client . registerAuthenticator ( cAuth ) <EOL> d = self . loopback ( server , client ) <EOL> d . addCallback ( lambda x : self . assertTrue ( server . authenticated ) ) <EOL> return d <EOL> class SMTPHelperTestCase ( unittest . TestCase ) : <EOL> def testMessageID ( self ) : <EOL> d = { } <EOL> for i in range ( <NUM_LIT:1000> ) : <EOL> m = smtp . messageid ( '<STR_LIT>' ) <EOL> self . failIf ( m in d ) <EOL> d [ m ] = None <EOL> def testQuoteAddr ( self ) : <EOL> cases = [ <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ smtp . Address ( '<STR_LIT>' ) , '<STR_LIT>' ] , <EOL> [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> [ smtp . Address ( '<STR_LIT>' ) , '<STR_LIT>' ] , <EOL> ] <EOL> for ( c , e ) in cases : <EOL> self . assertEqual ( smtp . quoteaddr ( c ) , e ) <EOL> def testUser ( self ) : <EOL> u = smtp . User ( '<STR_LIT>' , '<STR_LIT>' , None , None ) <EOL> self . assertEqual ( str ( u ) , '<STR_LIT>' ) <EOL> def testXtextEncoding ( self ) : <EOL> cases = [ <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> ] <EOL> for ( case , expected ) in cases : <EOL> self . assertEqual ( smtp . xtext_encode ( case ) , ( expected , len ( case ) ) ) <EOL> self . assertEqual ( case . encode ( '<STR_LIT>' ) , expected ) <EOL> self . assertEqual ( <EOL> smtp . xtext_decode ( expected ) , ( case , len ( expected ) ) ) <EOL> self . assertEqual ( expected . decode ( '<STR_LIT>' ) , case ) <EOL> def test_encodeWithErrors ( self ) : <EOL> """<STR_LIT>""" <EOL> text = u'<STR_LIT>' <EOL> self . assertEqual ( <EOL> smtp . xtext_encode ( text , '<STR_LIT:strict>' ) , <EOL> ( text . encode ( '<STR_LIT>' ) , len ( text ) ) ) <EOL> self . assertEqual ( <EOL> text . encode ( '<STR_LIT>' , '<STR_LIT:strict>' ) , <EOL> text . encode ( '<STR_LIT>' ) ) <EOL> def test_decodeWithErrors ( self ) : <EOL> """<STR_LIT>""" <EOL> bytes = '<STR_LIT>' <EOL> self . assertEqual ( <EOL> smtp . xtext_decode ( bytes , '<STR_LIT:strict>' ) , <EOL> ( bytes . decode ( '<STR_LIT>' ) , len ( bytes ) ) ) <EOL> self . assertEqual ( <EOL> bytes . decode ( '<STR_LIT>' , '<STR_LIT:strict>' ) , <EOL> bytes . decode ( '<STR_LIT>' ) ) <EOL> class NoticeTLSClient ( MyESMTPClient ) : <EOL> tls = False <EOL> def esmtpState_starttls ( self , code , resp ) : <EOL> MyESMTPClient . esmtpState_starttls ( self , code , resp ) <EOL> self . tls = True <EOL> class TLSTestCase ( unittest . TestCase , LoopbackMixin ) : <EOL> def testTLS ( self ) : <EOL> clientCTX = ClientTLSContext ( ) <EOL> serverCTX = ServerTLSContext ( ) <EOL> client = NoticeTLSClient ( contextFactory = clientCTX ) <EOL> server = DummyESMTP ( contextFactory = serverCTX ) <EOL> def check ( ignored ) : <EOL> self . assertEqual ( client . tls , True ) <EOL> self . assertEqual ( server . startedTLS , True ) <EOL> return self . loopback ( server , client ) . addCallback ( check ) <EOL> if ClientTLSContext is None : <EOL> for case in ( TLSTestCase , ) : <EOL> case . skip = "<STR_LIT>" <EOL> if not interfaces . IReactorSSL . providedBy ( reactor ) : <EOL> for case in ( TLSTestCase , ) : <EOL> case . skip = "<STR_LIT>" <EOL> class EmptyLineTestCase ( unittest . TestCase ) : <EOL> def test_emptyLineSyntaxError ( self ) : <EOL> """<STR_LIT>""" <EOL> proto = smtp . SMTP ( ) <EOL> transport = StringTransport ( ) <EOL> proto . makeConnection ( transport ) <EOL> proto . lineReceived ( '<STR_LIT>' ) <EOL> proto . setTimeout ( None ) <EOL> out = transport . value ( ) . splitlines ( ) <EOL> self . assertEqual ( len ( out ) , <NUM_LIT:2> ) <EOL> self . failUnless ( out [ <NUM_LIT:0> ] . startswith ( '<STR_LIT>' ) ) <EOL> self . assertEqual ( out [ <NUM_LIT:1> ] , "<STR_LIT>" ) <EOL> class TimeoutTestCase ( unittest . TestCase , LoopbackMixin ) : <EOL> """<STR_LIT>""" <EOL> def _timeoutTest ( self , onDone , clientFactory ) : <EOL> """<STR_LIT>""" <EOL> clock = task . Clock ( ) <EOL> client = clientFactory . buildProtocol ( <EOL> address . IPv4Address ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> ) ) <EOL> client . callLater = clock . callLater <EOL> t = StringTransport ( ) <EOL> client . makeConnection ( t ) <EOL> t . protocol = client <EOL> def check ( ign ) : <EOL> self . assertEqual ( clock . seconds ( ) , <NUM_LIT:0.5> ) <EOL> d = self . assertFailure ( onDone , smtp . SMTPTimeoutError <EOL> ) . addCallback ( check ) <EOL> clock . advance ( <NUM_LIT:0.1> ) <EOL> clock . advance ( <NUM_LIT> ) <EOL> return d <EOL> def test_SMTPClient ( self ) : <EOL> """<STR_LIT>""" <EOL> onDone = defer . Deferred ( ) <EOL> clientFactory = smtp . SMTPSenderFactory ( <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> StringIO ( "<STR_LIT>" ) , onDone , <EOL> retries = <NUM_LIT:0> , timeout = <NUM_LIT:0.5> ) <EOL> return self . _timeoutTest ( onDone , clientFactory ) <EOL> def test_ESMTPClient ( self ) : <EOL> """<STR_LIT>""" <EOL> onDone = defer . Deferred ( ) <EOL> clientFactory = smtp . ESMTPSenderFactory ( <EOL> '<STR_LIT:username>' , '<STR_LIT:password>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> StringIO ( "<STR_LIT>" ) , onDone , <EOL> retries = <NUM_LIT:0> , timeout = <NUM_LIT:0.5> ) <EOL> return self . _timeoutTest ( onDone , clientFactory ) <EOL> def test_resetTimeoutWhileSending ( self ) : <EOL> """<STR_LIT>""" <EOL> class SlowFile : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , size ) : <EOL> self . _size = size <EOL> def read ( self , max = None ) : <EOL> if self . _size : <EOL> self . _size -= <NUM_LIT:1> <EOL> return '<STR_LIT:x>' <EOL> return '<STR_LIT>' <EOL> failed = [ ] <EOL> onDone = defer . Deferred ( ) <EOL> onDone . addErrback ( failed . append ) <EOL> clientFactory = smtp . SMTPSenderFactory ( <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> SlowFile ( <NUM_LIT:1> ) , onDone , retries = <NUM_LIT:0> , timeout = <NUM_LIT:3> ) <EOL> clientFactory . domain = "<STR_LIT>" <EOL> clock = task . Clock ( ) <EOL> client = clientFactory . buildProtocol ( <EOL> address . IPv4Address ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> ) ) <EOL> client . callLater = clock . callLater <EOL> transport = StringTransport ( ) <EOL> client . makeConnection ( transport ) <EOL> client . dataReceived ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> ) <EOL> self . assertNotIdentical ( transport . producer , None ) <EOL> self . assertFalse ( transport . streaming ) <EOL> clock . advance ( <NUM_LIT:2> ) <EOL> self . assertEqual ( failed , [ ] ) <EOL> transport . producer . resumeProducing ( ) <EOL> clock . advance ( <NUM_LIT:2> ) <EOL> self . assertEqual ( failed , [ ] ) <EOL> transport . producer . resumeProducing ( ) <EOL> client . dataReceived ( "<STR_LIT>" ) <EOL> self . assertEqual ( failed , [ ] ) <EOL> self . assertEqual ( <EOL> transport . value ( ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class MultipleDeliveryFactorySMTPServerFactory ( protocol . ServerFactory ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , messageFactories ) : <EOL> self . _messageFactories = messageFactories <EOL> def buildProtocol ( self , addr ) : <EOL> p = protocol . ServerFactory . buildProtocol ( self , addr ) <EOL> p . delivery = SimpleDelivery ( self . _messageFactories . pop ( <NUM_LIT:0> ) ) <EOL> return p <EOL> class SMTPSenderFactoryRetryTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_retryAfterDisconnect ( self ) : <EOL> """<STR_LIT>""" <EOL> recipient = '<STR_LIT>' <EOL> message = "<STR_LIT>" <EOL> domain = DummyDomain ( [ recipient ] ) <EOL> class CleanSMTP ( smtp . SMTP ) : <EOL> """<STR_LIT>""" <EOL> def makeConnection ( innerSelf , transport ) : <EOL> self . addCleanup ( transport . loseConnection ) <EOL> smtp . SMTP . makeConnection ( innerSelf , transport ) <EOL> serverFactory = MultipleDeliveryFactorySMTPServerFactory ( [ <EOL> BrokenMessage , <EOL> lambda user : DummyMessage ( domain , user ) ] ) <EOL> serverFactory . protocol = CleanSMTP <EOL> serverPort = reactor . listenTCP ( <NUM_LIT:0> , serverFactory , interface = '<STR_LIT:127.0.0.1>' ) <EOL> serverHost = serverPort . getHost ( ) <EOL> self . addCleanup ( serverPort . stopListening ) <EOL> sentDeferred = defer . Deferred ( ) <EOL> clientFactory = smtp . SMTPSenderFactory ( <EOL> "<STR_LIT>" , recipient + "<STR_LIT>" , <EOL> StringIO ( message ) , sentDeferred ) <EOL> clientFactory . domain = "<STR_LIT>" <EOL> clientConnector = reactor . connectTCP ( <EOL> serverHost . host , serverHost . port , clientFactory ) <EOL> self . addCleanup ( clientConnector . disconnect ) <EOL> def cbSent ( ignored ) : <EOL> """<STR_LIT>""" <EOL> self . assertEqual ( <EOL> domain . messages , <EOL> { recipient : [ "<STR_LIT>" % ( message , ) ] } ) <EOL> self . assertEqual ( len ( self . flushLoggedErrors ( RuntimeError ) ) , <NUM_LIT:1> ) <EOL> sentDeferred . addCallback ( cbSent ) <EOL> return sentDeferred <EOL> class SingletonRealm ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( IRealm ) <EOL> def __init__ ( self , interface , avatar ) : <EOL> self . interface = interface <EOL> self . avatar = avatar <EOL> def requestAvatar ( self , avatarId , mind , * interfaces ) : <EOL> for iface in interfaces : <EOL> if iface is self . interface : <EOL> return iface , self . avatar , lambda : None <EOL> class NotImplementedDelivery ( object ) : <EOL> """<STR_LIT>""" <EOL> def validateFrom ( self , helo , origin ) : <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def validateTo ( self , user ) : <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> def receivedHeader ( self , helo , origin , recipients ) : <EOL> raise NotImplementedError ( "<STR_LIT>" ) <EOL> class SMTPServerTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def testSMTPGreetingHost ( self , serverClass = smtp . SMTP ) : <EOL> """<STR_LIT>""" <EOL> s = serverClass ( ) <EOL> s . host = "<STR_LIT>" <EOL> t = StringTransport ( ) <EOL> s . makeConnection ( t ) <EOL> s . connectionLost ( error . ConnectionDone ( ) ) <EOL> self . assertIn ( "<STR_LIT>" , t . value ( ) ) <EOL> def testSMTPGreetingNotExtended ( self ) : <EOL> """<STR_LIT>""" <EOL> s = smtp . SMTP ( ) <EOL> t = StringTransport ( ) <EOL> s . makeConnection ( t ) <EOL> s . connectionLost ( error . ConnectionDone ( ) ) <EOL> self . assertNotIn ( "<STR_LIT>" , t . value ( ) ) <EOL> def testESMTPGreetingHost ( self ) : <EOL> """<STR_LIT>""" <EOL> self . testSMTPGreetingHost ( smtp . ESMTP ) <EOL> def testESMTPGreetingExtended ( self ) : <EOL> """<STR_LIT>""" <EOL> s = smtp . ESMTP ( ) <EOL> t = StringTransport ( ) <EOL> s . makeConnection ( t ) <EOL> s . connectionLost ( error . ConnectionDone ( ) ) <EOL> self . assertIn ( "<STR_LIT>" , t . value ( ) ) <EOL> def test_acceptSenderAddress ( self ) : <EOL> """<STR_LIT>""" <EOL> class AcceptanceDelivery ( NotImplementedDelivery ) : <EOL> """<STR_LIT>""" <EOL> def validateFrom ( self , helo , origin ) : <EOL> return origin <EOL> realm = SingletonRealm ( smtp . IMessageDelivery , AcceptanceDelivery ( ) ) <EOL> portal = Portal ( realm , [ AllowAnonymousAccess ( ) ] ) <EOL> proto = smtp . SMTP ( ) <EOL> proto . portal = portal <EOL> trans = StringTransport ( ) <EOL> proto . makeConnection ( trans ) <EOL> proto . dataReceived ( '<STR_LIT>' ) <EOL> trans . clear ( ) <EOL> proto . dataReceived ( '<STR_LIT>' ) <EOL> proto . connectionLost ( error . ConnectionLost ( ) ) <EOL> self . assertEqual ( <EOL> trans . value ( ) , <EOL> '<STR_LIT>' ) <EOL> def test_deliveryRejectedSenderAddress ( self ) : <EOL> """<STR_LIT>""" <EOL> class RejectionDelivery ( NotImplementedDelivery ) : <EOL> """<STR_LIT>""" <EOL> def validateFrom ( self , helo , origin ) : <EOL> raise smtp . SMTPBadSender ( origin ) <EOL> realm = SingletonRealm ( smtp . IMessageDelivery , RejectionDelivery ( ) ) <EOL> portal = Portal ( realm , [ AllowAnonymousAccess ( ) ] ) <EOL> proto = smtp . SMTP ( ) <EOL> proto . portal = portal <EOL> trans = StringTransport ( ) <EOL> proto . makeConnection ( trans ) <EOL> proto . dataReceived ( '<STR_LIT>' ) <EOL> trans . clear ( ) <EOL> proto . dataReceived ( '<STR_LIT>' ) <EOL> proto . connectionLost ( error . ConnectionLost ( ) ) <EOL> self . assertEqual ( <EOL> trans . value ( ) , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_portalRejectedSenderAddress ( self ) : <EOL> """<STR_LIT>""" <EOL> class DisallowAnonymousAccess ( object ) : <EOL> """<STR_LIT>""" <EOL> implements ( ICredentialsChecker ) <EOL> credentialInterfaces = ( IAnonymous , ) <EOL> def requestAvatarId ( self , credentials ) : <EOL> return defer . fail ( UnauthorizedLogin ( ) ) <EOL> realm = SingletonRealm ( smtp . IMessageDelivery , NotImplementedDelivery ( ) ) <EOL> portal = Portal ( realm , [ DisallowAnonymousAccess ( ) ] ) <EOL> proto = smtp . SMTP ( ) <EOL> proto . portal = portal <EOL> trans = StringTransport ( ) <EOL> proto . makeConnection ( trans ) <EOL> proto . dataReceived ( '<STR_LIT>' ) <EOL> trans . clear ( ) <EOL> proto . dataReceived ( '<STR_LIT>' ) <EOL> proto . connectionLost ( error . ConnectionLost ( ) ) <EOL> self . assertEqual ( <EOL> trans . value ( ) , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_portalRejectedAnonymousSender ( self ) : <EOL> """<STR_LIT>""" <EOL> realm = SingletonRealm ( smtp . IMessageDelivery , NotImplementedDelivery ( ) ) <EOL> portal = Portal ( realm , [ ] ) <EOL> proto = smtp . SMTP ( ) <EOL> proto . portal = portal <EOL> trans = StringTransport ( ) <EOL> proto . makeConnection ( trans ) <EOL> proto . dataReceived ( '<STR_LIT>' ) <EOL> trans . clear ( ) <EOL> proto . dataReceived ( '<STR_LIT>' ) <EOL> proto . connectionLost ( error . ConnectionLost ( ) ) <EOL> self . assertEqual ( <EOL> trans . value ( ) , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> class ESMTPAuthenticationTestCase ( unittest . TestCase ) : <EOL> def assertServerResponse ( self , bytes , response ) : <EOL> """<STR_LIT>""" <EOL> self . transport . clear ( ) <EOL> self . server . dataReceived ( bytes ) <EOL> self . assertEqual ( <EOL> response , <EOL> self . transport . value ( ) . splitlines ( ) ) <EOL> def assertServerAuthenticated ( self , loginArgs , username = "<STR_LIT:username>" , password = "<STR_LIT:password>" ) : <EOL> """<STR_LIT>""" <EOL> d , credentials , mind , interfaces = loginArgs . pop ( ) <EOL> self . assertEqual ( loginArgs , [ ] ) <EOL> self . failUnless ( twisted . cred . credentials . IUsernamePassword . providedBy ( credentials ) ) <EOL> self . assertEqual ( credentials . username , username ) <EOL> self . failUnless ( credentials . checkPassword ( password ) ) <EOL> self . assertIn ( smtp . IMessageDeliveryFactory , interfaces ) <EOL> self . assertIn ( smtp . IMessageDelivery , interfaces ) <EOL> d . callback ( ( smtp . IMessageDeliveryFactory , None , lambda : None ) ) <EOL> self . assertEqual ( <EOL> [ "<STR_LIT>" ] , <EOL> self . transport . value ( ) . splitlines ( ) ) <EOL> def setUp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . server = smtp . ESMTP ( { <EOL> '<STR_LIT>' : imap4 . LOGINCredentials } ) <EOL> self . server . host = '<STR_LIT:localhost>' <EOL> self . transport = StringTransport ( <EOL> peerAddress = address . IPv4Address ( '<STR_LIT>' , '<STR_LIT:127.0.0.1>' , <NUM_LIT> ) ) <EOL> self . server . makeConnection ( self . transport ) <EOL> def tearDown ( self ) : <EOL> """<STR_LIT>""" <EOL> self . server . connectionLost ( error . ConnectionDone ( ) ) <EOL> def portalFactory ( self , loginList ) : <EOL> class DummyPortal : <EOL> def login ( self , credentials , mind , * interfaces ) : <EOL> d = defer . Deferred ( ) <EOL> loginList . append ( ( d , credentials , mind , interfaces ) ) <EOL> return d <EOL> return DummyPortal ( ) <EOL> def test_authenticationCapabilityAdvertised ( self ) : <EOL> """<STR_LIT>""" <EOL> self . transport . clear ( ) <EOL> self . server . dataReceived ( '<STR_LIT>' ) <EOL> responseLines = self . transport . value ( ) . splitlines ( ) <EOL> self . assertEqual ( <EOL> responseLines [ <NUM_LIT:0> ] , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( <EOL> responseLines [ <NUM_LIT:1> ] , <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( len ( responseLines ) , <NUM_LIT:2> ) <EOL> def test_plainAuthentication ( self ) : <EOL> """<STR_LIT>""" <EOL> loginArgs = [ ] <EOL> self . server . portal = self . portalFactory ( loginArgs ) <EOL> self . server . dataReceived ( '<STR_LIT>' ) <EOL> self . transport . clear ( ) <EOL> self . assertServerResponse ( <EOL> '<STR_LIT>' , <EOL> [ "<STR_LIT>" + "<STR_LIT>" . encode ( '<STR_LIT>' ) . strip ( ) ] ) <EOL> self . assertServerResponse ( <EOL> '<STR_LIT:username>' . encode ( '<STR_LIT>' ) + '<STR_LIT:\r\n>' , <EOL> [ "<STR_LIT>" + "<STR_LIT>" . encode ( '<STR_LIT>' ) . strip ( ) ] ) <EOL> self . assertServerResponse ( <EOL> '<STR_LIT:password>' . encode ( '<STR_LIT>' ) . strip ( ) + '<STR_LIT:\r\n>' , <EOL> [ ] ) <EOL> self . assertServerAuthenticated ( loginArgs ) <EOL> def test_plainAuthenticationEmptyPassword ( self ) : <EOL> """<STR_LIT>""" <EOL> loginArgs = [ ] <EOL> self . server . portal = self . portalFactory ( loginArgs ) <EOL> self . server . dataReceived ( '<STR_LIT>' ) <EOL> self . transport . clear ( ) <EOL> self . assertServerResponse ( <EOL> '<STR_LIT>' , <EOL> [ "<STR_LIT>" + "<STR_LIT>" . encode ( '<STR_LIT>' ) . strip ( ) ] ) <EOL> self . assertServerResponse ( <EOL> '<STR_LIT:username>' . encode ( '<STR_LIT>' ) + '<STR_LIT:\r\n>' , <EOL> [ "<STR_LIT>" + "<STR_LIT>" . encode ( '<STR_LIT>' ) . strip ( ) ] ) <EOL> self . assertServerResponse ( '<STR_LIT:\r\n>' , [ ] ) <EOL> self . assertServerAuthenticated ( loginArgs , password = '<STR_LIT>' ) <EOL> def test_plainAuthenticationInitialResponse ( self ) : <EOL> """<STR_LIT>""" <EOL> loginArgs = [ ] <EOL> self . server . portal = self . portalFactory ( loginArgs ) <EOL> self . server . dataReceived ( '<STR_LIT>' ) <EOL> self . transport . clear ( ) <EOL> self . assertServerResponse ( <EOL> '<STR_LIT>' + "<STR_LIT:username>" . encode ( '<STR_LIT>' ) . strip ( ) + '<STR_LIT:\r\n>' , <EOL> [ "<STR_LIT>" + "<STR_LIT>" . encode ( '<STR_LIT>' ) . strip ( ) ] ) <EOL> self . assertServerResponse ( <EOL> '<STR_LIT:password>' . encode ( '<STR_LIT>' ) . strip ( ) + '<STR_LIT:\r\n>' , <EOL> [ ] ) <EOL> self . assertServerAuthenticated ( loginArgs ) <EOL> def test_abortAuthentication ( self ) : <EOL> """<STR_LIT>""" <EOL> loginArgs = [ ] <EOL> self . server . portal = self . portalFactory ( loginArgs ) <EOL> self . server . dataReceived ( '<STR_LIT>' ) <EOL> self . server . dataReceived ( '<STR_LIT>' ) <EOL> self . assertServerResponse ( <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] ) <EOL> def test_invalidBase64EncodedResponse ( self ) : <EOL> """<STR_LIT>""" <EOL> loginArgs = [ ] <EOL> self . server . portal = self . portalFactory ( loginArgs ) <EOL> self . server . dataReceived ( '<STR_LIT>' ) <EOL> self . server . dataReceived ( '<STR_LIT>' ) <EOL> self . assertServerResponse ( <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( loginArgs , [ ] ) <EOL> def test_invalidBase64EncodedInitialResponse ( self ) : <EOL> """<STR_LIT>""" <EOL> loginArgs = [ ] <EOL> self . server . portal = self . portalFactory ( loginArgs ) <EOL> self . server . dataReceived ( '<STR_LIT>' ) <EOL> self . assertServerResponse ( <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( loginArgs , [ ] ) <EOL> def test_unexpectedLoginFailure ( self ) : <EOL> """<STR_LIT>""" <EOL> loginArgs = [ ] <EOL> self . server . portal = self . portalFactory ( loginArgs ) <EOL> self . server . dataReceived ( '<STR_LIT>' ) <EOL> self . transport . clear ( ) <EOL> self . assertServerResponse ( <EOL> '<STR_LIT>' + '<STR_LIT:username>' . encode ( '<STR_LIT>' ) . strip ( ) + '<STR_LIT:\r\n>' , <EOL> [ '<STR_LIT>' + '<STR_LIT>' . encode ( '<STR_LIT>' ) . strip ( ) ] ) <EOL> self . assertServerResponse ( <EOL> '<STR_LIT:password>' . encode ( '<STR_LIT>' ) . strip ( ) + '<STR_LIT:\r\n>' , <EOL> [ ] ) <EOL> d , credentials , mind , interfaces = loginArgs . pop ( ) <EOL> d . errback ( RuntimeError ( "<STR_LIT>" ) ) <EOL> self . assertEqual ( <EOL> '<STR_LIT>' , <EOL> self . transport . value ( ) ) <EOL> self . assertEqual ( len ( self . flushLoggedErrors ( RuntimeError ) ) , <NUM_LIT:1> ) <EOL> class SMTPClientErrorTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_str ( self ) : <EOL> """<STR_LIT>""" <EOL> err = smtp . SMTPClientError ( <NUM_LIT> , "<STR_LIT>" ) <EOL> self . assertEqual ( str ( err ) , "<STR_LIT>" ) <EOL> def test_strWithNegativeCode ( self ) : <EOL> """<STR_LIT>""" <EOL> err = smtp . SMTPClientError ( - <NUM_LIT:1> , "<STR_LIT>" ) <EOL> self . assertEqual ( str ( err ) , "<STR_LIT>" ) <EOL> def test_strWithLog ( self ) : <EOL> """<STR_LIT>""" <EOL> log = LineLog ( <NUM_LIT:10> ) <EOL> log . append ( "<STR_LIT>" ) <EOL> log . append ( "<STR_LIT>" ) <EOL> err = smtp . SMTPClientError ( <NUM_LIT:100> , "<STR_LIT>" , log = log . str ( ) ) <EOL> self . assertEqual ( <EOL> str ( err ) , <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> class SenderMixinSentMailTests ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> def test_onlyLogFailedAddresses ( self ) : <EOL> """<STR_LIT>""" <EOL> onDone = self . assertFailure ( defer . Deferred ( ) , smtp . SMTPDeliveryError ) <EOL> onDone . addCallback ( lambda e : self . assertEqual ( <EOL> e . log , "<STR_LIT>" ) ) <EOL> clientFactory = smtp . SMTPSenderFactory ( <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> StringIO ( "<STR_LIT>" ) , onDone , <EOL> retries = <NUM_LIT:0> , timeout = <NUM_LIT:0.5> ) <EOL> client = clientFactory . buildProtocol ( <EOL> address . IPv4Address ( '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT> ) ) <EOL> addresses = [ ( "<STR_LIT>" , <NUM_LIT:200> , "<STR_LIT>" ) , <EOL> ( "<STR_LIT>" , <NUM_LIT> , "<STR_LIT>" ) ] <EOL> client . sentMail ( <NUM_LIT> , "<STR_LIT>" , <NUM_LIT:1> , addresses , client . log ) <EOL> return onDone </s>
