1
# Copyright (c) 2001-2004 Twisted Matrix Laboratories.
2
# See LICENSE for details.
6
Test cases for twisted.mail.smtp module.
10
from zope.interface import implements
12
from twisted.trial import unittest, util
13
from twisted import protocols
14
from twisted import internet
15
from twisted.protocols import loopback
16
from twisted.mail import smtp
17
from twisted.internet import defer, protocol, reactor, interfaces, address, error
18
from twisted.test.test_protocols import StringIOWithoutClosing
19
from twisted.test.proto_helpers import StringTransport
21
from twisted import cred
22
import twisted.cred.error
23
import twisted.cred.portal
24
import twisted.cred.checkers
25
import twisted.cred.credentials
27
from twisted.cred.portal import IRealm, Portal
28
from twisted.cred.checkers import ICredentialsChecker, AllowAnonymousAccess
29
from twisted.cred.credentials import IAnonymous
30
from twisted.cred.error import UnauthorizedLogin
32
from twisted.mail import imap4
36
from twisted.test.ssl_helpers import ClientTLSContext, ServerTLSContext
38
ClientTLSContext = ServerTLSContext = None
43
from cStringIO import StringIO
45
from StringIO import StringIO
47
def spameater(*spam, **eggs):
52
def __init__(self, domain, user):
57
def lineReceived(self, line):
58
# Throw away the generated Received: header
59
if not re.match('Received: From yyy.com \(\[.*\]\) by localhost;', line):
60
self.buffer.append(line)
62
def eomReceived(self):
63
message = '\n'.join(self.buffer) + '\n'
64
self.domain.messages[self.user.dest.local].append(message)
65
deferred = defer.Deferred()
66
deferred.callback("saved")
72
def __init__(self, names):
75
self.messages[name] = []
77
def exists(self, user):
78
if self.messages.has_key(user.dest.local):
79
return defer.succeed(lambda: self.startMessage(user))
80
return defer.fail(smtp.SMTPBadRcpt(user))
82
def startMessage(self, user):
83
return DummyMessage(self, user)
85
class SMTPTestCase(unittest.TestCase):
87
messages = [('foo@bar.com', ['foo@baz.com', 'qux@baz.com'], '''\
90
Someone set up us the bomb!\015
93
mbox = {'foo': ['Subject: urgent\n\nSomeone set up us the bomb!\n']}
96
self.factory = smtp.SMTPFactory()
97
self.factory.domains = {}
98
self.factory.domains['baz.com'] = DummyDomain(['foo'])
99
self.output = StringIOWithoutClosing()
100
self.transport = internet.protocol.FileWrapper(self.output)
102
def testMessages(self):
103
from twisted.mail import protocols
104
protocol = protocols.DomainSMTP()
105
protocol.service = self.factory
106
protocol.factory = self.factory
107
protocol.receivedHeader = spameater
108
protocol.makeConnection(self.transport)
109
protocol.lineReceived('HELO yyy.com')
110
for message in self.messages:
111
protocol.lineReceived('MAIL FROM:<%s>' % message[0])
112
for target in message[1]:
113
protocol.lineReceived('RCPT TO:<%s>' % target)
114
protocol.lineReceived('DATA')
115
protocol.dataReceived(message[2])
116
protocol.lineReceived('.')
117
protocol.lineReceived('QUIT')
118
if self.mbox != self.factory.domains['baz.com'].messages:
119
raise AssertionError(self.factory.domains['baz.com'].messages)
120
protocol.setTimeout(None)
122
testMessages.suppress = [util.suppress(message='DomainSMTP', category=DeprecationWarning)]
132
self.mail = 'moshez@foo.bar', ['moshez@foo.bar'], mail
134
def getMailFrom(self):
140
def getMailData(self):
141
return StringIO(self.mail[2])
143
def sentMail(self, code, resp, numOk, addresses, log):
144
self.mail = None, None, None
146
class MySMTPClient(MyClient, smtp.SMTPClient):
148
smtp.SMTPClient.__init__(self, 'foo.baz')
149
MyClient.__init__(self)
151
class MyESMTPClient(MyClient, smtp.ESMTPClient):
152
def __init__(self, secret = '', contextFactory = None):
153
smtp.ESMTPClient.__init__(self, secret, contextFactory, 'foo.baz')
154
MyClient.__init__(self)
157
def loopback(self, server, client):
158
return loopback.loopbackTCP(server, client)
160
class LoopbackTestCase(LoopbackMixin):
161
def testMessages(self):
162
factory = smtp.SMTPFactory()
164
factory.domains['foo.bar'] = DummyDomain(['moshez'])
165
from twisted.mail.protocols import DomainSMTP
166
protocol = DomainSMTP()
167
protocol.service = factory
168
protocol.factory = factory
169
clientProtocol = self.clientClass()
170
return self.loopback(protocol, clientProtocol)
171
testMessages.suppress = [util.suppress(message='DomainSMTP', category=DeprecationWarning)]
173
class LoopbackSMTPTestCase(LoopbackTestCase, unittest.TestCase):
174
clientClass = MySMTPClient
176
class LoopbackESMTPTestCase(LoopbackTestCase, unittest.TestCase):
177
clientClass = MyESMTPClient
180
class FakeSMTPServer(protocols.basic.LineReceiver):
183
'220 hello', '250 nice to meet you',
184
'250 great', '250 great', '354 go on, lad'
187
def connectionMade(self):
189
self.clientData = self.clientData[:]
190
self.clientData.reverse()
191
self.sendLine(self.clientData.pop())
193
def lineReceived(self, line):
194
self.buffer.append(line)
196
self.transport.write("221 see ya around\r\n")
197
self.transport.loseConnection()
199
self.transport.write("250 gotcha\r\n")
201
self.transport.loseConnection()
204
self.sendLine(self.clientData.pop())
207
class SMTPClientTestCase(unittest.TestCase, LoopbackMixin):
210
'HELO foo.baz', 'MAIL FROM:<moshez@foo.bar>',
211
'RCPT TO:<moshez@foo.bar>', 'DATA',
212
'Subject: hello', '', 'Goodbye', '.', 'RSET'
215
def testMessages(self):
216
# this test is disabled temporarily
217
client = MySMTPClient()
218
server = FakeSMTPServer()
219
d = self.loopback(server, client)
220
d.addCallback(lambda x :
221
self.assertEquals(server.buffer, self.expected_output))
224
class DummySMTPMessage:
226
def __init__(self, protocol, users):
227
self.protocol = protocol
231
def lineReceived(self, line):
232
self.buffer.append(line)
234
def eomReceived(self):
235
message = '\n'.join(self.buffer) + '\n'
236
helo, origin = self.users[0].helo[0], str(self.users[0].orig)
238
for user in self.users:
239
recipients.append(str(user))
240
self.protocol.message[tuple(recipients)] = (helo, origin, recipients, message)
241
return defer.succeed("saved")
244
def connectionMade(self):
245
self.dummyMixinBase.connectionMade(self)
248
def startMessage(self, users):
249
return DummySMTPMessage(self, users)
251
def receivedHeader(*spam):
254
def validateTo(self, user):
255
self.delivery = DummyDelivery()
256
return lambda: self.startMessage([user])
258
def validateFrom(self, helo, origin):
261
class DummySMTP(DummyProto, smtp.SMTP):
262
dummyMixinBase = smtp.SMTP
264
class DummyESMTP(DummyProto, smtp.ESMTP):
265
dummyMixinBase = smtp.ESMTP
267
class AnotherTestCase:
271
messages = [ ('foo.com', 'moshez@foo.com', ['moshez@bar.com'],
272
'moshez@foo.com', ['moshez@bar.com'], '''\
279
('foo.com', 'tttt@rrr.com', ['uuu@ooo', 'yyy@eee'],
280
'tttt@rrr.com', ['uuu@ooo', 'yyy@eee'], '''\
285
('foo.com', '@this,@is,@ignored:foo@bar.com',
286
['@ignore,@this,@too:bar@foo.com'],
287
'foo@bar.com', ['bar@foo.com'], '''\
298
('', '220.*\r\n$', None, None),
299
('HELO foo.com\r\n', '250.*\r\n$', None, None),
300
('RSET\r\n', '250.*\r\n$', None, None),
302
for helo_, from_, to_, realfrom, realto, msg in messages:
303
data.append(('MAIL FROM:<%s>\r\n' % from_, '250.*\r\n',
306
data.append(('RCPT TO:<%s>\r\n' % rcpt, '250.*\r\n',
309
data.append(('DATA\r\n','354.*\r\n',
311
(helo_, realfrom, realto, msg))))
314
def testBuffer(self):
315
output = StringIOWithoutClosing()
316
a = self.serverClass()
320
a.factory = fooFactory()
321
a.makeConnection(protocol.FileWrapper(output))
322
for (send, expect, msg, msgexpect) in self.data:
325
data = output.getvalue()
327
if not re.match(expect, data):
328
raise AssertionError, (send, expect, data)
329
if data[:3] == '354':
330
for line in msg.splitlines():
331
if line and line[0] == '.':
333
a.dataReceived(line + '\r\n')
334
a.dataReceived('.\r\n')
335
# Special case for DATA. Now we want a 250, and then
336
# we compare the messages
337
data = output.getvalue()
339
resp, msgdata = msgexpect
340
if not re.match(resp, data):
341
raise AssertionError, (resp, data)
342
for recip in msgdata[2]:
343
expected = list(msgdata[:])
344
expected[2] = [recip]
352
class AnotherESMTPTestCase(AnotherTestCase, unittest.TestCase):
353
serverClass = DummyESMTP
354
clientClass = MyESMTPClient
356
class AnotherSMTPTestCase(AnotherTestCase, unittest.TestCase):
357
serverClass = DummySMTP
358
clientClass = MySMTPClient
363
implements(cred.checkers.ICredentialsChecker)
366
'testuser': 'testpassword'
369
credentialInterfaces = (cred.credentials.IUsernameHashedPassword,)
371
def requestAvatarId(self, credentials):
372
return defer.maybeDeferred(
373
credentials.checkPassword, self.users[credentials.username]
374
).addCallback(self._cbCheck, credentials.username)
376
def _cbCheck(self, result, username):
379
raise cred.error.UnauthorizedLogin()
382
implements(smtp.IMessageDelivery)
384
def validateTo(self, user):
387
def validateFrom(self, helo, origin):
390
def receivedHeader(*args):
394
def requestAvatar(self, avatarId, mind, *interfaces):
395
return smtp.IMessageDelivery, DummyDelivery(), lambda: None
397
class AuthTestCase(unittest.TestCase, LoopbackMixin):
400
p = cred.portal.Portal(realm)
401
p.registerChecker(DummyChecker())
403
server = DummyESMTP({'CRAM-MD5': cred.credentials.CramMD5Credentials})
405
client = MyESMTPClient('testpassword')
407
cAuth = imap4.CramMD5ClientAuthenticator('testuser')
408
client.registerAuthenticator(cAuth)
410
d = self.loopback(server, client)
411
d.addCallback(lambda x : self.assertEquals(server.authenticated, 1))
414
class SMTPHelperTestCase(unittest.TestCase):
415
def testMessageID(self):
417
for i in range(1000):
418
m = smtp.messageid('testcase')
422
def testQuoteAddr(self):
424
['user@host.name', '<user@host.name>'],
425
['"User Name" <user@host.name>', '<user@host.name>'],
426
[smtp.Address('someguy@someplace'), '<someguy@someplace>'],
428
[smtp.Address(''), '<>'],
432
self.assertEquals(smtp.quoteaddr(c), e)
435
u = smtp.User('user@host', 'helo.host.name', None, None)
436
self.assertEquals(str(u), 'user@host')
438
def testXtextEncoding(self):
440
('Hello world', 'Hello+20world'),
441
('Hello+world', 'Hello+2Bworld'),
442
('\0\1\2\3\4\5', '+00+01+02+03+04+05'),
443
('e=mc2@example.com', 'e+3Dmc2@example.com')
446
for (case, expected) in cases:
447
self.assertEquals(case.encode('xtext'), expected)
448
self.assertEquals(expected.decode('xtext'), case)
451
class NoticeTLSClient(MyESMTPClient):
454
def esmtpState_starttls(self, code, resp):
455
MyESMTPClient.esmtpState_starttls(self, code, resp)
458
class TLSTestCase(unittest.TestCase, LoopbackMixin):
460
clientCTX = ClientTLSContext()
461
serverCTX = ServerTLSContext()
463
client = NoticeTLSClient(contextFactory=clientCTX)
464
server = DummyESMTP(contextFactory=serverCTX)
467
self.assertEquals(client.tls, True)
468
self.assertEquals(server.startedTLS, True)
470
return self.loopback(server, client).addCallback(check)
472
if ClientTLSContext is None:
473
for case in (TLSTestCase,):
474
case.skip = "OpenSSL not present"
476
if not interfaces.IReactorSSL.providedBy(reactor):
477
for case in (TLSTestCase,):
478
case.skip = "Reactor doesn't support SSL"
480
class EmptyLineTestCase(unittest.TestCase):
481
def testEmptyLineSyntaxError(self):
483
output = StringIOWithoutClosing()
484
transport = internet.protocol.FileWrapper(output)
485
proto.makeConnection(transport)
486
proto.lineReceived('')
487
proto.setTimeout(None)
489
out = output.getvalue().splitlines()
490
self.assertEquals(len(out), 2)
491
self.failUnless(out[0].startswith('220'))
492
self.assertEquals(out[1], "500 Error: bad syntax")
494
class TimeoutTestCase(unittest.TestCase, LoopbackMixin):
495
def _timeoutTest(self, onDone, clientFactory):
498
client = clientFactory.buildProtocol(
499
address.IPv4Address('TCP', 'example.net', 25))
500
server = protocol.Protocol()
504
self.failIf(after - before > 1.0)
505
return self.assertFailure(onDone, smtp.SMTPTimeoutError)
507
return self.loopback(client, server).addCallback(check)
510
def testSMTPClient(self):
511
onDone = defer.Deferred()
512
clientFactory = smtp.SMTPSenderFactory(
513
'source@address', 'recipient@address',
514
StringIO("Message body"), onDone,
515
retries=0, timeout=0.5)
516
return self._timeoutTest(onDone, clientFactory)
519
def testESMTPClient(self):
520
onDone = defer.Deferred()
521
clientFactory = smtp.ESMTPSenderFactory(
522
'username', 'password',
523
'source@address', 'recipient@address',
524
StringIO("Message body"), onDone,
525
retries=0, timeout=0.5)
526
return self._timeoutTest(onDone, clientFactory)
530
class SingletonRealm(object):
532
Trivial realm implementation which is constructed with an interface and an
533
avatar and returns that avatar when asked for that interface.
537
def __init__(self, interface, avatar):
538
self.interface = interface
542
def requestAvatar(self, avatarId, mind, *interfaces):
543
for iface in interfaces:
544
if iface is self.interface:
545
return iface, self.avatar, lambda: None
549
class NotImplementedDelivery(object):
551
Non-implementation of L{smtp.IMessageDelivery} which only has methods which
552
raise L{NotImplementedError}. Subclassed by various tests to provide the
553
particular behavior being tested.
555
def validateFrom(self, helo, origin):
556
raise NotImplementedError("This oughtn't be called in the course of this test.")
559
def validateTo(self, user):
560
raise NotImplementedError("This oughtn't be called in the course of this test.")
563
def receivedHeader(self, helo, origin, recipients):
564
raise NotImplementedError("This oughtn't be called in the course of this test.")
568
class SMTPServerTestCase(unittest.TestCase):
570
Test various behaviors of L{twisted.mail.smtp.SMTP} and
571
L{twisted.mail.smtp.ESMTP}.
573
def testSMTPGreetingHost(self, serverClass=smtp.SMTP):
575
Test that the specified hostname shows up in the SMTP server's
579
s.host = "example.com"
580
t = StringTransport()
582
s.connectionLost(error.ConnectionDone())
583
self.assertIn("example.com", t.value())
586
def testSMTPGreetingNotExtended(self):
588
Test that the string "ESMTP" does not appear in the SMTP server's
589
greeting since that string strongly suggests the presence of support
590
for various SMTP extensions which are not supported by L{smtp.SMTP}.
593
t = StringTransport()
595
s.connectionLost(error.ConnectionDone())
596
self.assertNotIn("ESMTP", t.value())
599
def testESMTPGreetingHost(self):
601
Similar to testSMTPGreetingHost, but for the L{smtp.ESMTP} class.
603
self.testSMTPGreetingHost(smtp.ESMTP)
606
def testESMTPGreetingExtended(self):
608
Test that the string "ESMTP" does appear in the ESMTP server's
609
greeting since L{smtp.ESMTP} does support the SMTP extensions which
610
that advertises to the client.
613
t = StringTransport()
615
s.connectionLost(error.ConnectionDone())
616
self.assertIn("ESMTP", t.value())
619
def test_acceptSenderAddress(self):
621
Test that a C{MAIL FROM} command with an acceptable address is
622
responded to with the correct success code.
624
class AcceptanceDelivery(NotImplementedDelivery):
626
Delivery object which accepts all senders as valid.
628
def validateFrom(self, helo, origin):
631
realm = SingletonRealm(smtp.IMessageDelivery, AcceptanceDelivery())
632
portal = Portal(realm, [AllowAnonymousAccess()])
634
proto.portal = portal
635
trans = StringTransport()
636
proto.makeConnection(trans)
638
# Deal with the necessary preliminaries
639
proto.dataReceived('HELO example.com\r\n')
642
# Try to specify our sender address
643
proto.dataReceived('MAIL FROM:<alice@example.com>\r\n')
645
# Clean up the protocol before doing anything that might raise an
647
proto.connectionLost(error.ConnectionLost())
649
# Make sure that we received exactly the correct response
652
'250 Sender address accepted\r\n')
655
def test_deliveryRejectedSenderAddress(self):
657
Test that a C{MAIL FROM} command with an address rejected by a
658
L{smtp.IMessageDelivery} instance is responded to with the correct
661
class RejectionDelivery(NotImplementedDelivery):
663
Delivery object which rejects all senders as invalid.
665
def validateFrom(self, helo, origin):
666
raise smtp.SMTPBadSender(origin)
668
realm = SingletonRealm(smtp.IMessageDelivery, RejectionDelivery())
669
portal = Portal(realm, [AllowAnonymousAccess()])
671
proto.portal = portal
672
trans = StringTransport()
673
proto.makeConnection(trans)
675
# Deal with the necessary preliminaries
676
proto.dataReceived('HELO example.com\r\n')
679
# Try to specify our sender address
680
proto.dataReceived('MAIL FROM:<alice@example.com>\r\n')
682
# Clean up the protocol before doing anything that might raise an
684
proto.connectionLost(error.ConnectionLost())
686
# Make sure that we received exactly the correct response
689
'550 Cannot receive from specified address '
690
'<alice@example.com>: Sender not acceptable\r\n')
693
def test_portalRejectedSenderAddress(self):
695
Test that a C{MAIL FROM} command with an address rejected by an
696
L{smtp.SMTP} instance's portal is responded to with the correct error
699
class DisallowAnonymousAccess(object):
701
Checker for L{IAnonymous} which rejects authentication attempts.
703
implements(ICredentialsChecker)
705
credentialInterfaces = (IAnonymous,)
707
def requestAvatarId(self, credentials):
708
return defer.fail(UnauthorizedLogin())
710
realm = SingletonRealm(smtp.IMessageDelivery, NotImplementedDelivery())
711
portal = Portal(realm, [DisallowAnonymousAccess()])
713
proto.portal = portal
714
trans = StringTransport()
715
proto.makeConnection(trans)
717
# Deal with the necessary preliminaries
718
proto.dataReceived('HELO example.com\r\n')
721
# Try to specify our sender address
722
proto.dataReceived('MAIL FROM:<alice@example.com>\r\n')
724
# Clean up the protocol before doing anything that might raise an
726
proto.connectionLost(error.ConnectionLost())
728
# Make sure that we received exactly the correct response
731
'550 Cannot receive from specified address '
732
'<alice@example.com>: Sender not acceptable\r\n')
735
def test_portalRejectedAnonymousSender(self):
737
Test that a C{MAIL FROM} command issued without first authenticating
738
when a portal has been configured to disallow anonymous logins is
739
responded to with the correct error code.
741
realm = SingletonRealm(smtp.IMessageDelivery, NotImplementedDelivery())
742
portal = Portal(realm, [])
744
proto.portal = portal
745
trans = StringTransport()
746
proto.makeConnection(trans)
748
# Deal with the necessary preliminaries
749
proto.dataReceived('HELO example.com\r\n')
752
# Try to specify our sender address
753
proto.dataReceived('MAIL FROM:<alice@example.com>\r\n')
755
# Clean up the protocol before doing anything that might raise an
757
proto.connectionLost(error.ConnectionLost())
759
# Make sure that we received exactly the correct response
762
'550 Cannot receive from specified address '
763
'<alice@example.com>: Unauthenticated senders not allowed\r\n')
767
class ESMTPAuthenticationTestCase(unittest.TestCase):
768
def assertServerResponse(self, bytes, response):
770
Assert that when the given bytes are delivered to the ESMTP server
771
instance, it responds with the indicated lines.
774
@type response: list of str
776
self.transport.clear()
777
self.server.dataReceived(bytes)
780
self.transport.value().splitlines())
783
def assertServerAuthenticated(self, loginArgs):
785
Assert that a login attempt has been made, that the credentials and
786
interfaces passed to it are correct, and that when the login request
787
is satisfied, a successful response is sent by the ESMTP server
790
@param loginArgs: A C{list} previously passed to L{portalFactory}.
792
d, credentials, mind, interfaces = loginArgs.pop()
793
self.assertEqual(loginArgs, [])
794
self.failUnless(twisted.cred.credentials.IUsernamePassword.providedBy(credentials))
795
self.assertEqual(credentials.username, 'username')
796
self.failUnless(credentials.checkPassword('password'))
797
self.assertIn(smtp.IMessageDeliveryFactory, interfaces)
798
self.assertIn(smtp.IMessageDelivery, interfaces)
799
d.callback((smtp.IMessageDeliveryFactory, None, lambda: None))
802
["235 Authentication successful."],
803
self.transport.value().splitlines())
808
Create an ESMTP instance attached to a StringTransport.
810
self.server = smtp.ESMTP({
811
'LOGIN': imap4.LOGINCredentials})
812
self.server.host = 'localhost'
813
self.transport = StringTransport(
814
peerAddress=address.IPv4Address('TCP', '127.0.0.1', 12345))
815
self.server.makeConnection(self.transport)
820
Disconnect the ESMTP instance to clean up its timeout DelayedCall.
822
self.server.connectionLost(error.ConnectionDone())
825
def portalFactory(self, loginList):
827
def login(self, credentials, mind, *interfaces):
829
loginList.append((d, credentials, mind, interfaces))
834
def test_authenticationCapabilityAdvertised(self):
836
Test that AUTH is advertised to clients which issue an EHLO command.
838
self.transport.clear()
839
self.server.dataReceived('EHLO\r\n')
840
responseLines = self.transport.value().splitlines()
843
"250-localhost Hello 127.0.0.1, nice to meet you")
847
self.assertEqual(len(responseLines), 2)
850
def test_plainAuthentication(self):
852
Test that the LOGIN authentication mechanism can be used
855
self.server.portal = self.portalFactory(loginArgs)
857
self.server.dataReceived('EHLO\r\n')
858
self.transport.clear()
860
self.assertServerResponse(
862
["334 " + "User Name\0".encode('base64').strip()])
864
self.assertServerResponse(
865
'username'.encode('base64') + '\r\n',
866
["334 " + "Password\0".encode('base64').strip()])
868
self.assertServerResponse(
869
'password'.encode('base64').strip() + '\r\n',
872
self.assertServerAuthenticated(loginArgs)
875
def test_plainAuthenticationInitialResponse(self):
877
The response to the first challenge may be included on the AUTH command
878
line. Test that this is also supported.
881
self.server.portal = self.portalFactory(loginArgs)
883
self.server.dataReceived('EHLO\r\n')
884
self.transport.clear()
886
self.assertServerResponse(
887
'AUTH LOGIN ' + "username".encode('base64').strip() + '\r\n',
888
["334 " + "Password\0".encode('base64').strip()])
890
self.assertServerResponse(
891
'password'.encode('base64').strip() + '\r\n',
894
self.assertServerAuthenticated(loginArgs)
897
def test_abortAuthentication(self):
899
Test that a challenge/response sequence can be aborted by the client.
902
self.server.portal = self.portalFactory(loginArgs)
904
self.server.dataReceived('EHLO\r\n')
905
self.server.dataReceived('AUTH LOGIN\r\n')
907
self.assertServerResponse(
909
['501 Authentication aborted'])
912
def test_invalidBase64EncodedResponse(self):
914
Test that a response which is not properly Base64 encoded results in
915
the appropriate error code.
918
self.server.portal = self.portalFactory(loginArgs)
920
self.server.dataReceived('EHLO\r\n')
921
self.server.dataReceived('AUTH LOGIN\r\n')
923
self.assertServerResponse(
925
['501 Syntax error in parameters or arguments'])
927
self.assertEqual(loginArgs, [])
930
def test_invalidBase64EncodedInitialResponse(self):
932
Like L{test_invalidBase64EncodedResponse} but for the case of an
933
initial response included with the C{AUTH} command.
936
self.server.portal = self.portalFactory(loginArgs)
938
self.server.dataReceived('EHLO\r\n')
939
self.assertServerResponse(
941
['501 Syntax error in parameters or arguments'])
943
self.assertEqual(loginArgs, [])