1
# -*- test-case-name: twisted.mail.test.test_imap -*-
2
# Copyright (c) 2001-2010 Twisted Matrix Laboratories.
3
# See LICENSE for details.
7
Test case for twisted.mail.imap4
11
from cStringIO import StringIO
13
from StringIO import StringIO
19
from zope.interface import implements
21
from twisted.mail.imap4 import MessageSet
22
from twisted.mail import imap4
23
from twisted.protocols import loopback
24
from twisted.internet import defer
25
from twisted.internet import error
26
from twisted.internet import reactor
27
from twisted.internet import interfaces
28
from twisted.internet.task import Clock
29
from twisted.trial import unittest
30
from twisted.python import util
31
from twisted.python import failure
33
from twisted import cred
34
import twisted.cred.error
35
import twisted.cred.checkers
36
import twisted.cred.credentials
37
import twisted.cred.portal
39
from twisted.test.proto_helpers import StringTransport, StringTransportWithDisconnection
42
from twisted.test.ssl_helpers import ClientTLSContext, ServerTLSContext
44
ClientTLSContext = ServerTLSContext = None
47
return lambda result, f=f: f()
52
for i in range(len(l)):
53
if isinstance(l[i], types.ListType):
55
elif isinstance(l[i], types.TupleType):
56
l[i] = tuple(sortNest(list(l[i])))
59
class IMAP4UTF7TestCase(unittest.TestCase):
61
[u'Hello world', 'Hello world'],
62
[u'Hello & world', 'Hello &- world'],
63
[u'Hello\xffworld', 'Hello&AP8-world'],
64
[u'\xff\xfe\xfd\xfc', '&AP8A,gD9APw-'],
65
[u'~peter/mail/\u65e5\u672c\u8a9e/\u53f0\u5317',
66
'~peter/mail/&ZeVnLIqe-/&U,BTFw-'], # example from RFC 2060
69
def test_encodeWithErrors(self):
71
Specifying an error policy to C{unicode.encode} with the
72
I{imap4-utf-7} codec should produce the same result as not
73
specifying the error policy.
77
text.encode('imap4-utf-7', 'strict'),
78
text.encode('imap4-utf-7'))
81
def test_decodeWithErrors(self):
83
Similar to L{test_encodeWithErrors}, but for C{str.decode}.
87
bytes.decode('imap4-utf-7', 'strict'),
88
bytes.decode('imap4-utf-7'))
91
def test_getreader(self):
93
C{codecs.getreader('imap4-utf-7')} returns the I{imap4-utf-7} stream
96
reader = codecs.getreader('imap4-utf-7')(StringIO('Hello&AP8-world'))
97
self.assertEquals(reader.read(), u'Hello\xffworld')
100
def test_getwriter(self):
102
C{codecs.getwriter('imap4-utf-7')} returns the I{imap4-utf-7} stream
106
writer = codecs.getwriter('imap4-utf-7')(output)
107
writer.write(u'Hello\xffworld')
108
self.assertEquals(output.getvalue(), 'Hello&AP8-world')
111
def test_encode(self):
113
The I{imap4-utf-7} can be used to encode a unicode string into a byte
114
string according to the IMAP4 modified UTF-7 encoding rules.
116
for (input, output) in self.tests:
117
self.assertEquals(input.encode('imap4-utf-7'), output)
120
def test_decode(self):
122
The I{imap4-utf-7} can be used to decode a byte string into a unicode
123
string according to the IMAP4 modified UTF-7 encoding rules.
125
for (input, output) in self.tests:
126
self.assertEquals(input, output.decode('imap4-utf-7'))
129
def test_printableSingletons(self):
131
The IMAP4 modified UTF-7 implementation encodes all printable
132
characters which are in ASCII using the corresponding ASCII byte.
134
# All printables represent themselves
135
for o in range(0x20, 0x26) + range(0x27, 0x7f):
136
self.failUnlessEqual(chr(o), chr(o).encode('imap4-utf-7'))
137
self.failUnlessEqual(chr(o), chr(o).decode('imap4-utf-7'))
138
self.failUnlessEqual('&'.encode('imap4-utf-7'), '&-')
139
self.failUnlessEqual('&-'.decode('imap4-utf-7'), '&')
143
class BufferingConsumer:
147
def write(self, bytes):
148
self.buffer.append(bytes)
150
self.consumer.resumeProducing()
152
def registerProducer(self, consumer, streaming):
153
self.consumer = consumer
154
self.consumer.resumeProducing()
156
def unregisterProducer(self):
159
class MessageProducerTestCase(unittest.TestCase):
160
def testSinglePart(self):
161
body = 'This is body text. Rar.'
162
headers = util.OrderedDict()
163
headers['from'] = 'sender@host'
164
headers['to'] = 'recipient@domain'
165
headers['subject'] = 'booga booga boo'
166
headers['content-type'] = 'text/plain'
168
msg = FakeyMessage(headers, (), None, body, 123, None )
170
c = BufferingConsumer()
171
p = imap4.MessageProducer(msg)
172
d = p.beginProducing(c)
174
def cbProduced(result):
175
self.assertIdentical(result, p)
180
'From: sender@host\r\n'
181
'To: recipient@domain\r\n'
182
'Subject: booga booga boo\r\n'
183
'Content-Type: text/plain\r\n'
186
return d.addCallback(cbProduced)
189
def testSingleMultiPart(self):
191
innerBody = 'Contained body message text. Squarge.'
192
headers = util.OrderedDict()
193
headers['from'] = 'sender@host'
194
headers['to'] = 'recipient@domain'
195
headers['subject'] = 'booga booga boo'
196
headers['content-type'] = 'multipart/alternative; boundary="xyz"'
198
innerHeaders = util.OrderedDict()
199
innerHeaders['subject'] = 'this is subject text'
200
innerHeaders['content-type'] = 'text/plain'
201
msg = FakeyMessage(headers, (), None, outerBody, 123,
202
[FakeyMessage(innerHeaders, (), None, innerBody,
206
c = BufferingConsumer()
207
p = imap4.MessageProducer(msg)
208
d = p.beginProducing(c)
210
def cbProduced(result):
211
self.failUnlessIdentical(result, p)
217
'From: sender@host\r\n'
218
'To: recipient@domain\r\n'
219
'Subject: booga booga boo\r\n'
220
'Content-Type: multipart/alternative; boundary="xyz"\r\n'
224
'Subject: this is subject text\r\n'
225
'Content-Type: text/plain\r\n'
230
return d.addCallback(cbProduced)
233
def testMultipleMultiPart(self):
235
innerBody1 = 'Contained body message text. Squarge.'
236
innerBody2 = 'Secondary <i>message</i> text of squarge body.'
237
headers = util.OrderedDict()
238
headers['from'] = 'sender@host'
239
headers['to'] = 'recipient@domain'
240
headers['subject'] = 'booga booga boo'
241
headers['content-type'] = 'multipart/alternative; boundary="xyz"'
242
innerHeaders = util.OrderedDict()
243
innerHeaders['subject'] = 'this is subject text'
244
innerHeaders['content-type'] = 'text/plain'
245
innerHeaders2 = util.OrderedDict()
246
innerHeaders2['subject'] = '<b>this is subject</b>'
247
innerHeaders2['content-type'] = 'text/html'
248
msg = FakeyMessage(headers, (), None, outerBody, 123, [
249
FakeyMessage(innerHeaders, (), None, innerBody1, None, None),
250
FakeyMessage(innerHeaders2, (), None, innerBody2, None, None)
254
c = BufferingConsumer()
255
p = imap4.MessageProducer(msg)
256
d = p.beginProducing(c)
258
def cbProduced(result):
259
self.failUnlessIdentical(result, p)
265
'From: sender@host\r\n'
266
'To: recipient@domain\r\n'
267
'Subject: booga booga boo\r\n'
268
'Content-Type: multipart/alternative; boundary="xyz"\r\n'
272
'Subject: this is subject text\r\n'
273
'Content-Type: text/plain\r\n'
277
'Subject: <b>this is subject</b>\r\n'
278
'Content-Type: text/html\r\n'
282
return d.addCallback(cbProduced)
286
class IMAP4HelperTestCase(unittest.TestCase):
287
def testFileProducer(self):
288
b = (('x' * 1) + ('y' * 1) + ('z' * 1)) * 10
289
c = BufferingConsumer()
291
p = imap4.FileProducer(f)
292
d = p.beginProducing(c)
294
def cbProduced(result):
295
self.failUnlessIdentical(result, p)
297
('{%d}\r\n' % len(b))+ b,
299
return d.addCallback(cbProduced)
301
def testWildcard(self):
304
['foo/bar', 'oo/lalagum/bar', 'foo/gumx/bar', 'foo/gum/baz'],
305
['foo/xgum/bar', 'foo/gum/bar'],
307
['foo', 'bar', 'fuz fuz fuz', 'foo/*/bar', 'foo/xyz/bar', 'foo/xx/baz'],
308
['foo/xyx/bar', 'foo/xx/bar', 'foo/xxxxxxxxxxxxxx/bar'],
309
], ['foo/xyz*abc/bar',
310
['foo/xyz/bar', 'foo/abc/bar', 'foo/xyzab/cbar', 'foo/xyza/bcbar'],
311
['foo/xyzabc/bar', 'foo/xyz/abc/bar', 'foo/xyz/123/abc/bar'],
315
for (wildcard, fail, succeed) in cases:
316
wildcard = imap4.wildcardToRegexp(wildcard, '/')
318
self.failIf(wildcard.match(x))
320
self.failUnless(wildcard.match(x))
322
def testWildcardNoDelim(self):
325
['foo/bar', 'oo/lalagum/bar', 'foo/gumx/bar', 'foo/gum/baz'],
326
['foo/xgum/bar', 'foo/gum/bar', 'foo/x/gum/bar'],
328
['foo', 'bar', 'fuz fuz fuz', 'foo/*/bar', 'foo/xyz/bar', 'foo/xx/baz'],
329
['foo/xyx/bar', 'foo/xx/bar', 'foo/xxxxxxxxxxxxxx/bar', 'foo/x/x/bar'],
330
], ['foo/xyz*abc/bar',
331
['foo/xyz/bar', 'foo/abc/bar', 'foo/xyzab/cbar', 'foo/xyza/bcbar'],
332
['foo/xyzabc/bar', 'foo/xyz/abc/bar', 'foo/xyz/123/abc/bar'],
336
for (wildcard, fail, succeed) in cases:
337
wildcard = imap4.wildcardToRegexp(wildcard, None)
339
self.failIf(wildcard.match(x), x)
341
self.failUnless(wildcard.match(x), x)
343
def testHeaderFormatter(self):
345
({'Header1': 'Value1', 'Header2': 'Value2'}, 'Header2: Value2\r\nHeader1: Value1\r\n'),
348
for (input, output) in cases:
349
self.assertEquals(imap4._formatHeaders(input), output)
351
def testMessageSet(self):
355
self.assertEquals(m1, m2)
358
self.assertEquals(len(m1), 3)
359
self.assertEquals(list(m1), [1, 2, 3])
362
self.assertEquals(m1, m2)
363
self.assertEquals(list(m1 + m2), [1, 2, 3])
365
def testQuotedSplitter(self):
368
'''Hello "World!"''',
369
'''World "Hello" "How are you?"''',
370
'''"Hello world" How "are you?"''',
371
'''foo bar "baz buz" NIL''',
372
'''foo bar "baz buz" "NIL"''',
373
'''foo NIL "baz buz" bar''',
374
'''foo "NIL" "baz buz" bar''',
375
'''"NIL" bar "baz buz" foo''',
391
['World', 'Hello', 'How are you?'],
392
['Hello world', 'How', 'are you?'],
393
['foo', 'bar', 'baz buz', None],
394
['foo', 'bar', 'baz buz', 'NIL'],
395
['foo', None, 'baz buz', 'bar'],
396
['foo', 'NIL', 'baz buz', 'bar'],
397
['NIL', 'bar', 'baz buz', 'foo'],
398
['oo', '"oo"', 'oo'],
415
'"oops here is" another"',
419
self.assertRaises(imap4.MismatchedQuoting, imap4.splitQuoted, s)
421
for (case, expected) in zip(cases, answers):
422
self.assertEquals(imap4.splitQuoted(case), expected)
425
def testStringCollapser(self):
427
['a', 'b', 'c', 'd', 'e'],
428
['a', ' ', '"', 'b', 'c', ' ', '"', ' ', 'd', 'e'],
429
[['a', 'b', 'c'], 'd', 'e'],
430
['a', ['b', 'c', 'd'], 'e'],
431
['a', 'b', ['c', 'd', 'e']],
432
['"', 'a', ' ', '"', ['b', 'c', 'd'], '"', ' ', 'e', '"'],
433
['a', ['"', ' ', 'b', 'c', ' ', ' ', '"'], 'd', 'e'],
442
['a ', ['bcd'], ' e'],
443
['a', [' bc '], 'de'],
446
for (case, expected) in zip(cases, answers):
447
self.assertEquals(imap4.collapseStrings(case), expected)
449
def testParenParser(self):
450
s = '\r\n'.join(['xx'] * 4)
452
'(BODY.PEEK[HEADER.FIELDS.NOT (subject bcc cc)] {%d}\r\n%s)' % (len(s), s,),
454
# '(FLAGS (\Seen) INTERNALDATE "17-Jul-1996 02:44:25 -0700" '
455
# 'RFC822.SIZE 4286 ENVELOPE ("Wed, 17 Jul 1996 02:23:25 -0700 (PDT)" '
456
# '"IMAP4rev1 WG mtg summary and minutes" '
457
# '(("Terry Gray" NIL "gray" "cac.washington.edu")) '
458
# '(("Terry Gray" NIL "gray" "cac.washington.edu")) '
459
# '(("Terry Gray" NIL "gray" "cac.washington.edu")) '
460
# '((NIL NIL "imap" "cac.washington.edu")) '
461
# '((NIL NIL "minutes" "CNRI.Reston.VA.US") '
462
# '("John Klensin" NIL "KLENSIN" "INFOODS.MIT.EDU")) NIL NIL '
463
# '"<B27397-0100000@cac.washington.edu>") '
464
# 'BODY ("TEXT" "PLAIN" ("CHARSET" "US-ASCII") NIL NIL "7BIT" 3028 92))',
466
'(FLAGS (\Seen) INTERNALDATE "17-Jul-1996 02:44:25 -0700" '
467
'RFC822.SIZE 4286 ENVELOPE ("Wed, 17 Jul 1996 02:23:25 -0700 (PDT)" '
468
'"IMAP4rev1 WG mtg summary and minutes" '
469
'(("Terry Gray" NIL gray cac.washington.edu)) '
470
'(("Terry Gray" NIL gray cac.washington.edu)) '
471
'(("Terry Gray" NIL gray cac.washington.edu)) '
472
'((NIL NIL imap cac.washington.edu)) '
473
'((NIL NIL minutes CNRI.Reston.VA.US) '
474
'("John Klensin" NIL KLENSIN INFOODS.MIT.EDU)) NIL NIL '
475
'<B27397-0100000@cac.washington.edu>) '
476
'BODY (TEXT PLAIN (CHARSET US-ASCII) NIL NIL 7BIT 3028 92))',
477
'("oo \\"oo\\" oo")',
488
['BODY.PEEK', ['HEADER.FIELDS.NOT', ['subject', 'bcc', 'cc']], s],
490
['FLAGS', [r'\Seen'], 'INTERNALDATE',
491
'17-Jul-1996 02:44:25 -0700', 'RFC822.SIZE', '4286', 'ENVELOPE',
492
['Wed, 17 Jul 1996 02:23:25 -0700 (PDT)',
493
'IMAP4rev1 WG mtg summary and minutes', [["Terry Gray", None,
494
"gray", "cac.washington.edu"]], [["Terry Gray", None,
495
"gray", "cac.washington.edu"]], [["Terry Gray", None,
496
"gray", "cac.washington.edu"]], [[None, None, "imap",
497
"cac.washington.edu"]], [[None, None, "minutes",
498
"CNRI.Reston.VA.US"], ["John Klensin", None, "KLENSIN",
499
"INFOODS.MIT.EDU"]], None, None,
500
"<B27397-0100000@cac.washington.edu>"], "BODY", ["TEXT", "PLAIN",
501
["CHARSET", "US-ASCII"], None, None, "7BIT", "3028", "92"]],
511
for (case, expected) in zip(cases, answers):
512
self.assertEquals(imap4.parseNestedParens(case), [expected])
514
# XXX This code used to work, but changes occurred within the
515
# imap4.py module which made it no longer necessary for *all* of it
516
# to work. In particular, only the part that makes
517
# 'BODY.PEEK[HEADER.FIELDS.NOT (Subject Bcc Cc)]' come out correctly
518
# no longer needs to work. So, I am loathe to delete the entire
519
# section of the test. --exarkun
522
# for (case, expected) in zip(answers, cases):
523
# self.assertEquals('(' + imap4.collapseNestedLists(case) + ')', expected)
525
def testFetchParserSimple(self):
527
['ENVELOPE', 'Envelope'],
529
['INTERNALDATE', 'InternalDate'],
530
['RFC822.HEADER', 'RFC822Header'],
531
['RFC822.SIZE', 'RFC822Size'],
532
['RFC822.TEXT', 'RFC822Text'],
533
['RFC822', 'RFC822'],
535
['BODYSTRUCTURE', 'BodyStructure'],
538
for (inp, outp) in cases:
539
p = imap4._FetchParser()
541
self.assertEquals(len(p.result), 1)
542
self.failUnless(isinstance(p.result[0], getattr(p, outp)))
544
def testFetchParserMacros(self):
546
['ALL', (4, ['flags', 'internaldate', 'rfc822.size', 'envelope'])],
547
['FULL', (5, ['flags', 'internaldate', 'rfc822.size', 'envelope', 'body'])],
548
['FAST', (3, ['flags', 'internaldate', 'rfc822.size'])],
551
for (inp, outp) in cases:
552
p = imap4._FetchParser()
554
self.assertEquals(len(p.result), outp[0])
555
p = [str(p).lower() for p in p.result]
558
self.assertEquals(p, outp[1])
560
def testFetchParserBody(self):
561
P = imap4._FetchParser
564
p.parseString('BODY')
565
self.assertEquals(len(p.result), 1)
566
self.failUnless(isinstance(p.result[0], p.Body))
567
self.assertEquals(p.result[0].peek, False)
568
self.assertEquals(p.result[0].header, None)
569
self.assertEquals(str(p.result[0]), 'BODY')
572
p.parseString('BODY.PEEK')
573
self.assertEquals(len(p.result), 1)
574
self.failUnless(isinstance(p.result[0], p.Body))
575
self.assertEquals(p.result[0].peek, True)
576
self.assertEquals(str(p.result[0]), 'BODY')
579
p.parseString('BODY[]')
580
self.assertEquals(len(p.result), 1)
581
self.failUnless(isinstance(p.result[0], p.Body))
582
self.assertEquals(p.result[0].empty, True)
583
self.assertEquals(str(p.result[0]), 'BODY[]')
586
p.parseString('BODY[HEADER]')
587
self.assertEquals(len(p.result), 1)
588
self.failUnless(isinstance(p.result[0], p.Body))
589
self.assertEquals(p.result[0].peek, False)
590
self.failUnless(isinstance(p.result[0].header, p.Header))
591
self.assertEquals(p.result[0].header.negate, True)
592
self.assertEquals(p.result[0].header.fields, ())
593
self.assertEquals(p.result[0].empty, False)
594
self.assertEquals(str(p.result[0]), 'BODY[HEADER]')
597
p.parseString('BODY.PEEK[HEADER]')
598
self.assertEquals(len(p.result), 1)
599
self.failUnless(isinstance(p.result[0], p.Body))
600
self.assertEquals(p.result[0].peek, True)
601
self.failUnless(isinstance(p.result[0].header, p.Header))
602
self.assertEquals(p.result[0].header.negate, True)
603
self.assertEquals(p.result[0].header.fields, ())
604
self.assertEquals(p.result[0].empty, False)
605
self.assertEquals(str(p.result[0]), 'BODY[HEADER]')
608
p.parseString('BODY[HEADER.FIELDS (Subject Cc Message-Id)]')
609
self.assertEquals(len(p.result), 1)
610
self.failUnless(isinstance(p.result[0], p.Body))
611
self.assertEquals(p.result[0].peek, False)
612
self.failUnless(isinstance(p.result[0].header, p.Header))
613
self.assertEquals(p.result[0].header.negate, False)
614
self.assertEquals(p.result[0].header.fields, ['SUBJECT', 'CC', 'MESSAGE-ID'])
615
self.assertEquals(p.result[0].empty, False)
616
self.assertEquals(str(p.result[0]), 'BODY[HEADER.FIELDS (Subject Cc Message-Id)]')
619
p.parseString('BODY.PEEK[HEADER.FIELDS (Subject Cc Message-Id)]')
620
self.assertEquals(len(p.result), 1)
621
self.failUnless(isinstance(p.result[0], p.Body))
622
self.assertEquals(p.result[0].peek, True)
623
self.failUnless(isinstance(p.result[0].header, p.Header))
624
self.assertEquals(p.result[0].header.negate, False)
625
self.assertEquals(p.result[0].header.fields, ['SUBJECT', 'CC', 'MESSAGE-ID'])
626
self.assertEquals(p.result[0].empty, False)
627
self.assertEquals(str(p.result[0]), 'BODY[HEADER.FIELDS (Subject Cc Message-Id)]')
630
p.parseString('BODY.PEEK[HEADER.FIELDS.NOT (Subject Cc Message-Id)]')
631
self.assertEquals(len(p.result), 1)
632
self.failUnless(isinstance(p.result[0], p.Body))
633
self.assertEquals(p.result[0].peek, True)
634
self.failUnless(isinstance(p.result[0].header, p.Header))
635
self.assertEquals(p.result[0].header.negate, True)
636
self.assertEquals(p.result[0].header.fields, ['SUBJECT', 'CC', 'MESSAGE-ID'])
637
self.assertEquals(p.result[0].empty, False)
638
self.assertEquals(str(p.result[0]), 'BODY[HEADER.FIELDS.NOT (Subject Cc Message-Id)]')
641
p.parseString('BODY[1.MIME]<10.50>')
642
self.assertEquals(len(p.result), 1)
643
self.failUnless(isinstance(p.result[0], p.Body))
644
self.assertEquals(p.result[0].peek, False)
645
self.failUnless(isinstance(p.result[0].mime, p.MIME))
646
self.assertEquals(p.result[0].part, (0,))
647
self.assertEquals(p.result[0].partialBegin, 10)
648
self.assertEquals(p.result[0].partialLength, 50)
649
self.assertEquals(p.result[0].empty, False)
650
self.assertEquals(str(p.result[0]), 'BODY[1.MIME]<10.50>')
653
p.parseString('BODY.PEEK[1.3.9.11.HEADER.FIELDS.NOT (Message-Id Date)]<103.69>')
654
self.assertEquals(len(p.result), 1)
655
self.failUnless(isinstance(p.result[0], p.Body))
656
self.assertEquals(p.result[0].peek, True)
657
self.failUnless(isinstance(p.result[0].header, p.Header))
658
self.assertEquals(p.result[0].part, (0, 2, 8, 10))
659
self.assertEquals(p.result[0].header.fields, ['MESSAGE-ID', 'DATE'])
660
self.assertEquals(p.result[0].partialBegin, 103)
661
self.assertEquals(p.result[0].partialLength, 69)
662
self.assertEquals(p.result[0].empty, False)
663
self.assertEquals(str(p.result[0]), 'BODY[1.3.9.11.HEADER.FIELDS.NOT (Message-Id Date)]<103.69>')
668
'foo', 'bar', 'baz', StringIO('this is a file\r\n'), 'buz'
671
output = '"foo" "bar" "baz" {16}\r\nthis is a file\r\n "buz"'
673
self.assertEquals(imap4.collapseNestedLists(inputStructure), output)
675
def testQuoteAvoider(self):
677
'foo', imap4.DontQuoteMe('bar'), "baz", StringIO('this is a file\r\n'),
678
imap4.DontQuoteMe('buz'), ""
681
output = '"foo" bar "baz" {16}\r\nthis is a file\r\n buz ""'
683
self.assertEquals(imap4.collapseNestedLists(input), output)
685
def testLiterals(self):
687
('({10}\r\n0123456789)', [['0123456789']]),
690
for (case, expected) in cases:
691
self.assertEquals(imap4.parseNestedParens(case), expected)
693
def testQueryBuilder(self):
695
imap4.Query(flagged=1),
696
imap4.Query(sorted=1, unflagged=1, deleted=1),
697
imap4.Or(imap4.Query(flagged=1), imap4.Query(deleted=1)),
698
imap4.Query(before='today'),
700
imap4.Query(deleted=1),
701
imap4.Query(unseen=1),
707
imap4.Query(sorted=1, since='yesterday', smaller=1000),
708
imap4.Query(sorted=1, before='tuesday', larger=10000),
709
imap4.Query(sorted=1, unseen=1, deleted=1, before='today'),
711
imap4.Query(subject='spam')
716
imap4.Query(uid='1:5')
723
'(DELETED UNFLAGGED)',
724
'(OR FLAGGED DELETED)',
726
'(OR DELETED (OR UNSEEN NEW))',
727
'(OR (NOT (OR (SINCE "yesterday" SMALLER 1000) ' # Continuing
728
'(OR (BEFORE "tuesday" LARGER 10000) (OR (BEFORE ' # Some more
729
'"today" DELETED UNSEEN) (NOT (SUBJECT "spam")))))) ' # And more
733
for (query, expected) in zip(inputs, outputs):
734
self.assertEquals(query, expected)
736
def testIdListParser(self):
755
MessageSet(5, None) + MessageSet(1, 2),
758
MessageSet(1) + MessageSet(3) + MessageSet(5),
761
MessageSet(1, 5) + MessageSet(10, 20),
762
MessageSet(1) + MessageSet(5, 10),
763
MessageSet(1) + MessageSet(5, 10) + MessageSet(15, 20),
764
MessageSet(1, 10) + MessageSet(15) + MessageSet(20, 25),
769
1, 2, 3, 10, 11, 16, 7, 13, 17,
772
for (input, expected) in zip(inputs, outputs):
773
self.assertEquals(imap4.parseIdList(input), expected)
775
for (input, expected) in zip(inputs, lengths):
777
L = len(imap4.parseIdList(input))
780
self.assertEquals(L, expected,
781
"len(%r) = %r != %r" % (input, L, expected))
784
implements(imap4.IMailboxInfo, imap4.IMailbox, imap4.ICloseableMailbox)
786
flags = ('\\Flag1', 'Flag2', '\\AnotherSysFlag', 'LastFlag')
794
self.addListener = self.listeners.append
795
self.removeListener = self.listeners.remove
800
def getUIDValidity(self):
803
def getUIDNext(self):
804
return len(self.messages) + 1
806
def getMessageCount(self):
809
def getRecentCount(self):
812
def getUnseenCount(self):
815
def isWriteable(self):
821
def getHierarchicalDelimiter(self):
824
def requestStatus(self, names):
826
if 'MESSAGES' in names:
827
r['MESSAGES'] = self.getMessageCount()
828
if 'RECENT' in names:
829
r['RECENT'] = self.getRecentCount()
830
if 'UIDNEXT' in names:
831
r['UIDNEXT'] = self.getMessageCount() + 1
832
if 'UIDVALIDITY' in names:
833
r['UIDVALIDITY'] = self.getUID()
834
if 'UNSEEN' in names:
835
r['UNSEEN'] = self.getUnseenCount()
836
return defer.succeed(r)
838
def addMessage(self, message, flags, date = None):
839
self.messages.append((message, flags, date, self.mUID))
841
return defer.succeed(None)
845
for i in self.messages:
846
if '\\Deleted' in i[1]:
849
self.messages.remove(i)
850
return [i[3] for i in delete]
855
class Account(imap4.MemoryAccount):
856
mailboxFactory = SimpleMailbox
857
def _emptyMailbox(self, name, id):
858
return self.mailboxFactory()
860
def select(self, name, rw=1):
861
mbox = imap4.MemoryAccount.select(self, name)
866
class SimpleServer(imap4.IMAP4Server):
867
def __init__(self, *args, **kw):
868
imap4.IMAP4Server.__init__(self, *args, **kw)
870
realm.theAccount = Account('testuser')
871
portal = cred.portal.Portal(realm)
872
c = cred.checkers.InMemoryUsernamePasswordDatabaseDontUse()
875
portal.registerChecker(c)
876
self.timeoutTest = False
878
def lineReceived(self, line):
880
#Do not send a respones
883
imap4.IMAP4Server.lineReceived(self, line)
885
_username = 'testuser'
886
_password = 'password-test'
887
def authenticateLogin(self, username, password):
888
if username == self._username and password == self._password:
889
return imap4.IAccount, self.theAccount, lambda: None
890
raise cred.error.UnauthorizedLogin()
893
class SimpleClient(imap4.IMAP4Client):
894
def __init__(self, deferred, contextFactory = None):
895
imap4.IMAP4Client.__init__(self, contextFactory)
896
self.deferred = deferred
899
def serverGreeting(self, caps):
900
self.deferred.callback(None)
902
def modeChanged(self, writeable):
903
self.events.append(['modeChanged', writeable])
904
self.transport.loseConnection()
906
def flagsChanged(self, newFlags):
907
self.events.append(['flagsChanged', newFlags])
908
self.transport.loseConnection()
910
def newMessages(self, exists, recent):
911
self.events.append(['newMessages', exists, recent])
912
self.transport.loseConnection()
916
class IMAP4HelperMixin:
922
self.server = SimpleServer(contextFactory=self.serverCTX)
923
self.client = SimpleClient(d, contextFactory=self.clientCTX)
926
SimpleMailbox.messages = []
927
theAccount = Account('testuser')
928
theAccount.mboxType = SimpleMailbox
929
SimpleServer.theAccount = theAccount
936
def _cbStopClient(self, ignore):
937
self.client.transport.loseConnection()
939
def _ebGeneral(self, failure):
940
self.client.transport.loseConnection()
941
self.server.transport.loseConnection()
942
failure.raiseException()
945
return loopback.loopbackAsync(self.server, self.client)
947
class IMAP4ServerTestCase(IMAP4HelperMixin, unittest.TestCase):
948
def testCapability(self):
953
self.server.transport.loseConnection()
954
return self.client.getCapabilities().addCallback(gotCaps)
955
d1 = self.connected.addCallback(strip(getCaps)).addErrback(self._ebGeneral)
956
d = defer.gatherResults([self.loopback(), d1])
957
expected = {'IMAP4rev1': None, 'NAMESPACE': None, 'IDLE': None}
958
return d.addCallback(lambda _: self.assertEquals(expected, caps))
960
def testCapabilityWithAuth(self):
962
self.server.challengers['CRAM-MD5'] = cred.credentials.CramMD5Credentials
966
self.server.transport.loseConnection()
967
return self.client.getCapabilities().addCallback(gotCaps)
968
d1 = self.connected.addCallback(strip(getCaps)).addErrback(self._ebGeneral)
969
d = defer.gatherResults([self.loopback(), d1])
971
expCap = {'IMAP4rev1': None, 'NAMESPACE': None,
972
'IDLE': None, 'AUTH': ['CRAM-MD5']}
974
return d.addCallback(lambda _: self.assertEquals(expCap, caps))
976
def testLogout(self):
981
self.client.logout().addCallback(strip(setLoggedOut))
982
self.connected.addCallback(strip(logout)).addErrback(self._ebGeneral)
984
return d.addCallback(lambda _: self.assertEquals(self.loggedOut, 1))
987
self.responses = None
989
def setResponses(responses):
990
self.responses = responses
991
self.server.transport.loseConnection()
992
self.client.noop().addCallback(setResponses)
993
self.connected.addCallback(strip(noop)).addErrback(self._ebGeneral)
995
return d.addCallback(lambda _: self.assertEquals(self.responses, []))
999
d = self.client.login('testuser', 'password-test')
1000
d.addCallback(self._cbStopClient)
1001
d1 = self.connected.addCallback(strip(login)).addErrback(self._ebGeneral)
1002
d = defer.gatherResults([d1, self.loopback()])
1003
return d.addCallback(self._cbTestLogin)
1005
def _cbTestLogin(self, ignored):
1006
self.assertEquals(self.server.account, SimpleServer.theAccount)
1007
self.assertEquals(self.server.state, 'auth')
1009
def testFailedLogin(self):
1011
d = self.client.login('testuser', 'wrong-password')
1012
d.addBoth(self._cbStopClient)
1014
d1 = self.connected.addCallback(strip(login)).addErrback(self._ebGeneral)
1015
d2 = self.loopback()
1016
d = defer.gatherResults([d1, d2])
1017
return d.addCallback(self._cbTestFailedLogin)
1019
def _cbTestFailedLogin(self, ignored):
1020
self.assertEquals(self.server.account, None)
1021
self.assertEquals(self.server.state, 'unauth')
1024
def testLoginRequiringQuoting(self):
1025
self.server._username = '{test}user'
1026
self.server._password = '{test}password'
1029
d = self.client.login('{test}user', '{test}password')
1030
d.addBoth(self._cbStopClient)
1032
d1 = self.connected.addCallback(strip(login)).addErrback(self._ebGeneral)
1033
d = defer.gatherResults([self.loopback(), d1])
1034
return d.addCallback(self._cbTestLoginRequiringQuoting)
1036
def _cbTestLoginRequiringQuoting(self, ignored):
1037
self.assertEquals(self.server.account, SimpleServer.theAccount)
1038
self.assertEquals(self.server.state, 'auth')
1041
def testNamespace(self):
1042
self.namespaceArgs = None
1044
return self.client.login('testuser', 'password-test')
1046
def gotNamespace(args):
1047
self.namespaceArgs = args
1048
self._cbStopClient(None)
1049
return self.client.namespace().addCallback(gotNamespace)
1051
d1 = self.connected.addCallback(strip(login))
1052
d1.addCallback(strip(namespace))
1053
d1.addErrback(self._ebGeneral)
1054
d2 = self.loopback()
1055
d = defer.gatherResults([d1, d2])
1056
d.addCallback(lambda _: self.assertEquals(self.namespaceArgs,
1057
[[['', '/']], [], []]))
1060
def testSelect(self):
1061
SimpleServer.theAccount.addMailbox('test-mailbox')
1062
self.selectedArgs = None
1064
return self.client.login('testuser', 'password-test')
1067
self.selectedArgs = args
1068
self._cbStopClient(None)
1069
d = self.client.select('test-mailbox')
1070
d.addCallback(selected)
1073
d1 = self.connected.addCallback(strip(login))
1074
d1.addCallback(strip(select))
1075
d1.addErrback(self._ebGeneral)
1076
d2 = self.loopback()
1077
return defer.gatherResults([d1, d2]).addCallback(self._cbTestSelect)
1079
def _cbTestSelect(self, ignored):
1080
mbox = SimpleServer.theAccount.mailboxes['TEST-MAILBOX']
1081
self.assertEquals(self.server.mbox, mbox)
1082
self.assertEquals(self.selectedArgs, {
1083
'EXISTS': 9, 'RECENT': 3, 'UIDVALIDITY': 42,
1084
'FLAGS': ('\\Flag1', 'Flag2', '\\AnotherSysFlag', 'LastFlag'),
1089
def test_examine(self):
1091
L{IMAP4Client.examine} issues an I{EXAMINE} command to the server and
1092
returns a L{Deferred} which fires with a C{dict} with as many of the
1093
following keys as the server includes in its response: C{'FLAGS'},
1094
C{'EXISTS'}, C{'RECENT'}, C{'UNSEEN'}, C{'READ-WRITE'}, C{'READ-ONLY'},
1095
C{'UIDVALIDITY'}, and C{'PERMANENTFLAGS'}.
1097
Unfortunately the server doesn't generate all of these so it's hard to
1098
test the client's handling of them here. See
1099
L{IMAP4ClientExamineTests} below.
1101
See U{RFC 3501<http://www.faqs.org/rfcs/rfc3501.html>}, section 6.3.2,
1104
SimpleServer.theAccount.addMailbox('test-mailbox')
1105
self.examinedArgs = None
1107
return self.client.login('testuser', 'password-test')
1110
self.examinedArgs = args
1111
self._cbStopClient(None)
1112
d = self.client.examine('test-mailbox')
1113
d.addCallback(examined)
1116
d1 = self.connected.addCallback(strip(login))
1117
d1.addCallback(strip(examine))
1118
d1.addErrback(self._ebGeneral)
1119
d2 = self.loopback()
1120
d = defer.gatherResults([d1, d2])
1121
return d.addCallback(self._cbTestExamine)
1124
def _cbTestExamine(self, ignored):
1125
mbox = SimpleServer.theAccount.mailboxes['TEST-MAILBOX']
1126
self.assertEquals(self.server.mbox, mbox)
1127
self.assertEquals(self.examinedArgs, {
1128
'EXISTS': 9, 'RECENT': 3, 'UIDVALIDITY': 42,
1129
'FLAGS': ('\\Flag1', 'Flag2', '\\AnotherSysFlag', 'LastFlag'),
1130
'READ-WRITE': False})
1133
def testCreate(self):
1134
succeed = ('testbox', 'test/box', 'test/', 'test/box/box', 'INBOX')
1135
fail = ('testbox', 'test/box')
1137
def cb(): self.result.append(1)
1138
def eb(failure): self.result.append(0)
1141
return self.client.login('testuser', 'password-test')
1143
for name in succeed + fail:
1144
d = self.client.create(name)
1145
d.addCallback(strip(cb)).addErrback(eb)
1146
d.addCallbacks(self._cbStopClient, self._ebGeneral)
1149
d1 = self.connected.addCallback(strip(login)).addCallback(strip(create))
1150
d2 = self.loopback()
1151
d = defer.gatherResults([d1, d2])
1152
return d.addCallback(self._cbTestCreate, succeed, fail)
1154
def _cbTestCreate(self, ignored, succeed, fail):
1155
self.assertEquals(self.result, [1] * len(succeed) + [0] * len(fail))
1156
mbox = SimpleServer.theAccount.mailboxes.keys()
1157
answers = ['inbox', 'testbox', 'test/box', 'test', 'test/box/box']
1160
self.assertEquals(mbox, [a.upper() for a in answers])
1162
def testDelete(self):
1163
SimpleServer.theAccount.addMailbox('delete/me')
1166
return self.client.login('testuser', 'password-test')
1168
return self.client.delete('delete/me')
1169
d1 = self.connected.addCallback(strip(login))
1170
d1.addCallbacks(strip(delete), self._ebGeneral)
1171
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1172
d2 = self.loopback()
1173
d = defer.gatherResults([d1, d2])
1174
d.addCallback(lambda _:
1175
self.assertEquals(SimpleServer.theAccount.mailboxes.keys(), []))
1178
def testIllegalInboxDelete(self):
1181
return self.client.login('testuser', 'password-test')
1183
return self.client.delete('inbox')
1185
self.stashed = result
1187
d1 = self.connected.addCallback(strip(login))
1188
d1.addCallbacks(strip(delete), self._ebGeneral)
1190
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1191
d2 = self.loopback()
1192
d = defer.gatherResults([d1, d2])
1193
d.addCallback(lambda _: self.failUnless(isinstance(self.stashed,
1198
def testNonExistentDelete(self):
1200
return self.client.login('testuser', 'password-test')
1202
return self.client.delete('delete/me')
1203
def deleteFailed(failure):
1204
self.failure = failure
1207
d1 = self.connected.addCallback(strip(login))
1208
d1.addCallback(strip(delete)).addErrback(deleteFailed)
1209
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1210
d2 = self.loopback()
1211
d = defer.gatherResults([d1, d2])
1212
d.addCallback(lambda _: self.assertEquals(str(self.failure.value),
1217
def testIllegalDelete(self):
1219
m.flags = (r'\Noselect',)
1220
SimpleServer.theAccount.addMailbox('delete', m)
1221
SimpleServer.theAccount.addMailbox('delete/me')
1224
return self.client.login('testuser', 'password-test')
1226
return self.client.delete('delete')
1227
def deleteFailed(failure):
1228
self.failure = failure
1231
d1 = self.connected.addCallback(strip(login))
1232
d1.addCallback(strip(delete)).addErrback(deleteFailed)
1233
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1234
d2 = self.loopback()
1235
d = defer.gatherResults([d1, d2])
1236
expected = "Hierarchically inferior mailboxes exist and \\Noselect is set"
1237
d.addCallback(lambda _:
1238
self.assertEquals(str(self.failure.value), expected))
1241
def testRename(self):
1242
SimpleServer.theAccount.addMailbox('oldmbox')
1244
return self.client.login('testuser', 'password-test')
1246
return self.client.rename('oldmbox', 'newname')
1248
d1 = self.connected.addCallback(strip(login))
1249
d1.addCallbacks(strip(rename), self._ebGeneral)
1250
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1251
d2 = self.loopback()
1252
d = defer.gatherResults([d1, d2])
1253
d.addCallback(lambda _:
1254
self.assertEquals(SimpleServer.theAccount.mailboxes.keys(),
1258
def testIllegalInboxRename(self):
1261
return self.client.login('testuser', 'password-test')
1263
return self.client.rename('inbox', 'frotz')
1265
self.stashed = stuff
1267
d1 = self.connected.addCallback(strip(login))
1268
d1.addCallbacks(strip(rename), self._ebGeneral)
1270
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1271
d2 = self.loopback()
1272
d = defer.gatherResults([d1, d2])
1273
d.addCallback(lambda _:
1274
self.failUnless(isinstance(self.stashed, failure.Failure)))
1277
def testHierarchicalRename(self):
1278
SimpleServer.theAccount.create('oldmbox/m1')
1279
SimpleServer.theAccount.create('oldmbox/m2')
1281
return self.client.login('testuser', 'password-test')
1283
return self.client.rename('oldmbox', 'newname')
1285
d1 = self.connected.addCallback(strip(login))
1286
d1.addCallbacks(strip(rename), self._ebGeneral)
1287
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1288
d2 = self.loopback()
1289
d = defer.gatherResults([d1, d2])
1290
return d.addCallback(self._cbTestHierarchicalRename)
1292
def _cbTestHierarchicalRename(self, ignored):
1293
mboxes = SimpleServer.theAccount.mailboxes.keys()
1294
expected = ['newname', 'newname/m1', 'newname/m2']
1296
self.assertEquals(mboxes, [s.upper() for s in expected])
1298
def testSubscribe(self):
1300
return self.client.login('testuser', 'password-test')
1302
return self.client.subscribe('this/mbox')
1304
d1 = self.connected.addCallback(strip(login))
1305
d1.addCallbacks(strip(subscribe), self._ebGeneral)
1306
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1307
d2 = self.loopback()
1308
d = defer.gatherResults([d1, d2])
1309
d.addCallback(lambda _:
1310
self.assertEquals(SimpleServer.theAccount.subscriptions,
1314
def testUnsubscribe(self):
1315
SimpleServer.theAccount.subscriptions = ['THIS/MBOX', 'THAT/MBOX']
1317
return self.client.login('testuser', 'password-test')
1319
return self.client.unsubscribe('this/mbox')
1321
d1 = self.connected.addCallback(strip(login))
1322
d1.addCallbacks(strip(unsubscribe), self._ebGeneral)
1323
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1324
d2 = self.loopback()
1325
d = defer.gatherResults([d1, d2])
1326
d.addCallback(lambda _:
1327
self.assertEquals(SimpleServer.theAccount.subscriptions,
1331
def _listSetup(self, f):
1332
SimpleServer.theAccount.addMailbox('root/subthing')
1333
SimpleServer.theAccount.addMailbox('root/another-thing')
1334
SimpleServer.theAccount.addMailbox('non-root/subthing')
1337
return self.client.login('testuser', 'password-test')
1338
def listed(answers):
1339
self.listed = answers
1342
d1 = self.connected.addCallback(strip(login))
1343
d1.addCallbacks(strip(f), self._ebGeneral)
1344
d1.addCallbacks(listed, self._ebGeneral)
1345
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1346
d2 = self.loopback()
1347
return defer.gatherResults([d1, d2]).addCallback(lambda _: self.listed)
1351
return self.client.list('root', '%')
1352
d = self._listSetup(list)
1353
d.addCallback(lambda listed: self.assertEquals(
1356
(SimpleMailbox.flags, "/", "ROOT/SUBTHING"),
1357
(SimpleMailbox.flags, "/", "ROOT/ANOTHER-THING")
1363
SimpleServer.theAccount.subscribe('ROOT/SUBTHING')
1365
return self.client.lsub('root', '%')
1366
d = self._listSetup(lsub)
1367
d.addCallback(self.assertEquals,
1368
[(SimpleMailbox.flags, "/", "ROOT/SUBTHING")])
1371
def testStatus(self):
1372
SimpleServer.theAccount.addMailbox('root/subthing')
1374
return self.client.login('testuser', 'password-test')
1376
return self.client.status('root/subthing', 'MESSAGES', 'UIDNEXT', 'UNSEEN')
1377
def statused(result):
1378
self.statused = result
1380
self.statused = None
1381
d1 = self.connected.addCallback(strip(login))
1382
d1.addCallbacks(strip(status), self._ebGeneral)
1383
d1.addCallbacks(statused, self._ebGeneral)
1384
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1385
d2 = self.loopback()
1386
d = defer.gatherResults([d1, d2])
1387
d.addCallback(lambda _: self.assertEquals(
1389
{'MESSAGES': 9, 'UIDNEXT': '10', 'UNSEEN': 4}
1393
def testFailedStatus(self):
1395
return self.client.login('testuser', 'password-test')
1397
return self.client.status('root/nonexistent', 'MESSAGES', 'UIDNEXT', 'UNSEEN')
1398
def statused(result):
1399
self.statused = result
1400
def failed(failure):
1401
self.failure = failure
1403
self.statused = self.failure = None
1404
d1 = self.connected.addCallback(strip(login))
1405
d1.addCallbacks(strip(status), self._ebGeneral)
1406
d1.addCallbacks(statused, failed)
1407
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1408
d2 = self.loopback()
1409
return defer.gatherResults([d1, d2]).addCallback(self._cbTestFailedStatus)
1411
def _cbTestFailedStatus(self, ignored):
1416
self.failure.value.args,
1417
('Could not open mailbox',)
1420
def testFullAppend(self):
1421
infile = util.sibpath(__file__, 'rfc822.message')
1422
message = open(infile)
1423
SimpleServer.theAccount.addMailbox('root/subthing')
1425
return self.client.login('testuser', 'password-test')
1427
return self.client.append(
1430
('\\SEEN', '\\DELETED'),
1431
'Tue, 17 Jun 2003 11:22:16 -0600 (MDT)',
1434
d1 = self.connected.addCallback(strip(login))
1435
d1.addCallbacks(strip(append), self._ebGeneral)
1436
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1437
d2 = self.loopback()
1438
d = defer.gatherResults([d1, d2])
1439
return d.addCallback(self._cbTestFullAppend, infile)
1441
def _cbTestFullAppend(self, ignored, infile):
1442
mb = SimpleServer.theAccount.mailboxes['ROOT/SUBTHING']
1443
self.assertEquals(1, len(mb.messages))
1445
(['\\SEEN', '\\DELETED'], 'Tue, 17 Jun 2003 11:22:16 -0600 (MDT)', 0),
1448
self.assertEquals(open(infile).read(), mb.messages[0][0].getvalue())
1450
def testPartialAppend(self):
1451
infile = util.sibpath(__file__, 'rfc822.message')
1452
message = open(infile)
1453
SimpleServer.theAccount.addMailbox('PARTIAL/SUBTHING')
1455
return self.client.login('testuser', 'password-test')
1457
message = file(infile)
1458
return self.client.sendCommand(
1461
'PARTIAL/SUBTHING (\\SEEN) "Right now" {%d}' % os.path.getsize(infile),
1462
(), self.client._IMAP4Client__cbContinueAppend, message
1465
d1 = self.connected.addCallback(strip(login))
1466
d1.addCallbacks(strip(append), self._ebGeneral)
1467
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1468
d2 = self.loopback()
1469
d = defer.gatherResults([d1, d2])
1470
return d.addCallback(self._cbTestPartialAppend, infile)
1472
def _cbTestPartialAppend(self, ignored, infile):
1473
mb = SimpleServer.theAccount.mailboxes['PARTIAL/SUBTHING']
1474
self.assertEquals(1, len(mb.messages))
1476
(['\\SEEN'], 'Right now', 0),
1479
self.assertEquals(open(infile).read(), mb.messages[0][0].getvalue())
1481
def testCheck(self):
1482
SimpleServer.theAccount.addMailbox('root/subthing')
1484
return self.client.login('testuser', 'password-test')
1486
return self.client.select('root/subthing')
1488
return self.client.check()
1490
d = self.connected.addCallback(strip(login))
1491
d.addCallbacks(strip(select), self._ebGeneral)
1492
d.addCallbacks(strip(check), self._ebGeneral)
1493
d.addCallbacks(self._cbStopClient, self._ebGeneral)
1494
return self.loopback()
1496
# Okay, that was fun
1498
def testClose(self):
1501
('Message 1', ('\\Deleted', 'AnotherFlag'), None, 0),
1502
('Message 2', ('AnotherFlag',), None, 1),
1503
('Message 3', ('\\Deleted',), None, 2),
1505
SimpleServer.theAccount.addMailbox('mailbox', m)
1507
return self.client.login('testuser', 'password-test')
1509
return self.client.select('mailbox')
1511
return self.client.close()
1513
d = self.connected.addCallback(strip(login))
1514
d.addCallbacks(strip(select), self._ebGeneral)
1515
d.addCallbacks(strip(close), self._ebGeneral)
1516
d.addCallbacks(self._cbStopClient, self._ebGeneral)
1517
d2 = self.loopback()
1518
return defer.gatherResults([d, d2]).addCallback(self._cbTestClose, m)
1520
def _cbTestClose(self, ignored, m):
1521
self.assertEquals(len(m.messages), 1)
1522
self.assertEquals(m.messages[0], ('Message 2', ('AnotherFlag',), None, 1))
1523
self.failUnless(m.closed)
1525
def testExpunge(self):
1528
('Message 1', ('\\Deleted', 'AnotherFlag'), None, 0),
1529
('Message 2', ('AnotherFlag',), None, 1),
1530
('Message 3', ('\\Deleted',), None, 2),
1532
SimpleServer.theAccount.addMailbox('mailbox', m)
1534
return self.client.login('testuser', 'password-test')
1536
return self.client.select('mailbox')
1538
return self.client.expunge()
1539
def expunged(results):
1540
self.failIf(self.server.mbox is None)
1541
self.results = results
1544
d1 = self.connected.addCallback(strip(login))
1545
d1.addCallbacks(strip(select), self._ebGeneral)
1546
d1.addCallbacks(strip(expunge), self._ebGeneral)
1547
d1.addCallbacks(expunged, self._ebGeneral)
1548
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1549
d2 = self.loopback()
1550
d = defer.gatherResults([d1, d2])
1551
return d.addCallback(self._cbTestExpunge, m)
1553
def _cbTestExpunge(self, ignored, m):
1554
self.assertEquals(len(m.messages), 1)
1555
self.assertEquals(m.messages[0], ('Message 2', ('AnotherFlag',), None, 1))
1557
self.assertEquals(self.results, [0, 2])
1561
class IMAP4ServerSearchTestCase(IMAP4HelperMixin, unittest.TestCase):
1563
Tests for the behavior of the search_* functions in L{imap4.IMAP4Server}.
1566
IMAP4HelperMixin.setUp(self)
1567
self.earlierQuery = ["10-Dec-2009"]
1568
self.sameDateQuery = ["13-Dec-2009"]
1569
self.laterQuery = ["16-Dec-2009"]
1571
self.msg = FakeyMessage({"date" : "Mon, 13 Dec 2009 21:25:10 GMT"}, [],
1575
def test_searchSentBefore(self):
1577
L{imap4.IMAP4Server.search_SENTBEFORE} returns True if the message date
1578
is earlier than the query date.
1581
self.server.search_SENTBEFORE(self.earlierQuery, self.seq, self.msg))
1583
self.server.search_SENTBEFORE(self.laterQuery, self.seq, self.msg))
1586
def test_searchSentOn(self):
1588
L{imap4.IMAP4Server.search_SENTON} returns True if the message date is
1589
the same as the query date.
1592
self.server.search_SENTON(self.earlierQuery, self.seq, self.msg))
1594
self.server.search_SENTON(self.sameDateQuery, self.seq, self.msg))
1596
self.server.search_SENTON(self.laterQuery, self.seq, self.msg))
1599
def test_searchSentSince(self):
1601
L{imap4.IMAP4Server.search_SENTSINCE} returns True if the message date
1602
is later than the query date.
1605
self.server.search_SENTSINCE(self.earlierQuery, self.seq, self.msg))
1607
self.server.search_SENTSINCE(self.laterQuery, self.seq, self.msg))
1610
def test_searchOr(self):
1612
L{imap4.IMAP4Server.search_OR} returns true if either of the two
1613
expressions supplied to it returns true and returns false if neither
1617
self.server.search_OR(
1618
["SENTSINCE"] + self.earlierQuery +
1619
["SENTSINCE"] + self.laterQuery,
1620
self.seq, self.msg, None))
1622
self.server.search_OR(
1623
["SENTSINCE"] + self.laterQuery +
1624
["SENTSINCE"] + self.earlierQuery,
1625
self.seq, self.msg, None))
1627
self.server.search_OR(
1628
["SENTON"] + self.laterQuery +
1629
["SENTSINCE"] + self.laterQuery,
1630
self.seq, self.msg, None))
1633
def test_searchNot(self):
1635
L{imap4.IMAP4Server.search_NOT} returns the negation of the result
1636
of the expression supplied to it.
1638
self.assertFalse(self.server.search_NOT(
1639
["SENTSINCE"] + self.earlierQuery, self.seq, self.msg, None))
1640
self.assertTrue(self.server.search_NOT(
1641
["SENTON"] + self.laterQuery, self.seq, self.msg, None))
1648
def requestAvatar(self, avatarId, mind, *interfaces):
1649
return imap4.IAccount, self.theAccount, lambda: None
1652
credentialInterfaces = (cred.credentials.IUsernameHashedPassword, cred.credentials.IUsernamePassword)
1655
'testuser': 'secret'
1658
def requestAvatarId(self, credentials):
1659
if credentials.username in self.users:
1660
return defer.maybeDeferred(
1661
credentials.checkPassword, self.users[credentials.username]
1662
).addCallback(self._cbCheck, credentials.username)
1664
def _cbCheck(self, result, username):
1667
raise cred.error.UnauthorizedLogin()
1669
class AuthenticatorTestCase(IMAP4HelperMixin, unittest.TestCase):
1671
IMAP4HelperMixin.setUp(self)
1674
realm.theAccount = Account('testuser')
1675
portal = cred.portal.Portal(realm)
1676
portal.registerChecker(TestChecker())
1677
self.server.portal = portal
1679
self.authenticated = 0
1680
self.account = realm.theAccount
1682
def testCramMD5(self):
1683
self.server.challengers['CRAM-MD5'] = cred.credentials.CramMD5Credentials
1684
cAuth = imap4.CramMD5ClientAuthenticator('testuser')
1685
self.client.registerAuthenticator(cAuth)
1688
return self.client.authenticate('secret')
1690
self.authenticated = 1
1692
d1 = self.connected.addCallback(strip(auth))
1693
d1.addCallbacks(strip(authed), self._ebGeneral)
1694
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1695
d2 = self.loopback()
1696
d = defer.gatherResults([d1, d2])
1697
return d.addCallback(self._cbTestCramMD5)
1699
def _cbTestCramMD5(self, ignored):
1700
self.assertEquals(self.authenticated, 1)
1701
self.assertEquals(self.server.account, self.account)
1703
def testFailedCramMD5(self):
1704
self.server.challengers['CRAM-MD5'] = cred.credentials.CramMD5Credentials
1705
cAuth = imap4.CramMD5ClientAuthenticator('testuser')
1706
self.client.registerAuthenticator(cAuth)
1709
return self.client.authenticate('not the secret')
1711
self.authenticated = 1
1713
self.authenticated = -1
1715
d1 = self.connected.addCallback(strip(misauth))
1716
d1.addCallbacks(strip(authed), strip(misauthed))
1717
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1718
d = defer.gatherResults([self.loopback(), d1])
1719
return d.addCallback(self._cbTestFailedCramMD5)
1721
def _cbTestFailedCramMD5(self, ignored):
1722
self.assertEquals(self.authenticated, -1)
1723
self.assertEquals(self.server.account, None)
1725
def testLOGIN(self):
1726
self.server.challengers['LOGIN'] = imap4.LOGINCredentials
1727
cAuth = imap4.LOGINAuthenticator('testuser')
1728
self.client.registerAuthenticator(cAuth)
1731
return self.client.authenticate('secret')
1733
self.authenticated = 1
1735
d1 = self.connected.addCallback(strip(auth))
1736
d1.addCallbacks(strip(authed), self._ebGeneral)
1737
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1738
d = defer.gatherResults([self.loopback(), d1])
1739
return d.addCallback(self._cbTestLOGIN)
1741
def _cbTestLOGIN(self, ignored):
1742
self.assertEquals(self.authenticated, 1)
1743
self.assertEquals(self.server.account, self.account)
1745
def testFailedLOGIN(self):
1746
self.server.challengers['LOGIN'] = imap4.LOGINCredentials
1747
cAuth = imap4.LOGINAuthenticator('testuser')
1748
self.client.registerAuthenticator(cAuth)
1751
return self.client.authenticate('not the secret')
1753
self.authenticated = 1
1755
self.authenticated = -1
1757
d1 = self.connected.addCallback(strip(misauth))
1758
d1.addCallbacks(strip(authed), strip(misauthed))
1759
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1760
d = defer.gatherResults([self.loopback(), d1])
1761
return d.addCallback(self._cbTestFailedLOGIN)
1763
def _cbTestFailedLOGIN(self, ignored):
1764
self.assertEquals(self.authenticated, -1)
1765
self.assertEquals(self.server.account, None)
1767
def testPLAIN(self):
1768
self.server.challengers['PLAIN'] = imap4.PLAINCredentials
1769
cAuth = imap4.PLAINAuthenticator('testuser')
1770
self.client.registerAuthenticator(cAuth)
1773
return self.client.authenticate('secret')
1775
self.authenticated = 1
1777
d1 = self.connected.addCallback(strip(auth))
1778
d1.addCallbacks(strip(authed), self._ebGeneral)
1779
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1780
d = defer.gatherResults([self.loopback(), d1])
1781
return d.addCallback(self._cbTestPLAIN)
1783
def _cbTestPLAIN(self, ignored):
1784
self.assertEquals(self.authenticated, 1)
1785
self.assertEquals(self.server.account, self.account)
1787
def testFailedPLAIN(self):
1788
self.server.challengers['PLAIN'] = imap4.PLAINCredentials
1789
cAuth = imap4.PLAINAuthenticator('testuser')
1790
self.client.registerAuthenticator(cAuth)
1793
return self.client.authenticate('not the secret')
1795
self.authenticated = 1
1797
self.authenticated = -1
1799
d1 = self.connected.addCallback(strip(misauth))
1800
d1.addCallbacks(strip(authed), strip(misauthed))
1801
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
1802
d = defer.gatherResults([self.loopback(), d1])
1803
return d.addCallback(self._cbTestFailedPLAIN)
1805
def _cbTestFailedPLAIN(self, ignored):
1806
self.assertEquals(self.authenticated, -1)
1807
self.assertEquals(self.server.account, None)
1811
class SASLPLAINTestCase(unittest.TestCase):
1813
Tests for I{SASL PLAIN} authentication, as implemented by
1814
L{imap4.PLAINAuthenticator} and L{imap4.PLAINCredentials}.
1816
@see: U{http://www.faqs.org/rfcs/rfc2595.html}
1817
@see: U{http://www.faqs.org/rfcs/rfc4616.html}
1819
def test_authenticatorChallengeResponse(self):
1821
L{PLAINAuthenticator.challengeResponse} returns challenge strings of
1824
NUL<authn-id>NUL<secret>
1826
username = 'testuser'
1829
cAuth = imap4.PLAINAuthenticator(username)
1830
response = cAuth.challengeResponse(secret, chal)
1831
self.assertEquals(response, '\0%s\0%s' % (username, secret))
1834
def test_credentialsSetResponse(self):
1836
L{PLAINCredentials.setResponse} parses challenge strings of the
1839
NUL<authn-id>NUL<secret>
1841
cred = imap4.PLAINCredentials()
1842
cred.setResponse('\0testuser\0secret')
1843
self.assertEquals(cred.username, 'testuser')
1844
self.assertEquals(cred.password, 'secret')
1847
def test_credentialsInvalidResponse(self):
1849
L{PLAINCredentials.setResponse} raises L{imap4.IllegalClientResponse}
1850
when passed a string not of the expected form.
1852
cred = imap4.PLAINCredentials()
1854
imap4.IllegalClientResponse, cred.setResponse, 'hello')
1856
imap4.IllegalClientResponse, cred.setResponse, 'hello\0world')
1858
imap4.IllegalClientResponse, cred.setResponse,
1859
'hello\0world\0Zoom!\0')
1863
class UnsolicitedResponseTestCase(IMAP4HelperMixin, unittest.TestCase):
1864
def testReadWrite(self):
1866
return self.client.login('testuser', 'password-test')
1868
self.server.modeChanged(1)
1870
d1 = self.connected.addCallback(strip(login))
1871
d1.addCallback(strip(loggedIn)).addErrback(self._ebGeneral)
1872
d = defer.gatherResults([self.loopback(), d1])
1873
return d.addCallback(self._cbTestReadWrite)
1875
def _cbTestReadWrite(self, ignored):
1876
E = self.client.events
1877
self.assertEquals(E, [['modeChanged', 1]])
1879
def testReadOnly(self):
1881
return self.client.login('testuser', 'password-test')
1883
self.server.modeChanged(0)
1885
d1 = self.connected.addCallback(strip(login))
1886
d1.addCallback(strip(loggedIn)).addErrback(self._ebGeneral)
1887
d = defer.gatherResults([self.loopback(), d1])
1888
return d.addCallback(self._cbTestReadOnly)
1890
def _cbTestReadOnly(self, ignored):
1891
E = self.client.events
1892
self.assertEquals(E, [['modeChanged', 0]])
1894
def testFlagChange(self):
1896
1: ['\\Answered', '\\Deleted'],
1901
return self.client.login('testuser', 'password-test')
1903
self.server.flagsChanged(flags)
1905
d1 = self.connected.addCallback(strip(login))
1906
d1.addCallback(strip(loggedIn)).addErrback(self._ebGeneral)
1907
d = defer.gatherResults([self.loopback(), d1])
1908
return d.addCallback(self._cbTestFlagChange, flags)
1910
def _cbTestFlagChange(self, ignored, flags):
1911
E = self.client.events
1912
expect = [['flagsChanged', {x[0]: x[1]}] for x in flags.items()]
1915
self.assertEquals(E, expect)
1917
def testNewMessages(self):
1919
return self.client.login('testuser', 'password-test')
1921
self.server.newMessages(10, None)
1923
d1 = self.connected.addCallback(strip(login))
1924
d1.addCallback(strip(loggedIn)).addErrback(self._ebGeneral)
1925
d = defer.gatherResults([self.loopback(), d1])
1926
return d.addCallback(self._cbTestNewMessages)
1928
def _cbTestNewMessages(self, ignored):
1929
E = self.client.events
1930
self.assertEquals(E, [['newMessages', 10, None]])
1932
def testNewRecentMessages(self):
1934
return self.client.login('testuser', 'password-test')
1936
self.server.newMessages(None, 10)
1938
d1 = self.connected.addCallback(strip(login))
1939
d1.addCallback(strip(loggedIn)).addErrback(self._ebGeneral)
1940
d = defer.gatherResults([self.loopback(), d1])
1941
return d.addCallback(self._cbTestNewRecentMessages)
1943
def _cbTestNewRecentMessages(self, ignored):
1944
E = self.client.events
1945
self.assertEquals(E, [['newMessages', None, 10]])
1947
def testNewMessagesAndRecent(self):
1949
return self.client.login('testuser', 'password-test')
1951
self.server.newMessages(20, 10)
1953
d1 = self.connected.addCallback(strip(login))
1954
d1.addCallback(strip(loggedIn)).addErrback(self._ebGeneral)
1955
d = defer.gatherResults([self.loopback(), d1])
1956
return d.addCallback(self._cbTestNewMessagesAndRecent)
1958
def _cbTestNewMessagesAndRecent(self, ignored):
1959
E = self.client.events
1960
self.assertEquals(E, [['newMessages', 20, None], ['newMessages', None, 10]])
1963
class ClientCapabilityTests(unittest.TestCase):
1965
Tests for issuance of the CAPABILITY command and handling of its response.
1969
Create an L{imap4.IMAP4Client} connected to a L{StringTransport}.
1971
self.transport = StringTransport()
1972
self.protocol = imap4.IMAP4Client()
1973
self.protocol.makeConnection(self.transport)
1974
self.protocol.dataReceived('* OK [IMAP4rev1]\r\n')
1977
def test_simpleAtoms(self):
1979
A capability response consisting only of atoms without C{'='} in them
1980
should result in a dict mapping those atoms to C{None}.
1982
capabilitiesResult = self.protocol.getCapabilities(useCache=False)
1983
self.protocol.dataReceived('* CAPABILITY IMAP4rev1 LOGINDISABLED\r\n')
1984
self.protocol.dataReceived('0001 OK Capability completed.\r\n')
1985
def gotCapabilities(capabilities):
1987
capabilities, {'IMAP4rev1': None, 'LOGINDISABLED': None})
1988
capabilitiesResult.addCallback(gotCapabilities)
1989
return capabilitiesResult
1992
def test_categoryAtoms(self):
1994
A capability response consisting of atoms including C{'='} should have
1995
those atoms split on that byte and have capabilities in the same
1996
category aggregated into lists in the resulting dictionary.
1998
(n.b. - I made up the word "category atom"; the protocol has no notion
1999
of structure here, but rather allows each capability to define the
2000
semantics of its entry in the capability response in a freeform manner.
2001
If I had realized this earlier, the API for capabilities would look
2002
different. As it is, we can hope that no one defines any crazy
2003
semantics which are incompatible with this API, or try to figure out a
2004
better API when someone does. -exarkun)
2006
capabilitiesResult = self.protocol.getCapabilities(useCache=False)
2007
self.protocol.dataReceived('* CAPABILITY IMAP4rev1 AUTH=LOGIN AUTH=PLAIN\r\n')
2008
self.protocol.dataReceived('0001 OK Capability completed.\r\n')
2009
def gotCapabilities(capabilities):
2011
capabilities, {'IMAP4rev1': None, 'AUTH': ['LOGIN', 'PLAIN']})
2012
capabilitiesResult.addCallback(gotCapabilities)
2013
return capabilitiesResult
2016
def test_mixedAtoms(self):
2018
A capability response consisting of both simple and category atoms of
2019
the same type should result in a list containing C{None} as well as the
2020
values for the category.
2022
capabilitiesResult = self.protocol.getCapabilities(useCache=False)
2023
# Exercise codepath for both orderings of =-having and =-missing
2025
self.protocol.dataReceived(
2026
'* CAPABILITY IMAP4rev1 FOO FOO=BAR BAR=FOO BAR\r\n')
2027
self.protocol.dataReceived('0001 OK Capability completed.\r\n')
2028
def gotCapabilities(capabilities):
2029
self.assertEqual(capabilities, {'IMAP4rev1': None,
2030
'FOO': [None, 'BAR'],
2031
'BAR': ['FOO', None]})
2032
capabilitiesResult.addCallback(gotCapabilities)
2033
return capabilitiesResult
2037
class StillSimplerClient(imap4.IMAP4Client):
2039
An IMAP4 client which keeps track of unsolicited flag changes.
2042
imap4.IMAP4Client.__init__(self)
2046
def flagsChanged(self, newFlags):
2047
self.flags.update(newFlags)
2051
class HandCraftedTestCase(IMAP4HelperMixin, unittest.TestCase):
2052
def testTrailingLiteral(self):
2053
transport = StringTransport()
2054
c = imap4.IMAP4Client()
2055
c.makeConnection(transport)
2056
c.lineReceived('* OK [IMAP4rev1]')
2058
def cbSelect(ignored):
2059
d = c.fetchMessage('1')
2060
c.dataReceived('* 1 FETCH (RFC822 {10}\r\n0123456789\r\n RFC822.SIZE 10)\r\n')
2061
c.dataReceived('0003 OK FETCH\r\n')
2064
def cbLogin(ignored):
2065
d = c.select('inbox')
2066
c.lineReceived('0002 OK SELECT')
2067
d.addCallback(cbSelect)
2070
d = c.login('blah', 'blah')
2071
c.dataReceived('0001 OK LOGIN\r\n')
2072
d.addCallback(cbLogin)
2076
def testPathelogicalScatteringOfLiterals(self):
2077
self.server.checker.addUser('testuser', 'password-test')
2078
transport = StringTransport()
2079
self.server.makeConnection(transport)
2082
self.server.dataReceived("01 LOGIN {8}\r\n")
2083
self.assertEquals(transport.value(), "+ Ready for 8 octets of text\r\n")
2086
self.server.dataReceived("testuser {13}\r\n")
2087
self.assertEquals(transport.value(), "+ Ready for 13 octets of text\r\n")
2090
self.server.dataReceived("password-test\r\n")
2091
self.assertEquals(transport.value(), "01 OK LOGIN succeeded\r\n")
2092
self.assertEquals(self.server.state, 'auth')
2094
self.server.connectionLost(error.ConnectionDone("Connection done."))
2097
def test_unsolicitedResponseMixedWithSolicitedResponse(self):
2099
If unsolicited data is received along with solicited data in the
2100
response to a I{FETCH} command issued by L{IMAP4Client.fetchSpecific},
2101
the unsolicited data is passed to the appropriate callback and not
2102
included in the result with wihch the L{Deferred} returned by
2103
L{IMAP4Client.fetchSpecific} fires.
2105
transport = StringTransport()
2106
c = StillSimplerClient()
2107
c.makeConnection(transport)
2108
c.lineReceived('* OK [IMAP4rev1]')
2111
d = c.login('blah', 'blah')
2112
c.dataReceived('0001 OK LOGIN\r\n')
2115
d = c.select('inbox')
2116
c.lineReceived('0002 OK SELECT')
2119
d = c.fetchSpecific('1:*',
2120
headerType='HEADER.FIELDS',
2121
headerArgs=['SUBJECT'])
2122
c.dataReceived('* 1 FETCH (BODY[HEADER.FIELDS ("SUBJECT")] {38}\r\n')
2123
c.dataReceived('Subject: Suprise for your woman...\r\n')
2124
c.dataReceived('\r\n')
2125
c.dataReceived(')\r\n')
2126
c.dataReceived('* 1 FETCH (FLAGS (\Seen))\r\n')
2127
c.dataReceived('* 2 FETCH (BODY[HEADER.FIELDS ("SUBJECT")] {75}\r\n')
2128
c.dataReceived('Subject: What you been doing. Order your meds here . ,. handcuff madsen\r\n')
2129
c.dataReceived('\r\n')
2130
c.dataReceived(')\r\n')
2131
c.dataReceived('0003 OK FETCH completed\r\n')
2134
self.assertEquals(res, {
2135
1: [['BODY', ['HEADER.FIELDS', ['SUBJECT']],
2136
'Subject: Suprise for your woman...\r\n\r\n']],
2137
2: [['BODY', ['HEADER.FIELDS', ['SUBJECT']],
2138
'Subject: What you been doing. Order your meds here . ,. handcuff madsen\r\n\r\n']]
2141
self.assertEquals(c.flags, {1: ['\\Seen']})
2144
).addCallback(strip(select)
2145
).addCallback(strip(fetch)
2149
def test_literalWithoutPrecedingWhitespace(self):
2151
Literals should be recognized even when they are not preceded by
2154
transport = StringTransport()
2155
protocol = imap4.IMAP4Client()
2157
protocol.makeConnection(transport)
2158
protocol.lineReceived('* OK [IMAP4rev1]')
2161
d = protocol.login('blah', 'blah')
2162
protocol.dataReceived('0001 OK LOGIN\r\n')
2165
d = protocol.select('inbox')
2166
protocol.lineReceived('0002 OK SELECT')
2169
d = protocol.fetchSpecific('1:*',
2170
headerType='HEADER.FIELDS',
2171
headerArgs=['SUBJECT'])
2172
protocol.dataReceived(
2173
'* 1 FETCH (BODY[HEADER.FIELDS ({7}\r\nSUBJECT)] "Hello")\r\n')
2174
protocol.dataReceived('0003 OK FETCH completed\r\n')
2178
result, {1: [['BODY', ['HEADER.FIELDS', ['SUBJECT']], 'Hello']]})
2181
d.addCallback(strip(select))
2182
d.addCallback(strip(fetch))
2187
def test_nonIntegerLiteralLength(self):
2189
If the server sends a literal length which cannot be parsed as an
2190
integer, L{IMAP4Client.lineReceived} should cause the protocol to be
2191
disconnected by raising L{imap4.IllegalServerResponse}.
2193
transport = StringTransport()
2194
protocol = imap4.IMAP4Client()
2196
protocol.makeConnection(transport)
2197
protocol.lineReceived('* OK [IMAP4rev1]')
2200
d = protocol.login('blah', 'blah')
2201
protocol.dataReceived('0001 OK LOGIN\r\n')
2204
d = protocol.select('inbox')
2205
protocol.lineReceived('0002 OK SELECT')
2208
d = protocol.fetchSpecific('1:*',
2209
headerType='HEADER.FIELDS',
2210
headerArgs=['SUBJECT'])
2212
imap4.IllegalServerResponse,
2213
protocol.dataReceived,
2214
'* 1 FETCH {xyz}\r\n...')
2216
d.addCallback(strip(select))
2217
d.addCallback(strip(fetch))
2221
def test_flagsChangedInsideFetchSpecificResponse(self):
2223
Any unrequested flag information received along with other requested
2224
information in an untagged I{FETCH} received in response to a request
2225
issued with L{IMAP4Client.fetchSpecific} is passed to the
2226
C{flagsChanged} callback.
2228
transport = StringTransport()
2229
c = StillSimplerClient()
2230
c.makeConnection(transport)
2231
c.lineReceived('* OK [IMAP4rev1]')
2234
d = c.login('blah', 'blah')
2235
c.dataReceived('0001 OK LOGIN\r\n')
2238
d = c.select('inbox')
2239
c.lineReceived('0002 OK SELECT')
2242
d = c.fetchSpecific('1:*',
2243
headerType='HEADER.FIELDS',
2244
headerArgs=['SUBJECT'])
2245
# This response includes FLAGS after the requested data.
2246
c.dataReceived('* 1 FETCH (BODY[HEADER.FIELDS ("SUBJECT")] {22}\r\n')
2247
c.dataReceived('Subject: subject one\r\n')
2248
c.dataReceived(' FLAGS (\\Recent))\r\n')
2249
# And this one includes it before! Either is possible.
2250
c.dataReceived('* 2 FETCH (FLAGS (\\Seen) BODY[HEADER.FIELDS ("SUBJECT")] {22}\r\n')
2251
c.dataReceived('Subject: subject two\r\n')
2252
c.dataReceived(')\r\n')
2253
c.dataReceived('0003 OK FETCH completed\r\n')
2257
self.assertEquals(res, {
2258
1: [['BODY', ['HEADER.FIELDS', ['SUBJECT']],
2259
'Subject: subject one\r\n']],
2260
2: [['BODY', ['HEADER.FIELDS', ['SUBJECT']],
2261
'Subject: subject two\r\n']]
2264
self.assertEquals(c.flags, {1: ['\\Recent'], 2: ['\\Seen']})
2267
).addCallback(strip(select)
2268
).addCallback(strip(fetch)
2272
def test_flagsChangedInsideFetchMessageResponse(self):
2274
Any unrequested flag information received along with other requested
2275
information in an untagged I{FETCH} received in response to a request
2276
issued with L{IMAP4Client.fetchMessage} is passed to the
2277
C{flagsChanged} callback.
2279
transport = StringTransport()
2280
c = StillSimplerClient()
2281
c.makeConnection(transport)
2282
c.lineReceived('* OK [IMAP4rev1]')
2285
d = c.login('blah', 'blah')
2286
c.dataReceived('0001 OK LOGIN\r\n')
2289
d = c.select('inbox')
2290
c.lineReceived('0002 OK SELECT')
2293
d = c.fetchMessage('1:*')
2294
c.dataReceived('* 1 FETCH (RFC822 {24}\r\n')
2295
c.dataReceived('Subject: first subject\r\n')
2296
c.dataReceived(' FLAGS (\Seen))\r\n')
2297
c.dataReceived('* 2 FETCH (FLAGS (\Recent \Seen) RFC822 {25}\r\n')
2298
c.dataReceived('Subject: second subject\r\n')
2299
c.dataReceived(')\r\n')
2300
c.dataReceived('0003 OK FETCH completed\r\n')
2304
self.assertEquals(res, {
2305
1: {'RFC822': 'Subject: first subject\r\n'},
2306
2: {'RFC822': 'Subject: second subject\r\n'}})
2309
c.flags, {1: ['\\Seen'], 2: ['\\Recent', '\\Seen']})
2312
).addCallback(strip(select)
2313
).addCallback(strip(fetch)
2318
class PreauthIMAP4ClientMixin:
2320
Mixin for L{unittest.TestCase} subclasses which provides a C{setUp} method
2321
which creates an L{IMAP4Client} connected to a L{StringTransport} and puts
2322
it into the I{authenticated} state.
2324
@ivar transport: A L{StringTransport} to which C{client} is connected.
2325
@ivar client: An L{IMAP4Client} which is connected to C{transport}.
2327
clientProtocol = imap4.IMAP4Client
2331
Create an IMAP4Client connected to a fake transport and in the
2332
authenticated state.
2334
self.transport = StringTransport()
2335
self.client = self.clientProtocol()
2336
self.client.makeConnection(self.transport)
2337
self.client.dataReceived('* PREAUTH Hello unittest\r\n')
2340
def _extractDeferredResult(self, d):
2342
Synchronously extract the result of the given L{Deferred}. Fail the
2343
test if that is not possible.
2347
d.addCallbacks(result.append, error.append)
2351
error[0].raiseException()
2353
self.fail("Expected result not available")
2357
class IMAP4ClientExamineTests(PreauthIMAP4ClientMixin, unittest.TestCase):
2359
Tests for the L{IMAP4Client.examine} method.
2361
An example of usage of the EXAMINE command from RFC 3501, section 6.3.2::
2365
S: * OK [UNSEEN 8] Message 8 is first unseen
2366
S: * OK [UIDVALIDITY 3857529045] UIDs valid
2367
S: * OK [UIDNEXT 4392] Predicted next UID
2368
S: * FLAGS (\\Answered \\Flagged \\Deleted \\Seen \\Draft)
2369
S: * OK [PERMANENTFLAGS ()] No permanent flags permitted
2370
S: A932 OK [READ-ONLY] EXAMINE completed
2374
Issue an examine command, assert that the correct bytes are written to
2375
the transport, and return the L{Deferred} returned by the C{examine}
2378
d = self.client.examine('foobox')
2379
self.assertEqual(self.transport.value(), '0001 EXAMINE foobox\r\n')
2383
def _response(self, *lines):
2385
Deliver the given (unterminated) response lines to C{self.client} and
2386
then deliver a tagged EXAMINE completion line to finish the EXAMINE
2390
self.client.dataReceived(line + '\r\n')
2391
self.client.dataReceived('0001 OK [READ-ONLY] EXAMINE completed\r\n')
2394
def test_exists(self):
2396
If the server response to an I{EXAMINE} command includes an I{EXISTS}
2397
response, the L{Deferred} return by L{IMAP4Client.examine} fires with a
2398
C{dict} including the value associated with the C{'EXISTS'} key.
2401
self._response('* 3 EXISTS')
2403
self._extractDeferredResult(d),
2404
{'READ-WRITE': False, 'EXISTS': 3})
2407
def test_nonIntegerExists(self):
2409
If the server returns a non-integer EXISTS value in its response to an
2410
I{EXAMINE} command, the L{Deferred} returned by L{IMAP4Client.examine}
2411
fails with L{IllegalServerResponse}.
2414
self._response('* foo EXISTS')
2416
imap4.IllegalServerResponse, self._extractDeferredResult, d)
2419
def test_recent(self):
2421
If the server response to an I{EXAMINE} command includes an I{RECENT}
2422
response, the L{Deferred} return by L{IMAP4Client.examine} fires with a
2423
C{dict} including the value associated with the C{'RECENT'} key.
2426
self._response('* 5 RECENT')
2428
self._extractDeferredResult(d),
2429
{'READ-WRITE': False, 'RECENT': 5})
2432
def test_nonIntegerRecent(self):
2434
If the server returns a non-integer RECENT value in its response to an
2435
I{EXAMINE} command, the L{Deferred} returned by L{IMAP4Client.examine}
2436
fails with L{IllegalServerResponse}.
2439
self._response('* foo RECENT')
2441
imap4.IllegalServerResponse, self._extractDeferredResult, d)
2444
def test_unseen(self):
2446
If the server response to an I{EXAMINE} command includes an I{UNSEEN}
2447
response, the L{Deferred} returned by L{IMAP4Client.examine} fires with
2448
a C{dict} including the value associated with the C{'UNSEEN'} key.
2451
self._response('* OK [UNSEEN 8] Message 8 is first unseen')
2453
self._extractDeferredResult(d),
2454
{'READ-WRITE': False, 'UNSEEN': 8})
2457
def test_nonIntegerUnseen(self):
2459
If the server returns a non-integer UNSEEN value in its response to an
2460
I{EXAMINE} command, the L{Deferred} returned by L{IMAP4Client.examine}
2461
fails with L{IllegalServerResponse}.
2464
self._response('* OK [UNSEEN foo] Message foo is first unseen')
2466
imap4.IllegalServerResponse, self._extractDeferredResult, d)
2469
def test_uidvalidity(self):
2471
If the server response to an I{EXAMINE} command includes an
2472
I{UIDVALIDITY} response, the L{Deferred} returned by
2473
L{IMAP4Client.examine} fires with a C{dict} including the value
2474
associated with the C{'UIDVALIDITY'} key.
2477
self._response('* OK [UIDVALIDITY 12345] UIDs valid')
2479
self._extractDeferredResult(d),
2480
{'READ-WRITE': False, 'UIDVALIDITY': 12345})
2483
def test_nonIntegerUIDVALIDITY(self):
2485
If the server returns a non-integer UIDVALIDITY value in its response
2486
to an I{EXAMINE} command, the L{Deferred} returned by
2487
L{IMAP4Client.examine} fails with L{IllegalServerResponse}.
2490
self._response('* OK [UIDVALIDITY foo] UIDs valid')
2492
imap4.IllegalServerResponse, self._extractDeferredResult, d)
2495
def test_uidnext(self):
2497
If the server response to an I{EXAMINE} command includes an I{UIDNEXT}
2498
response, the L{Deferred} returned by L{IMAP4Client.examine} fires with
2499
a C{dict} including the value associated with the C{'UIDNEXT'} key.
2502
self._response('* OK [UIDNEXT 4392] Predicted next UID')
2504
self._extractDeferredResult(d),
2505
{'READ-WRITE': False, 'UIDNEXT': 4392})
2508
def test_nonIntegerUIDNEXT(self):
2510
If the server returns a non-integer UIDNEXT value in its response to an
2511
I{EXAMINE} command, the L{Deferred} returned by L{IMAP4Client.examine}
2512
fails with L{IllegalServerResponse}.
2515
self._response('* OK [UIDNEXT foo] Predicted next UID')
2517
imap4.IllegalServerResponse, self._extractDeferredResult, d)
2520
def test_flags(self):
2522
If the server response to an I{EXAMINE} command includes an I{FLAGS}
2523
response, the L{Deferred} returned by L{IMAP4Client.examine} fires with
2524
a C{dict} including the value associated with the C{'FLAGS'} key.
2528
'* FLAGS (\\Answered \\Flagged \\Deleted \\Seen \\Draft)')
2530
self._extractDeferredResult(d), {
2531
'READ-WRITE': False,
2532
'FLAGS': ('\\Answered', '\\Flagged', '\\Deleted', '\\Seen',
2536
def test_permanentflags(self):
2538
If the server response to an I{EXAMINE} command includes an I{FLAGS}
2539
response, the L{Deferred} returned by L{IMAP4Client.examine} fires with
2540
a C{dict} including the value associated with the C{'FLAGS'} key.
2544
'* OK [PERMANENTFLAGS (\\Starred)] Just one permanent flag in '
2545
'that list up there')
2547
self._extractDeferredResult(d), {
2548
'READ-WRITE': False,
2549
'PERMANENTFLAGS': ('\\Starred',)})
2552
def test_unrecognizedOk(self):
2554
If the server response to an I{EXAMINE} command includes an I{OK} with
2555
unrecognized response code text, parsing does not fail.
2559
'* OK [X-MADE-UP] I just made this response text up.')
2560
# The value won't show up in the result. It would be okay if it did
2561
# someday, perhaps. This shouldn't ever happen, though.
2563
self._extractDeferredResult(d), {'READ-WRITE': False})
2566
def test_bareOk(self):
2568
If the server response to an I{EXAMINE} command includes an I{OK} with
2569
no response code text, parsing does not fail.
2572
self._response('* OK')
2574
self._extractDeferredResult(d), {'READ-WRITE': False})
2578
class IMAP4ClientExpungeTests(PreauthIMAP4ClientMixin, unittest.TestCase):
2580
Tests for the L{IMAP4Client.expunge} method.
2582
An example of usage of the EXPUNGE command from RFC 3501, section 6.4.3::
2589
S: A202 OK EXPUNGE completed
2592
d = self.client.expunge()
2593
self.assertEquals(self.transport.value(), '0001 EXPUNGE\r\n')
2594
self.transport.clear()
2598
def _response(self, sequenceNumbers):
2599
for number in sequenceNumbers:
2600
self.client.lineReceived('* %s EXPUNGE' % (number,))
2601
self.client.lineReceived('0001 OK EXPUNGE COMPLETED')
2604
def test_expunge(self):
2606
L{IMAP4Client.expunge} sends the I{EXPUNGE} command and returns a
2607
L{Deferred} which fires with a C{list} of message sequence numbers
2608
given by the server's response.
2611
self._response([3, 3, 5, 8])
2612
self.assertEquals(self._extractDeferredResult(d), [3, 3, 5, 8])
2615
def test_nonIntegerExpunged(self):
2617
If the server responds with a non-integer where a message sequence
2618
number is expected, the L{Deferred} returned by L{IMAP4Client.expunge}
2619
fails with L{IllegalServerResponse}.
2622
self._response([3, 3, 'foo', 8])
2624
imap4.IllegalServerResponse, self._extractDeferredResult, d)
2628
class IMAP4ClientSearchTests(PreauthIMAP4ClientMixin, unittest.TestCase):
2630
Tests for the L{IMAP4Client.search} method.
2632
An example of usage of the SEARCH command from RFC 3501, section 6.4.4::
2634
C: A282 SEARCH FLAGGED SINCE 1-Feb-1994 NOT FROM "Smith"
2635
S: * SEARCH 2 84 882
2636
S: A282 OK SEARCH completed
2637
C: A283 SEARCH TEXT "string not in mailbox"
2639
S: A283 OK SEARCH completed
2640
C: A284 SEARCH CHARSET UTF-8 TEXT {6}
2643
S: A284 OK SEARCH completed
2646
d = self.client.search(imap4.Query(text="ABCDEF"))
2648
self.transport.value(), '0001 SEARCH (TEXT "ABCDEF")\r\n')
2652
def _response(self, messageNumbers):
2653
self.client.lineReceived(
2654
"* SEARCH " + " ".join(map(str, messageNumbers)))
2655
self.client.lineReceived("0001 OK SEARCH completed")
2658
def test_search(self):
2660
L{IMAP4Client.search} sends the I{SEARCH} command and returns a
2661
L{Deferred} which fires with a C{list} of message sequence numbers
2662
given by the server's response.
2665
self._response([2, 5, 10])
2666
self.assertEquals(self._extractDeferredResult(d), [2, 5, 10])
2669
def test_nonIntegerFound(self):
2671
If the server responds with a non-integer where a message sequence
2672
number is expected, the L{Deferred} returned by L{IMAP4Client.search}
2673
fails with L{IllegalServerResponse}.
2676
self._response([2, "foo", 10])
2678
imap4.IllegalServerResponse, self._extractDeferredResult, d)
2682
class IMAP4ClientFetchTests(PreauthIMAP4ClientMixin, unittest.TestCase):
2684
Tests for the L{IMAP4Client.fetch} method.
2686
See RFC 3501, section 6.4.5.
2688
def test_fetchUID(self):
2690
L{IMAP4Client.fetchUID} sends the I{FETCH UID} command and returns a
2691
L{Deferred} which fires with a C{dict} mapping message sequence numbers
2692
to C{dict}s mapping C{'UID'} to that message's I{UID} in the server's
2695
d = self.client.fetchUID('1:7')
2696
self.assertEquals(self.transport.value(), '0001 FETCH 1:7 (UID)\r\n')
2697
self.client.lineReceived('* 2 FETCH (UID 22)')
2698
self.client.lineReceived('* 3 FETCH (UID 23)')
2699
self.client.lineReceived('* 4 FETCH (UID 24)')
2700
self.client.lineReceived('* 5 FETCH (UID 25)')
2701
self.client.lineReceived('0001 OK FETCH completed')
2703
self._extractDeferredResult(d), {
2710
def test_fetchUIDNonIntegerFound(self):
2712
If the server responds with a non-integer where a message sequence
2713
number is expected, the L{Deferred} returned by L{IMAP4Client.fetchUID}
2714
fails with L{IllegalServerResponse}.
2716
d = self.client.fetchUID('1')
2717
self.assertEquals(self.transport.value(), '0001 FETCH 1 (UID)\r\n')
2718
self.client.lineReceived('* foo FETCH (UID 22)')
2719
self.client.lineReceived('0001 OK FETCH completed')
2721
imap4.IllegalServerResponse, self._extractDeferredResult, d)
2724
def test_incompleteFetchUIDResponse(self):
2726
If the server responds with an incomplete I{FETCH} response line, the
2727
L{Deferred} returned by L{IMAP4Client.fetchUID} fails with
2728
L{IllegalServerResponse}.
2730
d = self.client.fetchUID('1:7')
2731
self.assertEquals(self.transport.value(), '0001 FETCH 1:7 (UID)\r\n')
2732
self.client.lineReceived('* 2 FETCH (UID 22)')
2733
self.client.lineReceived('* 3 FETCH (UID)')
2734
self.client.lineReceived('* 4 FETCH (UID 24)')
2735
self.client.lineReceived('0001 OK FETCH completed')
2737
imap4.IllegalServerResponse, self._extractDeferredResult, d)
2740
def test_fetchBody(self):
2742
L{IMAP4Client.fetchBody} sends the I{FETCH BODY} command and returns a
2743
L{Deferred} which fires with a C{dict} mapping message sequence numbers
2744
to C{dict}s mapping C{'RFC822.TEXT'} to that message's body as given in
2745
the server's response.
2747
d = self.client.fetchBody('3')
2749
self.transport.value(), '0001 FETCH 3 (RFC822.TEXT)\r\n')
2750
self.client.lineReceived('* 3 FETCH (RFC822.TEXT "Message text")')
2751
self.client.lineReceived('0001 OK FETCH completed')
2753
self._extractDeferredResult(d),
2754
{3: {'RFC822.TEXT': 'Message text'}})
2757
def test_fetchSpecific(self):
2759
L{IMAP4Client.fetchSpecific} sends the I{BODY[]} command if no
2760
parameters beyond the message set to retrieve are given. It returns a
2761
L{Deferred} which fires with a C{dict} mapping message sequence numbers
2762
to C{list}s of corresponding message data given by the server's
2765
d = self.client.fetchSpecific('7')
2767
self.transport.value(), '0001 FETCH 7 BODY[]\r\n')
2768
self.client.lineReceived('* 7 FETCH (BODY[] "Some body")')
2769
self.client.lineReceived('0001 OK FETCH completed')
2771
self._extractDeferredResult(d), {7: [['BODY', [], "Some body"]]})
2774
def test_fetchSpecificPeek(self):
2776
L{IMAP4Client.fetchSpecific} issues a I{BODY.PEEK[]} command if passed
2777
C{True} for the C{peek} parameter.
2779
d = self.client.fetchSpecific('6', peek=True)
2781
self.transport.value(), '0001 FETCH 6 BODY.PEEK[]\r\n')
2782
# BODY.PEEK responses are just BODY
2783
self.client.lineReceived('* 6 FETCH (BODY[] "Some body")')
2784
self.client.lineReceived('0001 OK FETCH completed')
2786
self._extractDeferredResult(d), {6: [['BODY', [], "Some body"]]})
2789
def test_fetchSpecificNumbered(self):
2791
L{IMAP4Client.fetchSpecific}, when passed a sequence for for
2792
C{headerNumber}, sends the I{BODY[N.M]} command. It returns a
2793
L{Deferred} which fires with a C{dict} mapping message sequence numbers
2794
to C{list}s of corresponding message data given by the server's
2797
d = self.client.fetchSpecific('7', headerNumber=(1, 2, 3))
2799
self.transport.value(), '0001 FETCH 7 BODY[1.2.3]\r\n')
2800
self.client.lineReceived('* 7 FETCH (BODY[1.2.3] "Some body")')
2801
self.client.lineReceived('0001 OK FETCH completed')
2803
self._extractDeferredResult(d),
2804
{7: [['BODY', ['1.2.3'], "Some body"]]})
2807
def test_fetchSpecificText(self):
2809
L{IMAP4Client.fetchSpecific}, when passed C{'TEXT'} for C{headerType},
2810
sends the I{BODY[TEXT]} command. It returns a L{Deferred} which fires
2811
with a C{dict} mapping message sequence numbers to C{list}s of
2812
corresponding message data given by the server's response.
2814
d = self.client.fetchSpecific('8', headerType='TEXT')
2816
self.transport.value(), '0001 FETCH 8 BODY[TEXT]\r\n')
2817
self.client.lineReceived('* 8 FETCH (BODY[TEXT] "Some body")')
2818
self.client.lineReceived('0001 OK FETCH completed')
2820
self._extractDeferredResult(d),
2821
{8: [['BODY', ['TEXT'], "Some body"]]})
2824
def test_fetchSpecificNumberedText(self):
2826
If passed a value for the C{headerNumber} parameter and C{'TEXT'} for
2827
the C{headerType} parameter, L{IMAP4Client.fetchSpecific} sends a
2828
I{BODY[number.TEXT]} request and returns a L{Deferred} which fires with
2829
a C{dict} mapping message sequence numbers to C{list}s of message data
2830
given by the server's response.
2832
d = self.client.fetchSpecific('4', headerType='TEXT', headerNumber=7)
2834
self.transport.value(), '0001 FETCH 4 BODY[7.TEXT]\r\n')
2835
self.client.lineReceived('* 4 FETCH (BODY[7.TEXT] "Some body")')
2836
self.client.lineReceived('0001 OK FETCH completed')
2838
self._extractDeferredResult(d),
2839
{4: [['BODY', ['7.TEXT'], "Some body"]]})
2842
def test_incompleteFetchSpecificTextResponse(self):
2844
If the server responds to a I{BODY[TEXT]} request with a I{FETCH} line
2845
which is truncated after the I{BODY[TEXT]} tokens, the L{Deferred}
2846
returned by L{IMAP4Client.fetchUID} fails with
2847
L{IllegalServerResponse}.
2849
d = self.client.fetchSpecific('8', headerType='TEXT')
2851
self.transport.value(), '0001 FETCH 8 BODY[TEXT]\r\n')
2852
self.client.lineReceived('* 8 FETCH (BODY[TEXT])')
2853
self.client.lineReceived('0001 OK FETCH completed')
2855
imap4.IllegalServerResponse, self._extractDeferredResult, d)
2858
def test_fetchSpecificMIME(self):
2860
L{IMAP4Client.fetchSpecific}, when passed C{'MIME'} for C{headerType},
2861
sends the I{BODY[MIME]} command. It returns a L{Deferred} which fires
2862
with a C{dict} mapping message sequence numbers to C{list}s of
2863
corresponding message data given by the server's response.
2865
d = self.client.fetchSpecific('8', headerType='MIME')
2867
self.transport.value(), '0001 FETCH 8 BODY[MIME]\r\n')
2868
self.client.lineReceived('* 8 FETCH (BODY[MIME] "Some body")')
2869
self.client.lineReceived('0001 OK FETCH completed')
2871
self._extractDeferredResult(d),
2872
{8: [['BODY', ['MIME'], "Some body"]]})
2875
def test_fetchSpecificPartial(self):
2877
L{IMAP4Client.fetchSpecific}, when passed C{offset} and C{length},
2878
sends a partial content request (like I{BODY[TEXT]<offset.length>}).
2879
It returns a L{Deferred} which fires with a C{dict} mapping message
2880
sequence numbers to C{list}s of corresponding message data given by the
2883
d = self.client.fetchSpecific(
2884
'9', headerType='TEXT', offset=17, length=3)
2886
self.transport.value(), '0001 FETCH 9 BODY[TEXT]<17.3>\r\n')
2887
self.client.lineReceived('* 9 FETCH (BODY[TEXT]<17> "foo")')
2888
self.client.lineReceived('0001 OK FETCH completed')
2890
self._extractDeferredResult(d),
2891
{9: [['BODY', ['TEXT'], '<17>', 'foo']]})
2894
def test_incompleteFetchSpecificPartialResponse(self):
2896
If the server responds to a I{BODY[TEXT]} request with a I{FETCH} line
2897
which is truncated after the I{BODY[TEXT]<offset>} tokens, the
2898
L{Deferred} returned by L{IMAP4Client.fetchUID} fails with
2899
L{IllegalServerResponse}.
2901
d = self.client.fetchSpecific('8', headerType='TEXT')
2903
self.transport.value(), '0001 FETCH 8 BODY[TEXT]\r\n')
2904
self.client.lineReceived('* 8 FETCH (BODY[TEXT]<17>)')
2905
self.client.lineReceived('0001 OK FETCH completed')
2907
imap4.IllegalServerResponse, self._extractDeferredResult, d)
2910
def test_fetchSpecificHTML(self):
2912
If the body of a message begins with I{<} and ends with I{>} (as,
2913
for example, HTML bodies typically will), this is still interpreted
2914
as the body by L{IMAP4Client.fetchSpecific} (and particularly, not
2915
as a length indicator for a response to a request for a partial
2918
d = self.client.fetchSpecific('7')
2920
self.transport.value(), '0001 FETCH 7 BODY[]\r\n')
2921
self.client.lineReceived('* 7 FETCH (BODY[] "<html>test</html>")')
2922
self.client.lineReceived('0001 OK FETCH completed')
2924
self._extractDeferredResult(d), {7: [['BODY', [], "<html>test</html>"]]})
2928
class IMAP4ClientStoreTests(PreauthIMAP4ClientMixin, unittest.TestCase):
2930
Tests for the L{IMAP4Client.setFlags}, L{IMAP4Client.addFlags}, and
2931
L{IMAP4Client.removeFlags} methods.
2933
An example of usage of the STORE command, in terms of which these three
2934
methods are implemented, from RFC 3501, section 6.4.6::
2936
C: A003 STORE 2:4 +FLAGS (\Deleted)
2937
S: * 2 FETCH (FLAGS (\Deleted \Seen))
2938
S: * 3 FETCH (FLAGS (\Deleted))
2939
S: * 4 FETCH (FLAGS (\Deleted \Flagged \Seen))
2940
S: A003 OK STORE completed
2942
clientProtocol = StillSimplerClient
2944
def _flagsTest(self, method, item):
2946
Test a non-silent flag modifying method. Call the method, assert that
2947
the correct bytes are sent, deliver a I{FETCH} response, and assert
2948
that the result of the Deferred returned by the method is correct.
2950
@param method: The name of the method to test.
2951
@param item: The data item which is expected to be specified.
2953
d = getattr(self.client, method)('3', ('\\Read', '\\Seen'), False)
2955
self.transport.value(),
2956
'0001 STORE 3 ' + item + ' (\\Read \\Seen)\r\n')
2957
self.client.lineReceived('* 3 FETCH (FLAGS (\\Read \\Seen))')
2958
self.client.lineReceived('0001 OK STORE completed')
2960
self._extractDeferredResult(d),
2961
{3: {'FLAGS': ['\\Read', '\\Seen']}})
2964
def _flagsSilentlyTest(self, method, item):
2966
Test a silent flag modifying method. Call the method, assert that the
2967
correct bytes are sent, deliver an I{OK} response, and assert that the
2968
result of the Deferred returned by the method is correct.
2970
@param method: The name of the method to test.
2971
@param item: The data item which is expected to be specified.
2973
d = getattr(self.client, method)('3', ('\\Read', '\\Seen'), True)
2975
self.transport.value(),
2976
'0001 STORE 3 ' + item + ' (\\Read \\Seen)\r\n')
2977
self.client.lineReceived('0001 OK STORE completed')
2978
self.assertEquals(self._extractDeferredResult(d), {})
2981
def _flagsSilentlyWithUnsolicitedDataTest(self, method, item):
2983
Test unsolicited data received in response to a silent flag modifying
2984
method. Call the method, assert that the correct bytes are sent,
2985
deliver the unsolicited I{FETCH} response, and assert that the result
2986
of the Deferred returned by the method is correct.
2988
@param method: The name of the method to test.
2989
@param item: The data item which is expected to be specified.
2991
d = getattr(self.client, method)('3', ('\\Read', '\\Seen'), True)
2993
self.transport.value(),
2994
'0001 STORE 3 ' + item + ' (\\Read \\Seen)\r\n')
2995
self.client.lineReceived('* 2 FETCH (FLAGS (\\Read \\Seen))')
2996
self.client.lineReceived('0001 OK STORE completed')
2997
self.assertEquals(self._extractDeferredResult(d), {})
2998
self.assertEquals(self.client.flags, {2: ['\\Read', '\\Seen']})
3001
def test_setFlags(self):
3003
When passed a C{False} value for the C{silent} parameter,
3004
L{IMAP4Client.setFlags} sends the I{STORE} command with a I{FLAGS} data
3005
item and returns a L{Deferred} which fires with a C{dict} mapping
3006
message sequence numbers to C{dict}s mapping C{'FLAGS'} to the new
3007
flags of those messages.
3009
self._flagsTest('setFlags', 'FLAGS')
3012
def test_setFlagsSilently(self):
3014
When passed a C{True} value for the C{silent} parameter,
3015
L{IMAP4Client.setFlags} sends the I{STORE} command with a
3016
I{FLAGS.SILENT} data item and returns a L{Deferred} which fires with an
3019
self._flagsSilentlyTest('setFlags', 'FLAGS.SILENT')
3022
def test_setFlagsSilentlyWithUnsolicitedData(self):
3024
If unsolicited flag data is received in response to a I{STORE}
3025
I{FLAGS.SILENT} request, that data is passed to the C{flagsChanged}
3028
self._flagsSilentlyWithUnsolicitedDataTest('setFlags', 'FLAGS.SILENT')
3031
def test_addFlags(self):
3033
L{IMAP4Client.addFlags} is like L{IMAP4Client.setFlags}, but sends
3034
I{+FLAGS} instead of I{FLAGS}.
3036
self._flagsTest('addFlags', '+FLAGS')
3039
def test_addFlagsSilently(self):
3041
L{IMAP4Client.addFlags} with a C{True} value for C{silent} behaves like
3042
L{IMAP4Client.setFlags} with a C{True} value for C{silent}, but it
3043
sends I{+FLAGS.SILENT} instead of I{FLAGS.SILENT}.
3045
self._flagsSilentlyTest('addFlags', '+FLAGS.SILENT')
3048
def test_addFlagsSilentlyWithUnsolicitedData(self):
3050
L{IMAP4Client.addFlags} behaves like L{IMAP4Client.setFlags} when used
3051
in silent mode and unsolicited data is received.
3053
self._flagsSilentlyWithUnsolicitedDataTest('addFlags', '+FLAGS.SILENT')
3056
def test_removeFlags(self):
3058
L{IMAP4Client.removeFlags} is like L{IMAP4Client.setFlags}, but sends
3059
I{-FLAGS} instead of I{FLAGS}.
3061
self._flagsTest('removeFlags', '-FLAGS')
3064
def test_removeFlagsSilently(self):
3066
L{IMAP4Client.removeFlags} with a C{True} value for C{silent} behaves
3067
like L{IMAP4Client.setFlags} with a C{True} value for C{silent}, but it
3068
sends I{-FLAGS.SILENT} instead of I{FLAGS.SILENT}.
3070
self._flagsSilentlyTest('removeFlags', '-FLAGS.SILENT')
3073
def test_removeFlagsSilentlyWithUnsolicitedData(self):
3075
L{IMAP4Client.removeFlags} behaves like L{IMAP4Client.setFlags} when
3076
used in silent mode and unsolicited data is received.
3078
self._flagsSilentlyWithUnsolicitedDataTest('removeFlags', '-FLAGS.SILENT')
3082
class FakeyServer(imap4.IMAP4Server):
3086
def sendServerGreeting(self):
3090
implements(imap4.IMessage)
3092
def __init__(self, headers, flags, date, body, uid, subpart):
3093
self.headers = headers
3095
self.body = StringIO(body)
3096
self.size = len(body)
3099
self.subpart = subpart
3101
def getHeaders(self, negate, *names):
3102
self.got_headers = negate, names
3108
def getInternalDate(self):
3111
def getBodyFile(self):
3120
def isMultipart(self):
3121
return self.subpart is not None
3123
def getSubPart(self, part):
3124
self.got_subpart = part
3125
return self.subpart[part]
3127
class NewStoreTestCase(unittest.TestCase, IMAP4HelperMixin):
3132
self.received_messages = self.received_uid = None
3134
self.server = imap4.IMAP4Server()
3135
self.server.state = 'select'
3136
self.server.mbox = self
3137
self.connected = defer.Deferred()
3138
self.client = SimpleClient(self.connected)
3140
def addListener(self, x):
3142
def removeListener(self, x):
3145
def store(self, *args, **kw):
3146
self.storeArgs = args, kw
3147
return self.response
3149
def _storeWork(self):
3151
return self.function(self.messages, self.flags, self.silent, self.uid)
3155
self.connected.addCallback(strip(connected)
3156
).addCallback(result
3157
).addCallback(self._cbStopClient
3158
).addErrback(self._ebGeneral)
3161
self.assertEquals(self.result, self.expected)
3162
self.assertEquals(self.storeArgs, self.expectedArgs)
3163
d = loopback.loopbackTCP(self.server, self.client, noisy=False)
3164
d.addCallback(check)
3167
def testSetFlags(self, uid=0):
3168
self.function = self.client.setFlags
3169
self.messages = '1,5,9'
3170
self.flags = ['\\A', '\\B', 'C']
3174
1: ['\\A', '\\B', 'C'],
3175
5: ['\\A', '\\B', 'C'],
3176
9: ['\\A', '\\B', 'C'],
3179
1: {'FLAGS': ['\\A', '\\B', 'C']},
3180
5: {'FLAGS': ['\\A', '\\B', 'C']},
3181
9: {'FLAGS': ['\\A', '\\B', 'C']},
3183
msg = imap4.MessageSet()
3187
self.expectedArgs = ((msg, ['\\A', '\\B', 'C'], 0), {'uid': 0})
3188
return self._storeWork()
3191
class NewFetchTestCase(unittest.TestCase, IMAP4HelperMixin):
3193
self.received_messages = self.received_uid = None
3196
self.server = imap4.IMAP4Server()
3197
self.server.state = 'select'
3198
self.server.mbox = self
3199
self.connected = defer.Deferred()
3200
self.client = SimpleClient(self.connected)
3202
def addListener(self, x):
3204
def removeListener(self, x):
3207
def fetch(self, messages, uid):
3208
self.received_messages = messages
3209
self.received_uid = uid
3210
return iter(zip(range(len(self.msgObjs)), self.msgObjs))
3212
def _fetchWork(self, uid):
3214
for (i, msg) in zip(range(len(self.msgObjs)), self.msgObjs):
3215
self.expected[i]['UID'] = str(msg.getUID())
3220
self.connected.addCallback(lambda _: self.function(self.messages, uid)
3221
).addCallback(result
3222
).addCallback(self._cbStopClient
3223
).addErrback(self._ebGeneral)
3225
d = loopback.loopbackTCP(self.server, self.client, noisy=False)
3226
d.addCallback(lambda x : self.assertEquals(self.result, self.expected))
3229
def testFetchUID(self):
3230
self.function = lambda m, u: self.client.fetchUID(m)
3234
FakeyMessage({}, (), '', '', 12345, None),
3235
FakeyMessage({}, (), '', '', 999, None),
3236
FakeyMessage({}, (), '', '', 10101, None),
3239
0: {'UID': '12345'},
3241
2: {'UID': '10101'},
3243
return self._fetchWork(0)
3245
def testFetchFlags(self, uid=0):
3246
self.function = self.client.fetchFlags
3249
FakeyMessage({}, ['FlagA', 'FlagB', '\\FlagC'], '', '', 54321, None),
3250
FakeyMessage({}, ['\\FlagC', 'FlagA', 'FlagB'], '', '', 12345, None),
3253
0: {'FLAGS': ['FlagA', 'FlagB', '\\FlagC']},
3254
1: {'FLAGS': ['\\FlagC', 'FlagA', 'FlagB']},
3256
return self._fetchWork(uid)
3258
def testFetchFlagsUID(self):
3259
return self.testFetchFlags(1)
3261
def testFetchInternalDate(self, uid=0):
3262
self.function = self.client.fetchInternalDate
3263
self.messages = '13'
3265
FakeyMessage({}, (), 'Fri, 02 Nov 2003 21:25:10 GMT', '', 23232, None),
3266
FakeyMessage({}, (), 'Thu, 29 Dec 2013 11:31:52 EST', '', 101, None),
3267
FakeyMessage({}, (), 'Mon, 10 Mar 1992 02:44:30 CST', '', 202, None),
3268
FakeyMessage({}, (), 'Sat, 11 Jan 2000 14:40:24 PST', '', 303, None),
3271
0: {'INTERNALDATE': '02-Nov-2003 21:25:10 +0000'},
3272
1: {'INTERNALDATE': '29-Dec-2013 11:31:52 -0500'},
3273
2: {'INTERNALDATE': '10-Mar-1992 02:44:30 -0600'},
3274
3: {'INTERNALDATE': '11-Jan-2000 14:40:24 -0800'},
3276
return self._fetchWork(uid)
3278
def testFetchInternalDateUID(self):
3279
return self.testFetchInternalDate(1)
3281
def testFetchEnvelope(self, uid=0):
3282
self.function = self.client.fetchEnvelope
3283
self.messages = '15'
3286
'from': 'user@domain', 'to': 'resu@domain',
3287
'date': 'thursday', 'subject': 'it is a message',
3288
'message-id': 'id-id-id-yayaya'}, (), '', '', 65656,
3293
['thursday', 'it is a message',
3294
[[None, None, 'user', 'domain']],
3295
[[None, None, 'user', 'domain']],
3296
[[None, None, 'user', 'domain']],
3297
[[None, None, 'resu', 'domain']],
3298
None, None, None, 'id-id-id-yayaya']
3301
return self._fetchWork(uid)
3303
def testFetchEnvelopeUID(self):
3304
return self.testFetchEnvelope(1)
3306
def testFetchBodyStructure(self, uid=0):
3307
self.function = self.client.fetchBodyStructure
3308
self.messages = '3:9,10:*'
3309
self.msgObjs = [FakeyMessage({
3310
'content-type': 'text/plain; name=thing; key="value"',
3311
'content-id': 'this-is-the-content-id',
3312
'content-description': 'describing-the-content-goes-here!',
3313
'content-transfer-encoding': '8BIT',
3314
}, (), '', 'Body\nText\nGoes\nHere\n', 919293, None)]
3315
self.expected = {0: {'BODYSTRUCTURE': [
3316
'text', 'plain', [['name', 'thing'], ['key', 'value']],
3317
'this-is-the-content-id', 'describing-the-content-goes-here!',
3318
'8BIT', '20', '4', None, None, None]}}
3319
return self._fetchWork(uid)
3321
def testFetchBodyStructureUID(self):
3322
return self.testFetchBodyStructure(1)
3324
def testFetchSimplifiedBody(self, uid=0):
3325
self.function = self.client.fetchSimplifiedBody
3326
self.messages = '21'
3327
self.msgObjs = [FakeyMessage({}, (), '', 'Yea whatever', 91825,
3328
[FakeyMessage({'content-type': 'image/jpg'}, (), '',
3329
'Body Body Body', None, None
3334
[None, None, [], None, None, None,
3340
return self._fetchWork(uid)
3342
def testFetchSimplifiedBodyUID(self):
3343
return self.testFetchSimplifiedBody(1)
3345
def testFetchSimplifiedBodyText(self, uid=0):
3346
self.function = self.client.fetchSimplifiedBody
3347
self.messages = '21'
3348
self.msgObjs = [FakeyMessage({'content-type': 'text/plain'},
3349
(), '', 'Yea whatever', 91825, None)]
3352
['text', 'plain', [], None, None, None,
3358
return self._fetchWork(uid)
3360
def testFetchSimplifiedBodyTextUID(self):
3361
return self.testFetchSimplifiedBodyText(1)
3363
def testFetchSimplifiedBodyRFC822(self, uid=0):
3364
self.function = self.client.fetchSimplifiedBody
3365
self.messages = '21'
3366
self.msgObjs = [FakeyMessage({'content-type': 'message/rfc822'},
3367
(), '', 'Yea whatever', 91825,
3368
[FakeyMessage({'content-type': 'image/jpg'}, (), '',
3369
'Body Body Body', None, None
3374
['message', 'rfc822', [], None, None, None,
3375
'12', [None, None, [[None, None, None]],
3376
[[None, None, None]], None, None, None,
3377
None, None, None], ['image', 'jpg', [],
3378
None, None, None, '14'], '1'
3383
return self._fetchWork(uid)
3385
def testFetchSimplifiedBodyRFC822UID(self):
3386
return self.testFetchSimplifiedBodyRFC822(1)
3388
def testFetchMessage(self, uid=0):
3389
self.function = self.client.fetchMessage
3390
self.messages = '1,3,7,10101'
3392
FakeyMessage({'Header': 'Value'}, (), '', 'BODY TEXT\r\n', 91, None),
3395
0: {'RFC822': 'Header: Value\r\n\r\nBODY TEXT\r\n'}
3397
return self._fetchWork(uid)
3399
def testFetchMessageUID(self):
3400
return self.testFetchMessage(1)
3402
def testFetchHeaders(self, uid=0):
3403
self.function = self.client.fetchHeaders
3404
self.messages = '9,6,2'
3406
FakeyMessage({'H1': 'V1', 'H2': 'V2'}, (), '', '', 99, None),
3409
0: {'RFC822.HEADER': imap4._formatHeaders({'H1': 'V1', 'H2': 'V2'})},
3411
return self._fetchWork(uid)
3413
def testFetchHeadersUID(self):
3414
return self.testFetchHeaders(1)
3416
def testFetchBody(self, uid=0):
3417
self.function = self.client.fetchBody
3418
self.messages = '1,2,3,4,5,6,7'
3420
FakeyMessage({'Header': 'Value'}, (), '', 'Body goes here\r\n', 171, None),
3423
0: {'RFC822.TEXT': 'Body goes here\r\n'},
3425
return self._fetchWork(uid)
3427
def testFetchBodyUID(self):
3428
return self.testFetchBody(1)
3430
def testFetchBodyParts(self):
3432
Test the server's handling of requests for specific body sections.
3434
self.function = self.client.fetchSpecific
3437
innerBody1 = 'Contained body message text. Squarge.'
3438
innerBody2 = 'Secondary <i>message</i> text of squarge body.'
3439
headers = util.OrderedDict()
3440
headers['from'] = 'sender@host'
3441
headers['to'] = 'recipient@domain'
3442
headers['subject'] = 'booga booga boo'
3443
headers['content-type'] = 'multipart/alternative; boundary="xyz"'
3444
innerHeaders = util.OrderedDict()
3445
innerHeaders['subject'] = 'this is subject text'
3446
innerHeaders['content-type'] = 'text/plain'
3447
innerHeaders2 = util.OrderedDict()
3448
innerHeaders2['subject'] = '<b>this is subject</b>'
3449
innerHeaders2['content-type'] = 'text/html'
3450
self.msgObjs = [FakeyMessage(
3451
headers, (), None, outerBody, 123,
3452
[FakeyMessage(innerHeaders, (), None, innerBody1, None, None),
3453
FakeyMessage(innerHeaders2, (), None, innerBody2, None, None)])]
3455
0: [['BODY', ['1'], 'Contained body message text. Squarge.']]}
3460
self.connected.addCallback(
3461
lambda _: self.function(self.messages, headerNumber=1))
3462
self.connected.addCallback(result)
3463
self.connected.addCallback(self._cbStopClient)
3464
self.connected.addErrback(self._ebGeneral)
3466
d = loopback.loopbackTCP(self.server, self.client, noisy=False)
3467
d.addCallback(lambda ign: self.assertEquals(self.result, self.expected))
3471
def test_fetchBodyPartOfNonMultipart(self):
3473
Single-part messages have an implicit first part which clients
3474
should be able to retrieve explicitly. Test that a client
3475
requesting part 1 of a text/plain message receives the body of the
3478
self.function = self.client.fetchSpecific
3481
outerBody = 'DA body'
3482
headers = util.OrderedDict()
3483
headers['from'] = 'sender@host'
3484
headers['to'] = 'recipient@domain'
3485
headers['subject'] = 'booga booga boo'
3486
headers['content-type'] = 'text/plain'
3487
self.msgObjs = [FakeyMessage(
3488
headers, (), None, outerBody, 123, None)]
3490
self.expected = {0: [['BODY', ['1'], 'DA body']]}
3495
self.connected.addCallback(
3496
lambda _: self.function(self.messages, headerNumber=parts))
3497
self.connected.addCallback(result)
3498
self.connected.addCallback(self._cbStopClient)
3499
self.connected.addErrback(self._ebGeneral)
3501
d = loopback.loopbackTCP(self.server, self.client, noisy=False)
3502
d.addCallback(lambda ign: self.assertEquals(self.result, self.expected))
3506
def testFetchSize(self, uid=0):
3507
self.function = self.client.fetchSize
3508
self.messages = '1:100,2:*'
3510
FakeyMessage({}, (), '', 'x' * 20, 123, None),
3513
0: {'RFC822.SIZE': '20'},
3515
return self._fetchWork(uid)
3517
def testFetchSizeUID(self):
3518
return self.testFetchSize(1)
3520
def testFetchFull(self, uid=0):
3521
self.function = self.client.fetchFull
3522
self.messages = '1,3'
3524
FakeyMessage({}, ('\\XYZ', '\\YZX', 'Abc'),
3525
'Sun, 25 Jul 2010 06:20:30 -0400 (EDT)',
3526
'xyz' * 2, 654, None),
3527
FakeyMessage({}, ('\\One', '\\Two', 'Three'),
3528
'Mon, 14 Apr 2003 19:43:44 -0400',
3529
'abc' * 4, 555, None),
3532
0: {'FLAGS': ['\\XYZ', '\\YZX', 'Abc'],
3533
'INTERNALDATE': '25-Jul-2010 06:20:30 -0400',
3535
'ENVELOPE': [None, None, [[None, None, None]], [[None, None, None]], None, None, None, None, None, None],
3536
'BODY': [None, None, [], None, None, None, '6']},
3537
1: {'FLAGS': ['\\One', '\\Two', 'Three'],
3538
'INTERNALDATE': '14-Apr-2003 19:43:44 -0400',
3539
'RFC822.SIZE': '12',
3540
'ENVELOPE': [None, None, [[None, None, None]], [[None, None, None]], None, None, None, None, None, None],
3541
'BODY': [None, None, [], None, None, None, '12']},
3543
return self._fetchWork(uid)
3545
def testFetchFullUID(self):
3546
return self.testFetchFull(1)
3548
def testFetchAll(self, uid=0):
3549
self.function = self.client.fetchAll
3550
self.messages = '1,2:3'
3552
FakeyMessage({}, (), 'Mon, 14 Apr 2003 19:43:44 +0400',
3553
'Lalala', 10101, None),
3554
FakeyMessage({}, (), 'Tue, 15 Apr 2003 19:43:44 +0200',
3555
'Alalal', 20202, None),
3558
0: {'ENVELOPE': [None, None, [[None, None, None]], [[None, None, None]], None, None, None, None, None, None],
3560
'INTERNALDATE': '14-Apr-2003 19:43:44 +0400',
3562
1: {'ENVELOPE': [None, None, [[None, None, None]], [[None, None, None]], None, None, None, None, None, None],
3564
'INTERNALDATE': '15-Apr-2003 19:43:44 +0200',
3567
return self._fetchWork(uid)
3569
def testFetchAllUID(self):
3570
return self.testFetchAll(1)
3572
def testFetchFast(self, uid=0):
3573
self.function = self.client.fetchFast
3576
FakeyMessage({}, ('\\X',), '19 Mar 2003 19:22:21 -0500', '', 9, None),
3579
0: {'FLAGS': ['\\X'],
3580
'INTERNALDATE': '19-Mar-2003 19:22:21 -0500',
3581
'RFC822.SIZE': '0'},
3583
return self._fetchWork(uid)
3585
def testFetchFastUID(self):
3586
return self.testFetchFast(1)
3590
class DefaultSearchTestCase(IMAP4HelperMixin, unittest.TestCase):
3592
Test the behavior of the server's SEARCH implementation, particularly in
3593
the face of unhandled search terms.
3596
self.server = imap4.IMAP4Server()
3597
self.server.state = 'select'
3598
self.server.mbox = self
3599
self.connected = defer.Deferred()
3600
self.client = SimpleClient(self.connected)
3602
FakeyMessage({}, (), '', '', 999, None),
3603
FakeyMessage({}, (), '', '', 10101, None),
3604
FakeyMessage({}, (), '', '', 12345, None),
3608
def fetch(self, messages, uid):
3610
Pretend to be a mailbox and let C{self.server} lookup messages on me.
3612
return zip(range(1, len(self.msgObjs) + 1), self.msgObjs)
3615
def _messageSetSearchTest(self, queryTerms, expectedMessages):
3617
Issue a search with given query and verify that the returned messages
3618
match the given expected messages.
3620
@param queryTerms: A string giving the search query.
3621
@param expectedMessages: A list of the message sequence numbers
3622
expected as the result of the search.
3623
@return: A L{Deferred} which fires when the test is complete.
3626
return self.client.search(queryTerms)
3628
d = self.connected.addCallback(strip(search))
3629
def searched(results):
3630
self.assertEquals(results, expectedMessages)
3631
d.addCallback(searched)
3632
d.addCallback(self._cbStopClient)
3633
d.addErrback(self._ebGeneral)
3638
def test_searchMessageSet(self):
3640
Test that a search which starts with a message set properly limits
3641
the search results to messages in that set.
3643
return self._messageSetSearchTest('1', [1])
3646
def test_searchMessageSetWithStar(self):
3648
If the search filter ends with a star, all the message from the
3649
starting point are returned.
3651
return self._messageSetSearchTest('2:*', [2, 3])
3654
def test_searchMessageSetWithList(self):
3656
If the search filter contains nesting terms, one of which includes a
3657
message sequence set with a wildcard, IT ALL WORKS GOOD.
3659
# 5 is bigger than the biggest message sequence number, but that's
3660
# okay, because N:* includes the biggest message sequence number even
3661
# if N is bigger than that (read the rfc nub).
3662
return self._messageSetSearchTest('(5:*)', [3])
3665
def test_searchOr(self):
3667
If the search filter contains an I{OR} term, all messages
3668
which match either subexpression are returned.
3670
return self._messageSetSearchTest('OR 1 2', [1, 2])
3673
def test_searchOrMessageSet(self):
3675
If the search filter contains an I{OR} term with a
3676
subexpression which includes a message sequence set wildcard,
3677
all messages in that set are considered for inclusion in the
3680
return self._messageSetSearchTest('OR 2:* 2:*', [2, 3])
3683
def test_searchNot(self):
3685
If the search filter contains a I{NOT} term, all messages
3686
which do not match the subexpression are returned.
3688
return self._messageSetSearchTest('NOT 3', [1, 2])
3691
def test_searchNotMessageSet(self):
3693
If the search filter contains a I{NOT} term with a
3694
subexpression which includes a message sequence set wildcard,
3695
no messages in that set are considered for inclusion in the
3698
return self._messageSetSearchTest('NOT 2:*', [1])
3701
def test_searchAndMessageSet(self):
3703
If the search filter contains multiple terms implicitly
3704
conjoined with a message sequence set wildcard, only the
3705
intersection of the results of each term are returned.
3707
return self._messageSetSearchTest('2:* 3', [3])
3711
class FetchSearchStoreTestCase(unittest.TestCase, IMAP4HelperMixin):
3712
implements(imap4.ISearchableMailbox)
3715
self.expected = self.result = None
3716
self.server_received_query = None
3717
self.server_received_uid = None
3718
self.server_received_parts = None
3719
self.server_received_messages = None
3721
self.server = imap4.IMAP4Server()
3722
self.server.state = 'select'
3723
self.server.mbox = self
3724
self.connected = defer.Deferred()
3725
self.client = SimpleClient(self.connected)
3727
def search(self, query, uid):
3728
self.server_received_query = query
3729
self.server_received_uid = uid
3730
return self.expected
3732
def addListener(self, *a, **kw):
3734
removeListener = addListener
3736
def _searchWork(self, uid):
3738
return self.client.search(self.query, uid=uid)
3742
self.connected.addCallback(strip(search)
3743
).addCallback(result
3744
).addCallback(self._cbStopClient
3745
).addErrback(self._ebGeneral)
3748
# Ensure no short-circuiting wierdness is going on
3749
self.failIf(self.result is self.expected)
3751
self.assertEquals(self.result, self.expected)
3752
self.assertEquals(self.uid, self.server_received_uid)
3754
imap4.parseNestedParens(self.query),
3755
self.server_received_query
3757
d = loopback.loopbackTCP(self.server, self.client, noisy=False)
3758
d.addCallback(check)
3761
def testSearch(self):
3762
self.query = imap4.Or(
3763
imap4.Query(header=('subject', 'substring')),
3764
imap4.Query(larger=1024, smaller=4096),
3766
self.expected = [1, 4, 5, 7]
3768
return self._searchWork(0)
3770
def testUIDSearch(self):
3771
self.query = imap4.Or(
3772
imap4.Query(header=('subject', 'substring')),
3773
imap4.Query(larger=1024, smaller=4096),
3776
self.expected = [1, 2, 3]
3777
return self._searchWork(1)
3779
def getUID(self, msg):
3781
return self.expected[msg]['UID']
3782
except (TypeError, IndexError):
3783
return self.expected[msg-1]
3787
def fetch(self, messages, uid):
3788
self.server_received_uid = uid
3789
self.server_received_messages = str(messages)
3790
return self.expected
3792
def _fetchWork(self, fetch):
3796
self.connected.addCallback(strip(fetch)
3797
).addCallback(result
3798
).addCallback(self._cbStopClient
3799
).addErrback(self._ebGeneral)
3802
# Ensure no short-circuiting wierdness is going on
3803
self.failIf(self.result is self.expected)
3805
self.parts and self.parts.sort()
3806
self.server_received_parts and self.server_received_parts.sort()
3809
for (k, v) in self.expected.items():
3812
self.assertEquals(self.result, self.expected)
3813
self.assertEquals(self.uid, self.server_received_uid)
3814
self.assertEquals(self.parts, self.server_received_parts)
3815
self.assertEquals(imap4.parseIdList(self.messages),
3816
imap4.parseIdList(self.server_received_messages))
3818
d = loopback.loopbackTCP(self.server, self.client, noisy=False)
3819
d.addCallback(check)
3827
def addMessage(self, body, flags, date):
3828
self.args.append((body, flags, date))
3829
return defer.succeed(None)
3831
class FeaturefulMessage:
3832
implements(imap4.IMessageFile)
3837
def getInternalDate(self):
3838
return 'internaldate'
3841
return StringIO("open")
3843
class MessageCopierMailbox:
3844
implements(imap4.IMessageCopier)
3849
def copy(self, msg):
3850
self.msgs.append(msg)
3851
return len(self.msgs)
3853
class CopyWorkerTestCase(unittest.TestCase):
3854
def testFeaturefulMessage(self):
3855
s = imap4.IMAP4Server()
3857
# Yes. I am grabbing this uber-non-public method to test it.
3858
# It is complex. It needs to be tested directly!
3859
# Perhaps it should be refactored, simplified, or split up into
3860
# not-so-private components, but that is a task for another day.
3862
# Ha ha! Addendum! Soon it will be split up, and this test will
3863
# be re-written to just use the default adapter for IMailbox to
3864
# IMessageCopier and call .copy on that adapter.
3865
f = s._IMAP4Server__cbCopy
3868
d = f([(i, FeaturefulMessage()) for i in range(1, 11)], 'tag', m)
3870
def cbCopy(results):
3872
self.assertEquals(a[0].read(), "open")
3873
self.assertEquals(a[1], "flags")
3874
self.assertEquals(a[2], "internaldate")
3876
for (status, result) in results:
3877
self.failUnless(status)
3878
self.assertEquals(result, None)
3880
return d.addCallback(cbCopy)
3883
def testUnfeaturefulMessage(self):
3884
s = imap4.IMAP4Server()
3887
f = s._IMAP4Server__cbCopy
3890
msgs = [FakeyMessage({'Header-Counter': str(i)}, (), 'Date', 'Body %d' % (i,), i + 10, None) for i in range(1, 11)]
3891
d = f([im for im in zip(range(1, 11), msgs)], 'tag', m)
3893
def cbCopy(results):
3896
seen.append(a[0].read())
3897
self.assertEquals(a[1], ())
3898
self.assertEquals(a[2], "Date")
3901
exp = ["Header-Counter: %d\r\n\r\nBody %d" % (i, i) for i in range(1, 11)]
3903
self.assertEquals(seen, exp)
3905
for (status, result) in results:
3906
self.failUnless(status)
3907
self.assertEquals(result, None)
3909
return d.addCallback(cbCopy)
3911
def testMessageCopier(self):
3912
s = imap4.IMAP4Server()
3915
f = s._IMAP4Server__cbCopy
3917
m = MessageCopierMailbox()
3918
msgs = [object() for i in range(1, 11)]
3919
d = f([im for im in zip(range(1, 11), msgs)], 'tag', m)
3921
def cbCopy(results):
3922
self.assertEquals(results, zip([1] * 10, range(1, 11)))
3923
for (orig, new) in zip(msgs, m.msgs):
3924
self.assertIdentical(orig, new)
3926
return d.addCallback(cbCopy)
3929
class TLSTestCase(IMAP4HelperMixin, unittest.TestCase):
3930
serverCTX = ServerTLSContext and ServerTLSContext()
3931
clientCTX = ClientTLSContext and ClientTLSContext()
3934
return loopback.loopbackTCP(self.server, self.client, noisy=False)
3936
def testAPileOfThings(self):
3937
SimpleServer.theAccount.addMailbox('inbox')
3941
return self.client.login('testuser', 'password-test')
3944
return self.client.list('inbox', '%')
3947
return self.client.status('inbox', 'UIDNEXT')
3950
return self.client.examine('inbox')
3953
return self.client.logout()
3955
self.client.requireTransportSecurity = True
3957
methods = [login, list, status, examine, logout]
3958
map(self.connected.addCallback, map(strip, methods))
3959
self.connected.addCallbacks(self._cbStopClient, self._ebGeneral)
3961
self.assertEquals(self.server.startedTLS, True)
3962
self.assertEquals(self.client.startedTLS, True)
3963
self.assertEquals(len(called), len(methods))
3965
d.addCallback(check)
3968
def testLoginLogin(self):
3969
self.server.checker.addUser('testuser', 'password-test')
3971
self.client.registerAuthenticator(imap4.LOGINAuthenticator('testuser'))
3972
self.connected.addCallback(
3973
lambda _: self.client.authenticate('password-test')
3975
lambda _: self.client.logout()
3976
).addCallback(success.append
3977
).addCallback(self._cbStopClient
3978
).addErrback(self._ebGeneral)
3981
d.addCallback(lambda x : self.assertEquals(len(success), 1))
3985
def test_startTLS(self):
3987
L{IMAP4Client.startTLS} triggers TLS negotiation and returns a
3988
L{Deferred} which fires after the client's transport is using
3992
self.connected.addCallback(lambda _: self.client.startTLS())
3993
def checkSecure(ignored):
3995
interfaces.ISSLTransport.providedBy(self.client.transport))
3996
self.connected.addCallback(checkSecure)
3997
self.connected.addCallback(self._cbStopClient)
3998
self.connected.addCallback(success.append)
3999
self.connected.addErrback(self._ebGeneral)
4002
d.addCallback(lambda x : self.failUnless(success))
4003
return defer.gatherResults([d, self.connected])
4006
def testFailedStartTLS(self):
4008
def breakServerTLS(ign):
4009
self.server.canStartTLS = False
4011
self.connected.addCallback(breakServerTLS)
4012
self.connected.addCallback(lambda ign: self.client.startTLS())
4013
self.connected.addErrback(lambda err: failure.append(err.trap(imap4.IMAP4Exception)))
4014
self.connected.addCallback(self._cbStopClient)
4015
self.connected.addErrback(self._ebGeneral)
4018
self.failUnless(failure)
4019
self.assertIdentical(failure[0], imap4.IMAP4Exception)
4020
return self.loopback().addCallback(check)
4024
class SlowMailbox(SimpleMailbox):
4027
fetchDeferred = None
4029
# Not a very nice implementation of fetch(), but it'll
4030
# do for the purposes of testing.
4031
def fetch(self, messages, uid):
4032
d = defer.Deferred()
4033
self.callLater(self.howSlow, d.callback, ())
4034
self.fetchDeferred.callback(None)
4037
class Timeout(IMAP4HelperMixin, unittest.TestCase):
4039
def test_serverTimeout(self):
4041
The *client* has a timeout mechanism which will close connections that
4042
are inactive for a period.
4045
self.server.timeoutTest = True
4046
self.client.timeout = 5 #seconds
4047
self.client.callLater = c.callLater
4048
self.selectedArgs = None
4051
d = self.client.login('testuser', 'password-test')
4053
d.addErrback(timedOut)
4056
def timedOut(failure):
4057
self._cbStopClient(None)
4058
failure.trap(error.TimeoutError)
4060
d = self.connected.addCallback(strip(login))
4061
d.addErrback(self._ebGeneral)
4062
return defer.gatherResults([d, self.loopback()])
4065
def test_longFetchDoesntTimeout(self):
4067
The connection timeout does not take effect during fetches.
4070
SlowMailbox.callLater = c.callLater
4071
SlowMailbox.fetchDeferred = defer.Deferred()
4072
self.server.callLater = c.callLater
4073
SimpleServer.theAccount.mailboxFactory = SlowMailbox
4074
SimpleServer.theAccount.addMailbox('mailbox-test')
4076
self.server.setTimeout(1)
4079
return self.client.login('testuser', 'password-test')
4081
self.server.setTimeout(1)
4082
return self.client.select('mailbox-test')
4084
return self.client.fetchUID('1:*')
4085
def stillConnected():
4086
self.assertNotEquals(self.server.state, 'timeout')
4088
def cbAdvance(ignored):
4092
SlowMailbox.fetchDeferred.addCallback(cbAdvance)
4094
d1 = self.connected.addCallback(strip(login))
4095
d1.addCallback(strip(select))
4096
d1.addCallback(strip(fetch))
4097
d1.addCallback(strip(stillConnected))
4098
d1.addCallback(self._cbStopClient)
4099
d1.addErrback(self._ebGeneral)
4100
d = defer.gatherResults([d1, self.loopback()])
4104
def test_idleClientDoesDisconnect(self):
4106
The *server* has a timeout mechanism which will close connections that
4107
are inactive for a period.
4110
# Hook up our server protocol
4111
transport = StringTransportWithDisconnection()
4112
transport.protocol = self.server
4113
self.server.callLater = c.callLater
4114
self.server.makeConnection(transport)
4116
# Make sure we can notice when the connection goes away
4118
connLost = self.server.connectionLost
4119
self.server.connectionLost = lambda reason: (lost.append(None), connLost(reason))[1]
4121
# 2/3rds of the idle timeout elapses...
4122
c.pump([0.0] + [self.server.timeOut / 3.0] * 2)
4123
self.failIf(lost, lost)
4126
c.pump([0.0, self.server.timeOut / 2.0])
4127
self.failUnless(lost)
4131
class Disconnection(unittest.TestCase):
4132
def testClientDisconnectFailsDeferreds(self):
4133
c = imap4.IMAP4Client()
4134
t = StringTransportWithDisconnection()
4136
d = self.assertFailure(c.login('testuser', 'example.com'), error.ConnectionDone)
4137
c.connectionLost(error.ConnectionDone("Connection closed"))
4142
class SynchronousMailbox(object):
4144
Trivial, in-memory mailbox implementation which can produce a message
4147
def __init__(self, messages):
4148
self.messages = messages
4151
def fetch(self, msgset, uid):
4152
assert not uid, "Cannot handle uid requests."
4154
yield msg, self.messages[msg - 1]
4158
class StringTransportConsumer(StringTransport):
4162
def registerProducer(self, producer, streaming):
4163
self.producer = producer
4164
self.streaming = streaming
4168
class Pipelining(unittest.TestCase):
4170
Tests for various aspects of the IMAP4 server's pipelining support.
4173
FakeyMessage({}, [], '', '0', None, None),
4174
FakeyMessage({}, [], '', '1', None, None),
4175
FakeyMessage({}, [], '', '2', None, None),
4181
self.transport = StringTransportConsumer()
4182
self.server = imap4.IMAP4Server(None, None, self.iterateInReactor)
4183
self.server.makeConnection(self.transport)
4186
def iterateInReactor(self, iterator):
4187
d = defer.Deferred()
4188
self.iterators.append((iterator, d))
4193
self.server.connectionLost(failure.Failure(error.ConnectionDone()))
4196
def test_synchronousFetch(self):
4198
Test that pipelined FETCH commands which can be responded to
4199
synchronously are responded to correctly.
4201
mailbox = SynchronousMailbox(self.messages)
4203
# Skip over authentication and folder selection
4204
self.server.state = 'select'
4205
self.server.mbox = mailbox
4207
# Get rid of any greeting junk
4208
self.transport.clear()
4210
# Here's some pipelined stuff
4211
self.server.dataReceived(
4212
'01 FETCH 1 BODY[]\r\n'
4213
'02 FETCH 2 BODY[]\r\n'
4214
'03 FETCH 3 BODY[]\r\n')
4216
# Flush anything the server has scheduled to run
4217
while self.iterators:
4218
for e in self.iterators[0][0]:
4221
self.iterators.pop(0)[1].callback(None)
4223
# The bodies are empty because we aren't simulating a transport
4224
# exactly correctly (we have StringTransportConsumer but we never
4225
# call resumeProducing on its producer). It doesn't matter: just
4226
# make sure the surrounding structure is okay, and that no
4227
# exceptions occurred.
4229
self.transport.value(),
4230
'* 1 FETCH (BODY[] )\r\n'
4231
'01 OK FETCH completed\r\n'
4232
'* 2 FETCH (BODY[] )\r\n'
4233
'02 OK FETCH completed\r\n'
4234
'* 3 FETCH (BODY[] )\r\n'
4235
'03 OK FETCH completed\r\n')
4239
if ClientTLSContext is None:
4240
for case in (TLSTestCase,):
4241
case.skip = "OpenSSL not present"
4242
elif interfaces.IReactorSSL(reactor, None) is None:
4243
for case in (TLSTestCase,):
4244
case.skip = "Reactor doesn't support SSL"
added
removed
vendor/Twisted-10.0.0/twisted/mail/test/test_imap.py
