# -*- test-case-name: twisted.mail.test.test_imap -*-
# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.
"""
Test case for twisted.mail.imap4
"""
import codecs
import locale
import os
from io import BytesIO
from itertools import chain
from collections import OrderedDict
from zope.interface import implementer
from twisted.internet import defer
from twisted.internet import error
from twisted.internet import interfaces
from twisted.internet import reactor
from twisted.internet.task import Clock
from twisted.mail import imap4
from twisted.mail.imap4 import MessageSet
from twisted.protocols import loopback
from twisted.python import failure
from twisted.python import util, log
from twisted.python.compat import intToBytes, networkString, range
from twisted.trial import unittest
from twisted.cred.portal import Portal
from twisted.cred.checkers import InMemoryUsernamePasswordDatabaseDontUse
from twisted.cred.error import UnauthorizedLogin
from twisted.cred.credentials import (
IUsernameHashedPassword, IUsernamePassword, CramMD5Credentials)
from twisted.test.proto_helpers import StringTransport, StringTransportWithDisconnection
try:
from twisted.test.ssl_helpers import ClientTLSContext, ServerTLSContext
except ImportError:
ClientTLSContext = ServerTLSContext = None
def strip(f):
return lambda result, f=f: f()
def sortNest(l):
l = l[:]
l.sort()
for i in range(len(l)):
if isinstance(l[i], list):
l[i] = sortNest(l[i])
elif isinstance(l[i], tuple):
l[i] = tuple(sortNest(list(l[i])))
return l
class IMAP4UTF7Tests(unittest.TestCase):
tests = [
[u'Hello world', b'Hello world'],
[u'Hello & world', b'Hello &- world'],
[u'Hello\xffworld', b'Hello&AP8-world'],
[u'\xff\xfe\xfd\xfc', b'&AP8A,gD9APw-'],
[u'~peter/mail/\u65e5\u672c\u8a9e/\u53f0\u5317',
b'~peter/mail/&ZeVnLIqe-/&U,BTFw-'], # example from RFC 2060
]
def test_encodeWithErrors(self):
"""
Specifying an error policy to C{unicode.encode} with the
I{imap4-utf-7} codec should produce the same result as not
specifying the error policy.
"""
text = u'Hello world'
self.assertEqual(
text.encode('imap4-utf-7', 'strict'),
text.encode('imap4-utf-7'))
def test_decodeWithErrors(self):
"""
Similar to L{test_encodeWithErrors}, but for C{bytes.decode}.
"""
bytes = b'Hello world'
self.assertEqual(
bytes.decode('imap4-utf-7', 'strict'),
bytes.decode('imap4-utf-7'))
def test_getreader(self):
"""
C{codecs.getreader('imap4-utf-7')} returns the I{imap4-utf-7} stream
reader class.
"""
reader = codecs.getreader('imap4-utf-7')(BytesIO(b'Hello&AP8-world'))
self.assertEqual(reader.read(), u'Hello\xffworld')
def test_getwriter(self):
"""
C{codecs.getwriter('imap4-utf-7')} returns the I{imap4-utf-7} stream
writer class.
"""
output = BytesIO()
writer = codecs.getwriter('imap4-utf-7')(output)
writer.write(u'Hello\xffworld')
self.assertEqual(output.getvalue(), b'Hello&AP8-world')
def test_encode(self):
"""
The I{imap4-utf-7} can be used to encode a unicode string into a byte
string according to the IMAP4 modified UTF-7 encoding rules.
"""
for (input, output) in self.tests:
self.assertEqual(input.encode('imap4-utf-7'), output)
def test_decode(self):
"""
The I{imap4-utf-7} can be used to decode a byte string into a unicode
string according to the IMAP4 modified UTF-7 encoding rules.
"""
for (input, output) in self.tests:
self.assertEqual(input, output.decode('imap4-utf-7'))
def test_printableSingletons(self):
"""
The IMAP4 modified UTF-7 implementation encodes all printable
characters which are in ASCII using the corresponding ASCII byte.
"""
# All printables represent themselves
for o in chain(range(0x20, 0x26), range(0x27, 0x7f)):
charbyte = chr(o).encode()
self.assertEqual(charbyte, chr(o).encode('imap4-utf-7'))
self.assertEqual(chr(o), charbyte.decode('imap4-utf-7'))
self.assertEqual(u'&'.encode('imap4-utf-7'), b'&-')
self.assertEqual(b'&-'.decode('imap4-utf-7'), u'&')
class BufferingConsumer:
def __init__(self):
self.buffer = []
def write(self, bytes):
self.buffer.append(bytes)
if self.consumer:
self.consumer.resumeProducing()
def registerProducer(self, consumer, streaming):
self.consumer = consumer
self.consumer.resumeProducing()
def unregisterProducer(self):
self.consumer = None
class MessageProducerTests(unittest.TestCase):
def testSinglePart(self):
body = b'This is body text. Rar.'
headers = OrderedDict()
headers[b'from'] = b'sender@host'
headers[b'to'] = b'recipient@domain'
headers[b'subject'] = b'booga booga boo'
headers[b'content-type'] = b'text/plain'
msg = FakeyMessage(headers, (), None, body, 123, None )
c = BufferingConsumer()
p = imap4.MessageProducer(msg)
d = p.beginProducing(c)
def cbProduced(result):
self.assertIdentical(result, p)
self.assertEqual(
b''.join(c.buffer),
b'{119}\r\n'
b'From: sender@host\r\n'
b'To: recipient@domain\r\n'
b'Subject: booga booga boo\r\n'
b'Content-Type: text/plain\r\n'
b'\r\n'
+ body)
return d.addCallback(cbProduced)
def testSingleMultiPart(self):
outerBody = b''
innerBody = b'Contained body message text. Squarge.'
headers = OrderedDict()
headers[b'from'] = b'sender@host'
headers[b'to'] = b'recipient@domain'
headers[b'subject'] = b'booga booga boo'
headers[b'content-type'] = b'multipart/alternative; boundary="xyz"'
innerHeaders = OrderedDict()
innerHeaders[b'subject'] = b'this is subject text'
innerHeaders[b'content-type'] = b'text/plain'
msg = FakeyMessage(headers, (), None, outerBody, 123,
[FakeyMessage(innerHeaders, (), None, innerBody,
None, None)],
)
c = BufferingConsumer()
p = imap4.MessageProducer(msg)
d = p.beginProducing(c)
def cbProduced(result):
self.failUnlessIdentical(result, p)
self.assertEqual(
b''.join(c.buffer),
b'{239}\r\n'
b'From: sender@host\r\n'
b'To: recipient@domain\r\n'
b'Subject: booga booga boo\r\n'
b'Content-Type: multipart/alternative; boundary="xyz"\r\n'
b'\r\n'
b'\r\n'
b'--xyz\r\n'
b'Subject: this is subject text\r\n'
b'Content-Type: text/plain\r\n'
b'\r\n'
+ innerBody
+ b'\r\n--xyz--\r\n')
return d.addCallback(cbProduced)
def testMultipleMultiPart(self):
outerBody = b''
innerBody1 = b'Contained body message text. Squarge.'
innerBody2 = b'Secondary message text of squarge body.'
headers = OrderedDict()
headers[b'from'] = b'sender@host'
headers[b'to'] = b'recipient@domain'
headers[b'subject'] = b'booga booga boo'
headers[b'content-type'] = b'multipart/alternative; boundary="xyz"'
innerHeaders = OrderedDict()
innerHeaders[b'subject'] = b'this is subject text'
innerHeaders[b'content-type'] = b'text/plain'
innerHeaders2 = OrderedDict()
innerHeaders2[b'subject'] = b'this is subject'
innerHeaders2[b'content-type'] = b'text/html'
msg = FakeyMessage(headers, (), None, outerBody, 123, [
FakeyMessage(innerHeaders, (), None, innerBody1, None, None),
FakeyMessage(innerHeaders2, (), None, innerBody2, None, None)
],
)
c = BufferingConsumer()
p = imap4.MessageProducer(msg)
d = p.beginProducing(c)
def cbProduced(result):
self.failUnlessIdentical(result, p)
self.assertEqual(
b''.join(c.buffer),
b'{354}\r\n'
b'From: sender@host\r\n'
b'To: recipient@domain\r\n'
b'Subject: booga booga boo\r\n'
b'Content-Type: multipart/alternative; boundary="xyz"\r\n'
b'\r\n'
b'\r\n'
b'--xyz\r\n'
b'Subject: this is subject text\r\n'
b'Content-Type: text/plain\r\n'
b'\r\n'
+ innerBody1
+ b'\r\n--xyz\r\n'
b'Subject: this is subject\r\n'
b'Content-Type: text/html\r\n'
b'\r\n'
+ innerBody2
+ b'\r\n--xyz--\r\n')
return d.addCallback(cbProduced)
class IMAP4HelperTests(unittest.TestCase):
"""
Tests for various helper utilities in the IMAP4 module.
"""
def test_fileProducer(self):
b = (('x' * 1) + ('y' * 1) + ('z' * 1)) * 10
c = BufferingConsumer()
f = BytesIO(b)
p = imap4.FileProducer(f)
d = p.beginProducing(c)
def cbProduced(result):
self.failUnlessIdentical(result, p)
self.assertEqual(
('{%d}\r\n' % len(b))+ b,
''.join(c.buffer))
return d.addCallback(cbProduced)
def test_wildcard(self):
cases = [
['foo/%gum/bar',
['foo/bar', 'oo/lalagum/bar', 'foo/gumx/bar', 'foo/gum/baz'],
['foo/xgum/bar', 'foo/gum/bar'],
], ['foo/x%x/bar',
['foo', 'bar', 'fuz fuz fuz', 'foo/*/bar', 'foo/xyz/bar', 'foo/xx/baz'],
['foo/xyx/bar', 'foo/xx/bar', 'foo/xxxxxxxxxxxxxx/bar'],
], ['foo/xyz*abc/bar',
['foo/xyz/bar', 'foo/abc/bar', 'foo/xyzab/cbar', 'foo/xyza/bcbar'],
['foo/xyzabc/bar', 'foo/xyz/abc/bar', 'foo/xyz/123/abc/bar'],
]
]
for (wildcard, fail, succeed) in cases:
wildcard = imap4.wildcardToRegexp(wildcard, '/')
for x in fail:
self.assertFalse(wildcard.match(x))
for x in succeed:
self.assertTrue(wildcard.match(x))
def test_wildcardNoDelim(self):
cases = [
['foo/%gum/bar',
['foo/bar', 'oo/lalagum/bar', 'foo/gumx/bar', 'foo/gum/baz'],
['foo/xgum/bar', 'foo/gum/bar', 'foo/x/gum/bar'],
], ['foo/x%x/bar',
['foo', 'bar', 'fuz fuz fuz', 'foo/*/bar', 'foo/xyz/bar', 'foo/xx/baz'],
['foo/xyx/bar', 'foo/xx/bar', 'foo/xxxxxxxxxxxxxx/bar', 'foo/x/x/bar'],
], ['foo/xyz*abc/bar',
['foo/xyz/bar', 'foo/abc/bar', 'foo/xyzab/cbar', 'foo/xyza/bcbar'],
['foo/xyzabc/bar', 'foo/xyz/abc/bar', 'foo/xyz/123/abc/bar'],
]
]
for (wildcard, fail, succeed) in cases:
wildcard = imap4.wildcardToRegexp(wildcard, None)
for x in fail:
self.assertFalse(wildcard.match(x), x)
for x in succeed:
self.assertTrue(wildcard.match(x), x)
def test_headerFormatter(self):
"""
L{imap4._formatHeaders} accepts a C{dict} of header name/value pairs and
returns a string representing those headers in the standard multiline,
C{":"}-separated format.
"""
cases = [
({'Header1': 'Value1', 'Header2': 'Value2'}, 'Header2: Value2\r\nHeader1: Value1\r\n'),
]
for (input, expected) in cases:
output = imap4._formatHeaders(input)
self.assertEqual(sorted(output.splitlines(True)),
sorted(expected.splitlines(True)))
def test_messageSet(self):
m1 = MessageSet()
m2 = MessageSet()
self.assertEqual(m1, m2)
m1 = m1 + (1, 3)
self.assertEqual(len(m1), 3)
self.assertEqual(list(m1), [1, 2, 3])
m2 = m2 + (1, 3)
self.assertEqual(m1, m2)
self.assertEqual(list(m1 + m2), [1, 2, 3])
def test_messageSetStringRepresentationWithWildcards(self):
"""
In a L{MessageSet}, in the presence of wildcards, if the highest message
id is known, the wildcard should get replaced by that high value.
"""
inputs = [
MessageSet(imap4.parseIdList(b'*')),
MessageSet(imap4.parseIdList(b'3:*', 6)),
MessageSet(imap4.parseIdList(b'*:2', 6)),
]
outputs = [
b"*",
b"3:6",
b"2:6",
]
for i, o in zip(inputs, outputs):
self.assertEqual(str(i), o)
def test_messageSetStringRepresentationWithInversion(self):
"""
In a L{MessageSet}, inverting the high and low numbers in a range
doesn't affect the meaning of the range. For example, 3:2 displays just
like 2:3, because according to the RFC they have the same meaning.
"""
inputs = [
MessageSet(imap4.parseIdList(b'2:3')),
MessageSet(imap4.parseIdList(b'3:2')),
]
outputs = [
"2:3",
"2:3",
]
for i, o in zip(inputs, outputs):
self.assertEqual(str(i), o)
def test_quotedSplitter(self):
cases = [
'''Hello World''',
'''Hello "World!"''',
'''World "Hello" "How are you?"''',
'''"Hello world" How "are you?"''',
'''foo bar "baz buz" NIL''',
'''foo bar "baz buz" "NIL"''',
'''foo NIL "baz buz" bar''',
'''foo "NIL" "baz buz" bar''',
'''"NIL" bar "baz buz" foo''',
'oo \\"oo\\" oo',
'"oo \\"oo\\" oo"',
'oo \t oo',
'"oo \t oo"',
'oo \\t oo',
'"oo \\t oo"',
'oo \o oo',
'"oo \o oo"',
'oo \\o oo',
'"oo \\o oo"',
]
answers = [
['Hello', 'World'],
['Hello', 'World!'],
['World', 'Hello', 'How are you?'],
['Hello world', 'How', 'are you?'],
['foo', 'bar', 'baz buz', None],
['foo', 'bar', 'baz buz', 'NIL'],
['foo', None, 'baz buz', 'bar'],
['foo', 'NIL', 'baz buz', 'bar'],
['NIL', 'bar', 'baz buz', 'foo'],
['oo', '"oo"', 'oo'],
['oo "oo" oo'],
['oo', 'oo'],
['oo \t oo'],
['oo', '\\t', 'oo'],
['oo \\t oo'],
['oo', '\o', 'oo'],
['oo \o oo'],
['oo', '\\o', 'oo'],
['oo \\o oo'],
]
errors = [
'"mismatched quote',
'mismatched quote"',
'mismatched"quote',
'"oops here is" another"',
]
for s in errors:
self.assertRaises(imap4.MismatchedQuoting, imap4.splitQuoted, s)
for (case, expected) in zip(cases, answers):
self.assertEqual(imap4.splitQuoted(case), expected)
def test_stringCollapser(self):
cases = [
['a', 'b', 'c', 'd', 'e'],
['a', ' ', '"', 'b', 'c', ' ', '"', ' ', 'd', 'e'],
[['a', 'b', 'c'], 'd', 'e'],
['a', ['b', 'c', 'd'], 'e'],
['a', 'b', ['c', 'd', 'e']],
['"', 'a', ' ', '"', ['b', 'c', 'd'], '"', ' ', 'e', '"'],
['a', ['"', ' ', 'b', 'c', ' ', ' ', '"'], 'd', 'e'],
]
answers = [
['abcde'],
['a', 'bc ', 'de'],
[['abc'], 'de'],
['a', ['bcd'], 'e'],
['ab', ['cde']],
['a ', ['bcd'], ' e'],
['a', [' bc '], 'de'],
]
for (case, expected) in zip(cases, answers):
self.assertEqual(imap4.collapseStrings(case), expected)
def test_parenParser(self):
s = '\r\n'.join(['xx'] * 4)
cases = [
'(BODY.PEEK[HEADER.FIELDS.NOT (subject bcc cc)] {%d}\r\n%s)' % (len(s), s,),
# '(FLAGS (\Seen) INTERNALDATE "17-Jul-1996 02:44:25 -0700" '
# 'RFC822.SIZE 4286 ENVELOPE ("Wed, 17 Jul 1996 02:23:25 -0700 (PDT)" '
# '"IMAP4rev1 WG mtg summary and minutes" '
# '(("Terry Gray" NIL "gray" "cac.washington.edu")) '
# '(("Terry Gray" NIL "gray" "cac.washington.edu")) '
# '(("Terry Gray" NIL "gray" "cac.washington.edu")) '
# '((NIL NIL "imap" "cac.washington.edu")) '
# '((NIL NIL "minutes" "CNRI.Reston.VA.US") '
# '("John Klensin" NIL "KLENSIN" "INFOODS.MIT.EDU")) NIL NIL '
# '"") '
# 'BODY ("TEXT" "PLAIN" ("CHARSET" "US-ASCII") NIL NIL "7BIT" 3028 92))',
'(FLAGS (\Seen) INTERNALDATE "17-Jul-1996 02:44:25 -0700" '
'RFC822.SIZE 4286 ENVELOPE ("Wed, 17 Jul 1996 02:23:25 -0700 (PDT)" '
'"IMAP4rev1 WG mtg summary and minutes" '
'(("Terry Gray" NIL gray cac.washington.edu)) '
'(("Terry Gray" NIL gray cac.washington.edu)) '
'(("Terry Gray" NIL gray cac.washington.edu)) '
'((NIL NIL imap cac.washington.edu)) '
'((NIL NIL minutes CNRI.Reston.VA.US) '
'("John Klensin" NIL KLENSIN INFOODS.MIT.EDU)) NIL NIL '
') '
'BODY (TEXT PLAIN (CHARSET US-ASCII) NIL NIL 7BIT 3028 92))',
'("oo \\"oo\\" oo")',
'("oo \\\\ oo")',
'("oo \\ oo")',
'("oo \\o")',
'("oo \o")',
'(oo \o)',
'(oo \\o)',
]
answers = [
['BODY.PEEK', ['HEADER.FIELDS.NOT', ['subject', 'bcc', 'cc']], s],
['FLAGS', [r'\Seen'], 'INTERNALDATE',
'17-Jul-1996 02:44:25 -0700', 'RFC822.SIZE', '4286', 'ENVELOPE',
['Wed, 17 Jul 1996 02:23:25 -0700 (PDT)',
'IMAP4rev1 WG mtg summary and minutes', [["Terry Gray", None,
"gray", "cac.washington.edu"]], [["Terry Gray", None,
"gray", "cac.washington.edu"]], [["Terry Gray", None,
"gray", "cac.washington.edu"]], [[None, None, "imap",
"cac.washington.edu"]], [[None, None, "minutes",
"CNRI.Reston.VA.US"], ["John Klensin", None, "KLENSIN",
"INFOODS.MIT.EDU"]], None, None,
""], "BODY", ["TEXT", "PLAIN",
["CHARSET", "US-ASCII"], None, None, "7BIT", "3028", "92"]],
['oo "oo" oo'],
['oo \\\\ oo'],
['oo \\ oo'],
['oo \\o'],
['oo \o'],
['oo', '\o'],
['oo', '\\o'],
]
for (case, expected) in zip(cases, answers):
self.assertEqual(imap4.parseNestedParens(case), [expected])
# XXX This code used to work, but changes occurred within the
# imap4.py module which made it no longer necessary for *all* of it
# to work. In particular, only the part that makes
# 'BODY.PEEK[HEADER.FIELDS.NOT (Subject Bcc Cc)]' come out correctly
# no longer needs to work. So, I am loathe to delete the entire
# section of the test. --exarkun
#
# for (case, expected) in zip(answers, cases):
# self.assertEqual('(' + imap4.collapseNestedLists(case) + ')', expected)
def test_fetchParserSimple(self):
cases = [
['ENVELOPE', 'Envelope'],
['FLAGS', 'Flags'],
['INTERNALDATE', 'InternalDate'],
['RFC822.HEADER', 'RFC822Header'],
['RFC822.SIZE', 'RFC822Size'],
['RFC822.TEXT', 'RFC822Text'],
['RFC822', 'RFC822'],
['UID', 'UID'],
['BODYSTRUCTURE', 'BodyStructure'],
]
for (inp, outp) in cases:
p = imap4._FetchParser()
p.parseString(inp)
self.assertEqual(len(p.result), 1)
self.assertTrue(isinstance(p.result[0], getattr(p, outp)))
def test_fetchParserMacros(self):
cases = [
['ALL', (4, ['flags', 'internaldate', 'rfc822.size', 'envelope'])],
['FULL', (5, ['flags', 'internaldate', 'rfc822.size', 'envelope', 'body'])],
['FAST', (3, ['flags', 'internaldate', 'rfc822.size'])],
]
for (inp, outp) in cases:
p = imap4._FetchParser()
p.parseString(inp)
self.assertEqual(len(p.result), outp[0])
expectedResult = [str(token).lower() for token in p.result]
expectedResult.sort()
outp[1].sort()
self.assertEqual(expectedResult, outp[1])
def test_fetchParserBody(self):
P = imap4._FetchParser
p = P()
p.parseString('BODY')
self.assertEqual(len(p.result), 1)
self.assertTrue(isinstance(p.result[0], p.Body))
self.assertEqual(p.result[0].peek, False)
self.assertEqual(p.result[0].header, None)
self.assertEqual(str(p.result[0]), 'BODY')
p = P()
p.parseString('BODY.PEEK')
self.assertEqual(len(p.result), 1)
self.assertTrue(isinstance(p.result[0], p.Body))
self.assertEqual(p.result[0].peek, True)
self.assertEqual(str(p.result[0]), 'BODY')
p = P()
p.parseString('BODY[]')
self.assertEqual(len(p.result), 1)
self.assertTrue(isinstance(p.result[0], p.Body))
self.assertEqual(p.result[0].empty, True)
self.assertEqual(str(p.result[0]), 'BODY[]')
p = P()
p.parseString('BODY[HEADER]')
self.assertEqual(len(p.result), 1)
self.assertTrue(isinstance(p.result[0], p.Body))
self.assertEqual(p.result[0].peek, False)
self.assertTrue(isinstance(p.result[0].header, p.Header))
self.assertEqual(p.result[0].header.negate, True)
self.assertEqual(p.result[0].header.fields, ())
self.assertEqual(p.result[0].empty, False)
self.assertEqual(str(p.result[0]), 'BODY[HEADER]')
p = P()
p.parseString('BODY.PEEK[HEADER]')
self.assertEqual(len(p.result), 1)
self.assertTrue(isinstance(p.result[0], p.Body))
self.assertEqual(p.result[0].peek, True)
self.assertTrue(isinstance(p.result[0].header, p.Header))
self.assertEqual(p.result[0].header.negate, True)
self.assertEqual(p.result[0].header.fields, ())
self.assertEqual(p.result[0].empty, False)
self.assertEqual(str(p.result[0]), 'BODY[HEADER]')
p = P()
p.parseString('BODY[HEADER.FIELDS (Subject Cc Message-Id)]')
self.assertEqual(len(p.result), 1)
self.assertTrue(isinstance(p.result[0], p.Body))
self.assertEqual(p.result[0].peek, False)
self.assertTrue(isinstance(p.result[0].header, p.Header))
self.assertEqual(p.result[0].header.negate, False)
self.assertEqual(p.result[0].header.fields, ['SUBJECT', 'CC', 'MESSAGE-ID'])
self.assertEqual(p.result[0].empty, False)
self.assertEqual(str(p.result[0]), 'BODY[HEADER.FIELDS (Subject Cc Message-Id)]')
p = P()
p.parseString('BODY.PEEK[HEADER.FIELDS (Subject Cc Message-Id)]')
self.assertEqual(len(p.result), 1)
self.assertTrue(isinstance(p.result[0], p.Body))
self.assertEqual(p.result[0].peek, True)
self.assertTrue(isinstance(p.result[0].header, p.Header))
self.assertEqual(p.result[0].header.negate, False)
self.assertEqual(p.result[0].header.fields, ['SUBJECT', 'CC', 'MESSAGE-ID'])
self.assertEqual(p.result[0].empty, False)
self.assertEqual(str(p.result[0]), 'BODY[HEADER.FIELDS (Subject Cc Message-Id)]')
p = P()
p.parseString('BODY.PEEK[HEADER.FIELDS.NOT (Subject Cc Message-Id)]')
self.assertEqual(len(p.result), 1)
self.assertTrue(isinstance(p.result[0], p.Body))
self.assertEqual(p.result[0].peek, True)
self.assertTrue(isinstance(p.result[0].header, p.Header))
self.assertEqual(p.result[0].header.negate, True)
self.assertEqual(p.result[0].header.fields, ['SUBJECT', 'CC', 'MESSAGE-ID'])
self.assertEqual(p.result[0].empty, False)
self.assertEqual(str(p.result[0]), 'BODY[HEADER.FIELDS.NOT (Subject Cc Message-Id)]')
p = P()
p.parseString('BODY[1.MIME]<10.50>')
self.assertEqual(len(p.result), 1)
self.assertTrue(isinstance(p.result[0], p.Body))
self.assertEqual(p.result[0].peek, False)
self.assertTrue(isinstance(p.result[0].mime, p.MIME))
self.assertEqual(p.result[0].part, (0,))
self.assertEqual(p.result[0].partialBegin, 10)
self.assertEqual(p.result[0].partialLength, 50)
self.assertEqual(p.result[0].empty, False)
self.assertEqual(str(p.result[0]), 'BODY[1.MIME]<10.50>')
p = P()
p.parseString('BODY.PEEK[1.3.9.11.HEADER.FIELDS.NOT (Message-Id Date)]<103.69>')
self.assertEqual(len(p.result), 1)
self.assertTrue(isinstance(p.result[0], p.Body))
self.assertEqual(p.result[0].peek, True)
self.assertTrue(isinstance(p.result[0].header, p.Header))
self.assertEqual(p.result[0].part, (0, 2, 8, 10))
self.assertEqual(p.result[0].header.fields, ['MESSAGE-ID', 'DATE'])
self.assertEqual(p.result[0].partialBegin, 103)
self.assertEqual(p.result[0].partialLength, 69)
self.assertEqual(p.result[0].empty, False)
self.assertEqual(str(p.result[0]), 'BODY[1.3.9.11.HEADER.FIELDS.NOT (Message-Id Date)]<103.69>')
def test_files(self):
inputStructure = [
'foo', 'bar', 'baz', BytesIO(b'this is a file\r\n'), 'buz',
u'biz'
]
output = '"foo" "bar" "baz" {16}\r\nthis is a file\r\n "buz" "biz"'
self.assertEqual(imap4.collapseNestedLists(inputStructure), output)
def test_quoteAvoider(self):
input = [
'foo', imap4.DontQuoteMe('bar'), "baz", BytesIO(b'this is a file\r\n'),
imap4.DontQuoteMe('buz'), ""
]
output = '"foo" bar "baz" {16}\r\nthis is a file\r\n buz ""'
self.assertEqual(imap4.collapseNestedLists(input), output)
def test_literals(self):
cases = [
('({10}\r\n0123456789)', [['0123456789']]),
]
for (case, expected) in cases:
self.assertEqual(imap4.parseNestedParens(case), expected)
def test_queryBuilder(self):
inputs = [
imap4.Query(flagged=1),
imap4.Query(sorted=1, unflagged=1, deleted=1),
imap4.Or(imap4.Query(flagged=1), imap4.Query(deleted=1)),
imap4.Query(before='today'),
imap4.Or(
imap4.Query(deleted=1),
imap4.Query(unseen=1),
imap4.Query(new=1)
),
imap4.Or(
imap4.Not(
imap4.Or(
imap4.Query(sorted=1, since='yesterday', smaller=1000),
imap4.Query(sorted=1, before='tuesday', larger=10000),
imap4.Query(sorted=1, unseen=1, deleted=1, before='today'),
imap4.Not(
imap4.Query(subject='spam')
),
),
),
imap4.Not(
imap4.Query(uid='1:5')
),
)
]
outputs = [
'FLAGGED',
'(DELETED UNFLAGGED)',
'(OR FLAGGED DELETED)',
'(BEFORE "today")',
'(OR DELETED (OR UNSEEN NEW))',
'(OR (NOT (OR (SINCE "yesterday" SMALLER 1000) ' # Continuing
'(OR (BEFORE "tuesday" LARGER 10000) (OR (BEFORE ' # Some more
'"today" DELETED UNSEEN) (NOT (SUBJECT "spam")))))) ' # And more
'(NOT (UID 1:5)))',
]
for (query, expected) in zip(inputs, outputs):
self.assertEqual(query, expected)
def test_queryKeywordFlagWithQuotes(self):
"""
When passed the C{keyword} argument, L{imap4.Query} returns an unquoted
string.
@see: U{http://tools.ietf.org/html/rfc3501#section-9}
@see: U{http://tools.ietf.org/html/rfc3501#section-6.4.4}
"""
query = imap4.Query(keyword='twisted')
self.assertEqual('(KEYWORD twisted)', query)
def test_queryUnkeywordFlagWithQuotes(self):
"""
When passed the C{unkeyword} argument, L{imap4.Query} returns an
unquoted string.
@see: U{http://tools.ietf.org/html/rfc3501#section-9}
@see: U{http://tools.ietf.org/html/rfc3501#section-6.4.4}
"""
query = imap4.Query(unkeyword='twisted')
self.assertEqual('(UNKEYWORD twisted)', query)
def _keywordFilteringTest(self, keyword):
"""
Helper to implement tests for value filtering of KEYWORD and UNKEYWORD
queries.
@param keyword: A native string giving the name of the L{imap4.Query}
keyword argument to test.
"""
# Check all the printable exclusions
self.assertEqual(
'(%s twistedrocks)' % (keyword.upper(),),
imap4.Query(**{keyword: r'twisted (){%*"\] rocks'}))
# Check all the non-printable exclusions
self.assertEqual(
'(%s twistedrocks)' % (keyword.upper(),),
imap4.Query(**{
keyword: 'twisted %s rocks' % (
''.join(chr(ch) for ch in range(33)),)}))
def test_queryKeywordFlag(self):
"""
When passed the C{keyword} argument, L{imap4.Query} returns an
C{atom} that consists of one or more non-special characters.
List of the invalid characters:
( ) { % * " \ ] CTL SP
@see: U{ABNF definition of CTL and SP}
@see: U{IMAP4 grammar}
@see: U{IMAP4 SEARCH specification}
"""
self._keywordFilteringTest("keyword")
def test_queryUnkeywordFlag(self):
"""
When passed the C{unkeyword} argument, L{imap4.Query} returns an
C{atom} that consists of one or more non-special characters.
List of the invalid characters:
( ) { % * " \ ] CTL SP
@see: U{ABNF definition of CTL and SP}
@see: U{IMAP4 grammar}
@see: U{IMAP4 SEARCH specification}
"""
self._keywordFilteringTest("unkeyword")
def test_invalidIdListParser(self):
"""
Trying to parse an invalid representation of a sequence range raises an
L{IllegalIdentifierError}.
"""
inputs = [
'*:*',
'foo',
'4:',
'bar:5'
]
for input in inputs:
self.assertRaises(imap4.IllegalIdentifierError,
imap4.parseIdList, input, 12345)
def test_invalidIdListParserNonPositive(self):
"""
Zeroes and negative values are not accepted in id range expressions. RFC
3501 states that sequence numbers and sequence ranges consist of
non-negative numbers (RFC 3501 section 9, the seq-number grammar item).
"""
inputs = [
'0:5',
'0:0',
'*:0',
'0',
'-3:5',
'1:-2',
'-1'
]
for input in inputs:
self.assertRaises(imap4.IllegalIdentifierError,
imap4.parseIdList, input, 12345)
def test_parseIdList(self):
"""
The function to parse sequence ranges yields appropriate L{MessageSet}
objects.
"""
inputs = [
'1:*',
'5:*',
'1:2,5:*',
'*',
'1',
'1,2',
'1,3,5',
'1:10',
'1:10,11',
'1:5,10:20',
'1,5:10',
'1,5:10,15:20',
'1:10,15,20:25',
'4:2'
]
outputs = [
MessageSet(1, None),
MessageSet(5, None),
MessageSet(5, None) + MessageSet(1, 2),
MessageSet(None, None),
MessageSet(1),
MessageSet(1, 2),
MessageSet(1) + MessageSet(3) + MessageSet(5),
MessageSet(1, 10),
MessageSet(1, 11),
MessageSet(1, 5) + MessageSet(10, 20),
MessageSet(1) + MessageSet(5, 10),
MessageSet(1) + MessageSet(5, 10) + MessageSet(15, 20),
MessageSet(1, 10) + MessageSet(15) + MessageSet(20, 25),
MessageSet(2, 4),
]
lengths = [
None, None, None,
1, 1, 2, 3, 10, 11, 16, 7, 13, 17, 3
]
for (input, expected) in zip(inputs, outputs):
self.assertEqual(imap4.parseIdList(input), expected)
for (input, expected) in zip(inputs, lengths):
if expected is None:
self.assertRaises(TypeError, len, imap4.parseIdList(input))
else:
L = len(imap4.parseIdList(input))
self.assertEqual(L, expected,
"len(%r) = %r != %r" % (input, L, expected))
@implementer(imap4.IMailboxInfo, imap4.IMailbox, imap4.ICloseableMailbox)
class SimpleMailbox:
flags = ('\\Flag1', 'Flag2', '\\AnotherSysFlag', 'LastFlag')
messages = []
mUID = 0
rw = 1
closed = False
def __init__(self):
self.listeners = []
self.addListener = self.listeners.append
self.removeListener = self.listeners.remove
def getFlags(self):
return self.flags
def getUIDValidity(self):
return 42
def getUIDNext(self):
return len(self.messages) + 1
def getMessageCount(self):
return 9
def getRecentCount(self):
return 3
def getUnseenCount(self):
return 4
def isWriteable(self):
return self.rw
def destroy(self):
pass
def getHierarchicalDelimiter(self):
return '/'
def requestStatus(self, names):
r = {}
if 'MESSAGES' in names:
r['MESSAGES'] = self.getMessageCount()
if 'RECENT' in names:
r['RECENT'] = self.getRecentCount()
if 'UIDNEXT' in names:
r['UIDNEXT'] = self.getMessageCount() + 1
if 'UIDVALIDITY' in names:
r['UIDVALIDITY'] = self.getUID()
if 'UNSEEN' in names:
r['UNSEEN'] = self.getUnseenCount()
return defer.succeed(r)
def addMessage(self, message, flags, date = None):
self.messages.append((message, flags, date, self.mUID))
self.mUID += 1
return defer.succeed(None)
def expunge(self):
delete = []
for i in self.messages:
if '\\Deleted' in i[1]:
delete.append(i)
for i in delete:
self.messages.remove(i)
return [i[3] for i in delete]
def close(self):
self.closed = True
class Account(imap4.MemoryAccount):
mailboxFactory = SimpleMailbox
def _emptyMailbox(self, name, id):
return self.mailboxFactory()
def select(self, name, rw=1):
mbox = imap4.MemoryAccount.select(self, name)
if mbox is not None:
mbox.rw = rw
return mbox
class SimpleServer(imap4.IMAP4Server):
def __init__(self, *args, **kw):
imap4.IMAP4Server.__init__(self, *args, **kw)
realm = TestRealm()
realm.theAccount = Account(b'testuser')
portal = Portal(realm)
c = InMemoryUsernamePasswordDatabaseDontUse()
self.checker = c
self.portal = portal
portal.registerChecker(c)
self.timeoutTest = False
def lineReceived(self, line):
if self.timeoutTest:
#Do not send a response
return
imap4.IMAP4Server.lineReceived(self, line)
_username = b'testuser'
_password = b'password-test'
def authenticateLogin(self, username, password):
if username == self._username and password == self._password:
return imap4.IAccount, self.theAccount, lambda: None
raise UnauthorizedLogin()
class SimpleClient(imap4.IMAP4Client):
def __init__(self, deferred, contextFactory = None):
imap4.IMAP4Client.__init__(self, contextFactory)
self.deferred = deferred
self.events = []
def serverGreeting(self, caps):
self.deferred.callback(None)
def modeChanged(self, writeable):
self.events.append([b'modeChanged', writeable])
self.transport.loseConnection()
def flagsChanged(self, newFlags):
self.events.append([b'flagsChanged', newFlags])
self.transport.loseConnection()
def newMessages(self, exists, recent):
self.events.append([b'newMessages', exists, recent])
self.transport.loseConnection()
class IMAP4HelperMixin:
serverCTX = None
clientCTX = None
def setUp(self):
d = defer.Deferred()
self.server = SimpleServer(contextFactory=self.serverCTX)
self.client = SimpleClient(d, contextFactory=self.clientCTX)
self.connected = d
SimpleMailbox.messages = []
theAccount = Account(b'testuser')
theAccount.mboxType = SimpleMailbox
SimpleServer.theAccount = theAccount
def tearDown(self):
del self.server
del self.client
del self.connected
def _cbStopClient(self, ignore):
self.client.transport.loseConnection()
def _ebGeneral(self, failure):
self.client.transport.loseConnection()
self.server.transport.loseConnection()
log.err(failure, "Problem with " + str(self))
def loopback(self):
return loopback.loopbackAsync(self.server, self.client)
class IMAP4ServerTests(IMAP4HelperMixin, unittest.TestCase):
def testCapability(self):
caps = {}
def getCaps():
def gotCaps(c):
caps.update(c)
self.server.transport.loseConnection()
return self.client.getCapabilities().addCallback(gotCaps)
d1 = self.connected.addCallback(strip(getCaps)).addErrback(self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
expected = {b'IMAP4rev1': None, b'NAMESPACE': None, b'IDLE': None}
return d.addCallback(lambda _: self.assertEqual(expected, caps))
def testCapabilityWithAuth(self):
caps = {}
self.server.challengers[b'CRAM-MD5'] = CramMD5Credentials
def getCaps():
def gotCaps(c):
caps.update(c)
self.server.transport.loseConnection()
return self.client.getCapabilities().addCallback(gotCaps)
d1 = self.connected.addCallback(strip(getCaps)).addErrback(self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
expCap = {b'IMAP4rev1': None, b'NAMESPACE': None,
b'IDLE': None, b'AUTH': [b'CRAM-MD5']}
return d.addCallback(lambda _: self.assertEqual(expCap, caps))
def testLogout(self):
self.loggedOut = 0
def logout():
def setLoggedOut():
self.loggedOut = 1
self.client.logout().addCallback(strip(setLoggedOut))
self.connected.addCallback(strip(logout)).addErrback(self._ebGeneral)
d = self.loopback()
return d.addCallback(lambda _: self.assertEqual(self.loggedOut, 1))
def testNoop(self):
self.responses = None
def noop():
def setResponses(responses):
self.responses = responses
self.server.transport.loseConnection()
self.client.noop().addCallback(setResponses)
self.connected.addCallback(strip(noop)).addErrback(self._ebGeneral)
d = self.loopback()
return d.addCallback(lambda _: self.assertEqual(self.responses, []))
def testLogin(self):
def login():
d = self.client.login(b'testuser', b'password-test')
d.addCallback(self._cbStopClient)
d1 = self.connected.addCallback(strip(login)).addErrback(self._ebGeneral)
d = defer.gatherResults([d1, self.loopback()])
return d.addCallback(self._cbTestLogin)
def _cbTestLogin(self, ignored):
self.assertEqual(self.server.account, SimpleServer.theAccount)
self.assertEqual(self.server.state, 'auth')
def testFailedLogin(self):
def login():
d = self.client.login(b'testuser', b'wrong-password')
d.addBoth(self._cbStopClient)
d1 = self.connected.addCallback(strip(login)).addErrback(self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
return d.addCallback(self._cbTestFailedLogin)
def _cbTestFailedLogin(self, ignored):
self.assertEqual(self.server.account, None)
self.assertEqual(self.server.state, 'unauth')
def testLoginRequiringQuoting(self):
self.server._username = '{test}user'
self.server._password = '{test}password'
def login():
d = self.client.login('{test}user', '{test}password')
d.addBoth(self._cbStopClient)
d1 = self.connected.addCallback(strip(login)).addErrback(self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
return d.addCallback(self._cbTestLoginRequiringQuoting)
def _cbTestLoginRequiringQuoting(self, ignored):
self.assertEqual(self.server.account, SimpleServer.theAccount)
self.assertEqual(self.server.state, 'auth')
def testNamespace(self):
self.namespaceArgs = None
def login():
return self.client.login(b'testuser', b'password-test')
def namespace():
def gotNamespace(args):
self.namespaceArgs = args
self._cbStopClient(None)
return self.client.namespace().addCallback(gotNamespace)
d1 = self.connected.addCallback(strip(login))
d1.addCallback(strip(namespace))
d1.addErrback(self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
d.addCallback(lambda _: self.assertEqual(self.namespaceArgs,
[[['', '/']], [], []]))
return d
def testSelect(self):
SimpleServer.theAccount.addMailbox('test-mailbox')
self.selectedArgs = None
def login():
return self.client.login(b'testuser', b'password-test')
def select():
def selected(args):
self.selectedArgs = args
self._cbStopClient(None)
d = self.client.select('test-mailbox')
d.addCallback(selected)
return d
d1 = self.connected.addCallback(strip(login))
d1.addCallback(strip(select))
d1.addErrback(self._ebGeneral)
d2 = self.loopback()
return defer.gatherResults([d1, d2]).addCallback(self._cbTestSelect)
def _cbTestSelect(self, ignored):
mbox = SimpleServer.theAccount.mailboxes['TEST-MAILBOX']
self.assertEqual(self.server.mbox, mbox)
self.assertEqual(self.selectedArgs, {
'EXISTS': 9, 'RECENT': 3, 'UIDVALIDITY': 42,
'FLAGS': ('\\Flag1', 'Flag2', '\\AnotherSysFlag', 'LastFlag'),
'READ-WRITE': 1
})
def test_examine(self):
"""
L{IMAP4Client.examine} issues an I{EXAMINE} command to the server and
returns a L{Deferred} which fires with a C{dict} with as many of the
following keys as the server includes in its response: C{'FLAGS'},
C{'EXISTS'}, C{'RECENT'}, C{'UNSEEN'}, C{'READ-WRITE'}, C{'READ-ONLY'},
C{'UIDVALIDITY'}, and C{'PERMANENTFLAGS'}.
Unfortunately the server doesn't generate all of these so it's hard to
test the client's handling of them here. See
L{IMAP4ClientExamineTests} below.
See U{RFC 3501}, section 6.3.2,
for details.
"""
SimpleServer.theAccount.addMailbox('test-mailbox')
self.examinedArgs = None
def login():
return self.client.login(b'testuser', b'password-test')
def examine():
def examined(args):
self.examinedArgs = args
self._cbStopClient(None)
d = self.client.examine('test-mailbox')
d.addCallback(examined)
return d
d1 = self.connected.addCallback(strip(login))
d1.addCallback(strip(examine))
d1.addErrback(self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
return d.addCallback(self._cbTestExamine)
def _cbTestExamine(self, ignored):
mbox = SimpleServer.theAccount.mailboxes['TEST-MAILBOX']
self.assertEqual(self.server.mbox, mbox)
self.assertEqual(self.examinedArgs, {
'EXISTS': 9, 'RECENT': 3, 'UIDVALIDITY': 42,
'FLAGS': ('\\Flag1', 'Flag2', '\\AnotherSysFlag', 'LastFlag'),
'READ-WRITE': False})
def testCreate(self):
succeed = ('testbox', 'test/box', 'test/', 'test/box/box', 'INBOX')
fail = ('testbox', 'test/box')
def cb(): self.result.append(1)
def eb(failure): self.result.append(0)
def login():
return self.client.login(b'testuser', b'password-test')
def create():
for name in succeed + fail:
d = self.client.create(name)
d.addCallback(strip(cb)).addErrback(eb)
d.addCallbacks(self._cbStopClient, self._ebGeneral)
self.result = []
d1 = self.connected.addCallback(strip(login)).addCallback(strip(create))
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
return d.addCallback(self._cbTestCreate, succeed, fail)
def _cbTestCreate(self, ignored, succeed, fail):
self.assertEqual(self.result, [1] * len(succeed) + [0] * len(fail))
mbox = SimpleServer.theAccount.mailboxes.keys()
answers = ['inbox', 'testbox', 'test/box', 'test', 'test/box/box']
mbox.sort()
answers.sort()
self.assertEqual(mbox, [a.upper() for a in answers])
def testDelete(self):
SimpleServer.theAccount.addMailbox('delete/me')
def login():
return self.client.login(b'testuser', b'password-test')
def delete():
return self.client.delete('delete/me')
d1 = self.connected.addCallback(strip(login))
d1.addCallbacks(strip(delete), self._ebGeneral)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
d.addCallback(lambda _:
self.assertEqual(SimpleServer.theAccount.mailboxes.keys(), []))
return d
def testIllegalInboxDelete(self):
self.stashed = None
def login():
return self.client.login(b'testuser', b'password-test')
def delete():
return self.client.delete('inbox')
def stash(result):
self.stashed = result
d1 = self.connected.addCallback(strip(login))
d1.addCallbacks(strip(delete), self._ebGeneral)
d1.addBoth(stash)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
d.addCallback(lambda _: self.assertTrue(isinstance(self.stashed,
failure.Failure)))
return d
def testNonExistentDelete(self):
def login():
return self.client.login(b'testuser', b'password-test')
def delete():
return self.client.delete('delete/me')
def deleteFailed(failure):
self.failure = failure
self.failure = None
d1 = self.connected.addCallback(strip(login))
d1.addCallback(strip(delete)).addErrback(deleteFailed)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
d.addCallback(lambda _: self.assertEqual(str(self.failure.value),
'No such mailbox'))
return d
def testIllegalDelete(self):
m = SimpleMailbox()
m.flags = (r'\Noselect',)
SimpleServer.theAccount.addMailbox('delete', m)
SimpleServer.theAccount.addMailbox('delete/me')
def login():
return self.client.login(b'testuser', b'password-test')
def delete():
return self.client.delete('delete')
def deleteFailed(failure):
self.failure = failure
self.failure = None
d1 = self.connected.addCallback(strip(login))
d1.addCallback(strip(delete)).addErrback(deleteFailed)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
expected = "Hierarchically inferior mailboxes exist and \\Noselect is set"
d.addCallback(lambda _:
self.assertEqual(str(self.failure.value), expected))
return d
def testRename(self):
SimpleServer.theAccount.addMailbox(b'oldmbox')
def login():
return self.client.login(b'testuser', b'password-test')
def rename():
return self.client.rename(b'oldmbox', b'newname')
d1 = self.connected.addCallback(strip(login))
d1.addCallbacks(strip(rename), self._ebGeneral)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
d.addCallback(lambda _:
self.assertEqual(SimpleServer.theAccount.mailboxes.keys(),
[b'NEWNAME']))
return d
def testIllegalInboxRename(self):
self.stashed = None
def login():
return self.client.login(b'testuser', b'password-test')
def rename():
return self.client.rename('inbox', 'frotz')
def stash(stuff):
self.stashed = stuff
d1 = self.connected.addCallback(strip(login))
d1.addCallbacks(strip(rename), self._ebGeneral)
d1.addBoth(stash)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
d.addCallback(lambda _:
self.assertTrue(isinstance(self.stashed, failure.Failure)))
return d
def testHierarchicalRename(self):
SimpleServer.theAccount.create(b'oldmbox/m1')
SimpleServer.theAccount.create(b'oldmbox/m2')
def login():
return self.client.login(b'testuser', b'password-test')
def rename():
return self.client.rename(b'oldmbox', b'newname')
d1 = self.connected.addCallback(strip(login))
d1.addCallbacks(strip(rename), self._ebGeneral)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
return d.addCallback(self._cbTestHierarchicalRename)
def _cbTestHierarchicalRename(self, ignored):
mboxes = SimpleServer.theAccount.mailboxes.keys()
expected = [b'newname', b'newname/m1', b'newname/m2']
mboxes = list(sorted(mboxes))
self.assertEqual(mboxes, [s.upper() for s in expected])
def testSubscribe(self):
def login():
return self.client.login(b'testuser', b'password-test')
def subscribe():
return self.client.subscribe('this/mbox')
d1 = self.connected.addCallback(strip(login))
d1.addCallbacks(strip(subscribe), self._ebGeneral)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
d.addCallback(lambda _:
self.assertEqual(SimpleServer.theAccount.subscriptions,
['THIS/MBOX']))
return d
def testUnsubscribe(self):
SimpleServer.theAccount.subscriptions = ['THIS/MBOX', 'THAT/MBOX']
def login():
return self.client.login(b'testuser', b'password-test')
def unsubscribe():
return self.client.unsubscribe('this/mbox')
d1 = self.connected.addCallback(strip(login))
d1.addCallbacks(strip(unsubscribe), self._ebGeneral)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
d.addCallback(lambda _:
self.assertEqual(SimpleServer.theAccount.subscriptions,
['THAT/MBOX']))
return d
def _listSetup(self, f):
SimpleServer.theAccount.addMailbox('root/subthing')
SimpleServer.theAccount.addMailbox('root/another-thing')
SimpleServer.theAccount.addMailbox('non-root/subthing')
def login():
return self.client.login(b'testuser', b'password-test')
def listed(answers):
self.listed = answers
self.listed = None
d1 = self.connected.addCallback(strip(login))
d1.addCallbacks(strip(f), self._ebGeneral)
d1.addCallbacks(listed, self._ebGeneral)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
return defer.gatherResults([d1, d2]).addCallback(lambda _: self.listed)
def testList(self):
def mailboxList():
return self.client.list('root', '%')
d = self._listSetup(mailboxList)
d.addCallback(lambda listed: self.assertEqual(
sortNest(listed),
sortNest([
(SimpleMailbox.flags, "/", "ROOT/SUBTHING"),
(SimpleMailbox.flags, "/", "ROOT/ANOTHER-THING")
])
))
return d
def testLSub(self):
SimpleServer.theAccount.subscribe('ROOT/SUBTHING')
def lsub():
return self.client.lsub('root', '%')
d = self._listSetup(lsub)
d.addCallback(self.assertEqual,
[(SimpleMailbox.flags, "/", "ROOT/SUBTHING")])
return d
def testStatus(self):
SimpleServer.theAccount.addMailbox('root/subthing')
def login():
return self.client.login(b'testuser', b'password-test')
def status():
return self.client.status('root/subthing', 'MESSAGES', 'UIDNEXT', 'UNSEEN')
def statused(result):
self.statused = result
self.statused = None
d1 = self.connected.addCallback(strip(login))
d1.addCallbacks(strip(status), self._ebGeneral)
d1.addCallbacks(statused, self._ebGeneral)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
d.addCallback(lambda _: self.assertEqual(
self.statused,
{'MESSAGES': 9, 'UIDNEXT': '10', 'UNSEEN': 4}
))
return d
def testFailedStatus(self):
def login():
return self.client.login(b'testuser', b'password-test')
def status():
return self.client.status('root/nonexistent', 'MESSAGES', 'UIDNEXT', 'UNSEEN')
def statused(result):
self.statused = result
def failed(failure):
self.failure = failure
self.statused = self.failure = None
d1 = self.connected.addCallback(strip(login))
d1.addCallbacks(strip(status), self._ebGeneral)
d1.addCallbacks(statused, failed)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
return defer.gatherResults([d1, d2]).addCallback(self._cbTestFailedStatus)
def _cbTestFailedStatus(self, ignored):
self.assertEqual(
self.statused, None
)
self.assertEqual(
self.failure.value.args,
('Could not open mailbox',)
)
def testFullAppend(self):
infile = util.sibpath(__file__, 'rfc822.message')
message = open(infile)
SimpleServer.theAccount.addMailbox('root/subthing')
def login():
return self.client.login(b'testuser', b'password-test')
def append():
return self.client.append(
'root/subthing',
message,
('\\SEEN', '\\DELETED'),
'Tue, 17 Jun 2003 11:22:16 -0600 (MDT)',
)
d1 = self.connected.addCallback(strip(login))
d1.addCallbacks(strip(append), self._ebGeneral)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
return d.addCallback(self._cbTestFullAppend, infile)
def _cbTestFullAppend(self, ignored, infile):
mb = SimpleServer.theAccount.mailboxes['ROOT/SUBTHING']
self.assertEqual(1, len(mb.messages))
self.assertEqual(
(['\\SEEN', '\\DELETED'], 'Tue, 17 Jun 2003 11:22:16 -0600 (MDT)', 0),
mb.messages[0][1:]
)
with open(infile) as f:
self.assertEqual(f.read(), mb.messages[0][0].getvalue())
def testPartialAppend(self):
infile = util.sibpath(__file__, 'rfc822.message')
SimpleServer.theAccount.addMailbox('PARTIAL/SUBTHING')
def login():
return self.client.login(b'testuser', b'password-test')
def append():
message = open(infile)
return self.client.sendCommand(
imap4.Command(
'APPEND',
'PARTIAL/SUBTHING (\\SEEN) "Right now" {%d}' % os.path.getsize(infile),
(), self.client._IMAP4Client__cbContinueAppend, message
)
)
d1 = self.connected.addCallback(strip(login))
d1.addCallbacks(strip(append), self._ebGeneral)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
return d.addCallback(self._cbTestPartialAppend, infile)
def _cbTestPartialAppend(self, ignored, infile):
mb = SimpleServer.theAccount.mailboxes['PARTIAL/SUBTHING']
self.assertEqual(1, len(mb.messages))
self.assertEqual(
(['\\SEEN'], 'Right now', 0),
mb.messages[0][1:]
)
with open(infile) as f:
self.assertEqual(f.read(), mb.messages[0][0].getvalue())
def testCheck(self):
SimpleServer.theAccount.addMailbox(b'root/subthing')
def login():
return self.client.login(b'testuser', b'password-test')
def select():
return self.client.select(b'root/subthing')
def check():
return self.client.check()
d = self.connected.addCallback(strip(login))
d.addCallbacks(strip(select), self._ebGeneral)
d.addCallbacks(strip(check), self._ebGeneral)
d.addCallbacks(self._cbStopClient, self._ebGeneral)
return self.loopback()
# Okay, that was fun
def testClose(self):
m = SimpleMailbox()
m.messages = [
('Message 1', ('\\Deleted', 'AnotherFlag'), None, 0),
('Message 2', ('AnotherFlag',), None, 1),
('Message 3', ('\\Deleted',), None, 2),
]
SimpleServer.theAccount.addMailbox('mailbox', m)
def login():
return self.client.login(b'testuser', b'password-test')
def select():
return self.client.select(b'mailbox')
def close():
return self.client.close()
d = self.connected.addCallback(strip(login))
d.addCallbacks(strip(select), self._ebGeneral)
d.addCallbacks(strip(close), self._ebGeneral)
d.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
return defer.gatherResults([d, d2]).addCallback(self._cbTestClose, m)
def _cbTestClose(self, ignored, m):
self.assertEqual(len(m.messages), 1)
self.assertEqual(m.messages[0], ('Message 2', ('AnotherFlag',), None, 1))
self.assertTrue(m.closed)
def testExpunge(self):
m = SimpleMailbox()
m.messages = [
('Message 1', ('\\Deleted', 'AnotherFlag'), None, 0),
('Message 2', ('AnotherFlag',), None, 1),
('Message 3', ('\\Deleted',), None, 2),
]
SimpleServer.theAccount.addMailbox('mailbox', m)
def login():
return self.client.login(b'testuser', b'password-test')
def select():
return self.client.select('mailbox')
def expunge():
return self.client.expunge()
def expunged(results):
self.assertFalse(self.server.mbox is None)
self.results = results
self.results = None
d1 = self.connected.addCallback(strip(login))
d1.addCallbacks(strip(select), self._ebGeneral)
d1.addCallbacks(strip(expunge), self._ebGeneral)
d1.addCallbacks(expunged, self._ebGeneral)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
return d.addCallback(self._cbTestExpunge, m)
def _cbTestExpunge(self, ignored, m):
self.assertEqual(len(m.messages), 1)
self.assertEqual(m.messages[0], ('Message 2', ('AnotherFlag',), None, 1))
self.assertEqual(self.results, [0, 2])
class IMAP4ServerSearchTests(IMAP4HelperMixin, unittest.TestCase):
"""
Tests for the behavior of the search_* functions in L{imap4.IMAP4Server}.
"""
def setUp(self):
IMAP4HelperMixin.setUp(self)
self.earlierQuery = ["10-Dec-2009"]
self.sameDateQuery = ["13-Dec-2009"]
self.laterQuery = ["16-Dec-2009"]
self.seq = 0
self.msg = FakeyMessage({"date" : "Mon, 13 Dec 2009 21:25:10 GMT"}, [],
'13 Dec 2009 00:00:00 GMT', '', 1234, None)
def test_searchSentBefore(self):
"""
L{imap4.IMAP4Server.search_SENTBEFORE} returns True if the message date
is earlier than the query date.
"""
self.assertFalse(
self.server.search_SENTBEFORE(self.earlierQuery, self.seq, self.msg))
self.assertTrue(
self.server.search_SENTBEFORE(self.laterQuery, self.seq, self.msg))
def test_searchWildcard(self):
"""
L{imap4.IMAP4Server.search_UID} returns True if the message UID is in
the search range.
"""
self.assertFalse(
self.server.search_UID(['2:3'], self.seq, self.msg, (1, 1234)))
# 2:* should get translated to 2: and then to 1:2
self.assertTrue(
self.server.search_UID(['2:*'], self.seq, self.msg, (1, 1234)))
self.assertTrue(
self.server.search_UID(['*'], self.seq, self.msg, (1, 1234)))
def test_searchWildcardHigh(self):
"""
L{imap4.IMAP4Server.search_UID} should return True if there is a
wildcard, because a wildcard means "highest UID in the mailbox".
"""
self.assertTrue(
self.server.search_UID(['1235:*'], self.seq, self.msg, (1234, 1)))
def test_reversedSearchTerms(self):
"""
L{imap4.IMAP4Server.search_SENTON} returns True if the message date is
the same as the query date.
"""
msgset = imap4.parseIdList('4:2')
self.assertEqual(list(msgset), [2, 3, 4])
def test_searchSentOn(self):
"""
L{imap4.IMAP4Server.search_SENTON} returns True if the message date is
the same as the query date.
"""
self.assertFalse(
self.server.search_SENTON(self.earlierQuery, self.seq, self.msg))
self.assertTrue(
self.server.search_SENTON(self.sameDateQuery, self.seq, self.msg))
self.assertFalse(
self.server.search_SENTON(self.laterQuery, self.seq, self.msg))
def test_searchSentSince(self):
"""
L{imap4.IMAP4Server.search_SENTSINCE} returns True if the message date
is later than the query date.
"""
self.assertTrue(
self.server.search_SENTSINCE(self.earlierQuery, self.seq, self.msg))
self.assertFalse(
self.server.search_SENTSINCE(self.laterQuery, self.seq, self.msg))
def test_searchOr(self):
"""
L{imap4.IMAP4Server.search_OR} returns true if either of the two
expressions supplied to it returns true and returns false if neither
does.
"""
self.assertTrue(
self.server.search_OR(
["SENTSINCE"] + self.earlierQuery +
["SENTSINCE"] + self.laterQuery,
self.seq, self.msg, (None, None)))
self.assertTrue(
self.server.search_OR(
["SENTSINCE"] + self.laterQuery +
["SENTSINCE"] + self.earlierQuery,
self.seq, self.msg, (None, None)))
self.assertFalse(
self.server.search_OR(
["SENTON"] + self.laterQuery +
["SENTSINCE"] + self.laterQuery,
self.seq, self.msg, (None, None)))
def test_searchNot(self):
"""
L{imap4.IMAP4Server.search_NOT} returns the negation of the result
of the expression supplied to it.
"""
self.assertFalse(self.server.search_NOT(
["SENTSINCE"] + self.earlierQuery, self.seq, self.msg,
(None, None)))
self.assertTrue(self.server.search_NOT(
["SENTON"] + self.laterQuery, self.seq, self.msg,
(None, None)))
def test_searchBefore(self):
"""
L{imap4.IMAP4Server.search_BEFORE} returns True if the
internal message date is before the query date.
"""
self.assertFalse(
self.server.search_BEFORE(self.earlierQuery, self.seq, self.msg))
self.assertFalse(
self.server.search_BEFORE(self.sameDateQuery, self.seq, self.msg))
self.assertTrue(
self.server.search_BEFORE(self.laterQuery, self.seq, self.msg))
def test_searchOn(self):
"""
L{imap4.IMAP4Server.search_ON} returns True if the
internal message date is the same as the query date.
"""
self.assertFalse(
self.server.search_ON(self.earlierQuery, self.seq, self.msg))
self.assertFalse(
self.server.search_ON(self.sameDateQuery, self.seq, self.msg))
self.assertFalse(
self.server.search_ON(self.laterQuery, self.seq, self.msg))
def test_searchSince(self):
"""
L{imap4.IMAP4Server.search_SINCE} returns True if the
internal message date is greater than the query date.
"""
self.assertTrue(
self.server.search_SINCE(self.earlierQuery, self.seq, self.msg))
self.assertTrue(
self.server.search_SINCE(self.sameDateQuery, self.seq, self.msg))
self.assertFalse(
self.server.search_SINCE(self.laterQuery, self.seq, self.msg))
class TestRealm:
theAccount = None
def requestAvatar(self, avatarId, mind, *interfaces):
return imap4.IAccount, self.theAccount, lambda: None
class TestChecker:
credentialInterfaces = (IUsernameHashedPassword, IUsernamePassword)
users = {
b'testuser': b'secret'
}
def requestAvatarId(self, credentials):
if credentials.username in self.users:
return defer.maybeDeferred(
credentials.checkPassword, self.users[credentials.username]
).addCallback(self._cbCheck, credentials.username)
def _cbCheck(self, result, username):
if result:
return username
raise UnauthorizedLogin()
class AuthenticatorTests(IMAP4HelperMixin, unittest.TestCase):
def setUp(self):
IMAP4HelperMixin.setUp(self)
realm = TestRealm()
realm.theAccount = Account(b'testuser')
portal = Portal(realm)
portal.registerChecker(TestChecker())
self.server.portal = portal
self.authenticated = 0
self.account = realm.theAccount
def testCramMD5(self):
self.server.challengers[b'CRAM-MD5'] = CramMD5Credentials
cAuth = imap4.CramMD5ClientAuthenticator(b'testuser')
self.client.registerAuthenticator(cAuth)
def auth():
return self.client.authenticate(b'secret')
def authed():
self.authenticated = 1
d1 = self.connected.addCallback(strip(auth))
d1.addCallbacks(strip(authed), self._ebGeneral)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d2 = self.loopback()
d = defer.gatherResults([d1, d2])
return d.addCallback(self._cbTestCramMD5)
def _cbTestCramMD5(self, ignored):
self.assertEqual(self.authenticated, 1)
self.assertEqual(self.server.account, self.account)
def testFailedCramMD5(self):
self.server.challengers[b'CRAM-MD5'] = CramMD5Credentials
cAuth = imap4.CramMD5ClientAuthenticator(b'testuser')
self.client.registerAuthenticator(cAuth)
def misauth():
return self.client.authenticate(b'not the secret')
def authed():
self.authenticated = 1
def misauthed():
self.authenticated = -1
d1 = self.connected.addCallback(strip(misauth))
d1.addCallbacks(strip(authed), strip(misauthed))
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
return d.addCallback(self._cbTestFailedCramMD5)
def _cbTestFailedCramMD5(self, ignored):
self.assertEqual(self.authenticated, -1)
self.assertEqual(self.server.account, None)
def testLOGIN(self):
self.server.challengers[b'LOGIN'] = imap4.LOGINCredentials
cAuth = imap4.LOGINAuthenticator(b'testuser')
self.client.registerAuthenticator(cAuth)
def auth():
return self.client.authenticate(b'secret')
def authed():
self.authenticated = 1
d1 = self.connected.addCallback(strip(auth))
d1.addCallbacks(strip(authed), self._ebGeneral)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
return d.addCallback(self._cbTestLOGIN)
def _cbTestLOGIN(self, ignored):
self.assertEqual(self.authenticated, 1)
self.assertEqual(self.server.account, self.account)
def testFailedLOGIN(self):
self.server.challengers[b'LOGIN'] = imap4.LOGINCredentials
cAuth = imap4.LOGINAuthenticator(b'testuser')
self.client.registerAuthenticator(cAuth)
def misauth():
return self.client.authenticate(b'not the secret')
def authed():
self.authenticated = 1
def misauthed():
self.authenticated = -1
d1 = self.connected.addCallback(strip(misauth))
d1.addCallbacks(strip(authed), strip(misauthed))
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
return d.addCallback(self._cbTestFailedLOGIN)
def _cbTestFailedLOGIN(self, ignored):
self.assertEqual(self.authenticated, -1)
self.assertEqual(self.server.account, None)
def testPLAIN(self):
self.server.challengers[b'PLAIN'] = imap4.PLAINCredentials
cAuth = imap4.PLAINAuthenticator(b'testuser')
self.client.registerAuthenticator(cAuth)
def auth():
return self.client.authenticate(b'secret')
def authed():
self.authenticated = 1
d1 = self.connected.addCallback(strip(auth))
d1.addCallbacks(strip(authed), self._ebGeneral)
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
return d.addCallback(self._cbTestPLAIN)
def _cbTestPLAIN(self, ignored):
self.assertEqual(self.authenticated, 1)
self.assertEqual(self.server.account, self.account)
def testFailedPLAIN(self):
self.server.challengers[b'PLAIN'] = imap4.PLAINCredentials
cAuth = imap4.PLAINAuthenticator(b'testuser')
self.client.registerAuthenticator(cAuth)
def misauth():
return self.client.authenticate(b'not the secret')
def authed():
self.authenticated = 1
def misauthed():
self.authenticated = -1
d1 = self.connected.addCallback(strip(misauth))
d1.addCallbacks(strip(authed), strip(misauthed))
d1.addCallbacks(self._cbStopClient, self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
return d.addCallback(self._cbTestFailedPLAIN)
def _cbTestFailedPLAIN(self, ignored):
self.assertEqual(self.authenticated, -1)
self.assertEqual(self.server.account, None)
class SASLPLAINTests(unittest.TestCase):
"""
Tests for I{SASL PLAIN} authentication, as implemented by
L{imap4.PLAINAuthenticator} and L{imap4.PLAINCredentials}.
@see: U{http://www.faqs.org/rfcs/rfc2595.html}
@see: U{http://www.faqs.org/rfcs/rfc4616.html}
"""
def test_authenticatorChallengeResponse(self):
"""
L{PLAINAuthenticator.challengeResponse} returns challenge strings of
the form::
NULNUL
"""
username = b'testuser'
secret = b'secret'
chal = b'challenge'
cAuth = imap4.PLAINAuthenticator(username)
response = cAuth.challengeResponse(secret, chal)
self.assertEqual(response, b'\0' + username + b'\0' + secret)
def test_credentialsSetResponse(self):
"""
L{PLAINCredentials.setResponse} parses challenge strings of the
form::
NULNUL
"""
cred = imap4.PLAINCredentials()
cred.setResponse(b'\0testuser\0secret')
self.assertEqual(cred.username, b'testuser')
self.assertEqual(cred.password, b'secret')
def test_credentialsInvalidResponse(self):
"""
L{PLAINCredentials.setResponse} raises L{imap4.IllegalClientResponse}
when passed a string not of the expected form.
"""
cred = imap4.PLAINCredentials()
self.assertRaises(
imap4.IllegalClientResponse, cred.setResponse, b'hello')
self.assertRaises(
imap4.IllegalClientResponse, cred.setResponse, b'hello\0world')
self.assertRaises(
imap4.IllegalClientResponse, cred.setResponse,
b'hello\0world\0Zoom!\0')
class UnsolicitedResponseTests(IMAP4HelperMixin, unittest.TestCase):
def testReadWrite(self):
def login():
return self.client.login(b'testuser', b'password-test')
def loggedIn():
self.server.modeChanged(1)
d1 = self.connected.addCallback(strip(login))
d1.addCallback(strip(loggedIn)).addErrback(self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
return d.addCallback(self._cbTestReadWrite)
def _cbTestReadWrite(self, ignored):
E = self.client.events
self.assertEqual(E, [[b'modeChanged', 1]])
def testReadOnly(self):
def login():
return self.client.login(b'testuser', b'password-test')
def loggedIn():
self.server.modeChanged(0)
d1 = self.connected.addCallback(strip(login))
d1.addCallback(strip(loggedIn)).addErrback(self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
return d.addCallback(self._cbTestReadOnly)
def _cbTestReadOnly(self, ignored):
E = self.client.events
self.assertEqual(E, [[b'modeChanged', 0]])
def testFlagChange(self):
flags = {
1: [b'\\Answered', b'\\Deleted'],
5: [],
10: [b'\\Recent']
}
def login():
return self.client.login(b'testuser', b'password-test')
def loggedIn():
self.server.flagsChanged(flags)
d1 = self.connected.addCallback(strip(login))
d1.addCallback(strip(loggedIn)).addErrback(self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
return d.addCallback(self._cbTestFlagChange, flags)
def _cbTestFlagChange(self, ignored, flags):
E = self.client.events
expect = [[b'flagsChanged', {x[0]: x[1]}] for x in flags.items()]
E.sort()
expect.sort()
self.assertEqual(E, expect)
def testNewMessages(self):
def login():
return self.client.login(b'testuser', b'password-test')
def loggedIn():
self.server.newMessages(10, None)
d1 = self.connected.addCallback(strip(login))
d1.addCallback(strip(loggedIn)).addErrback(self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
return d.addCallback(self._cbTestNewMessages)
def _cbTestNewMessages(self, ignored):
E = self.client.events
self.assertEqual(E, [[b'newMessages', 10, None]])
def testNewRecentMessages(self):
def login():
return self.client.login(b'testuser', b'password-test')
def loggedIn():
self.server.newMessages(None, 10)
d1 = self.connected.addCallback(strip(login))
d1.addCallback(strip(loggedIn)).addErrback(self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
return d.addCallback(self._cbTestNewRecentMessages)
def _cbTestNewRecentMessages(self, ignored):
E = self.client.events
self.assertEqual(E, [[b'newMessages', None, 10]])
def testNewMessagesAndRecent(self):
def login():
return self.client.login(b'testuser', b'password-test')
def loggedIn():
self.server.newMessages(20, 10)
d1 = self.connected.addCallback(strip(login))
d1.addCallback(strip(loggedIn)).addErrback(self._ebGeneral)
d = defer.gatherResults([self.loopback(), d1])
return d.addCallback(self._cbTestNewMessagesAndRecent)
def _cbTestNewMessagesAndRecent(self, ignored):
E = self.client.events
self.assertEqual(E, [[b'newMessages', 20, None], [b'newMessages', None, 10]])
class ClientCapabilityTests(unittest.TestCase):
"""
Tests for issuance of the CAPABILITY command and handling of its response.
"""
def setUp(self):
"""
Create an L{imap4.IMAP4Client} connected to a L{StringTransport}.
"""
self.transport = StringTransport()
self.protocol = imap4.IMAP4Client()
self.protocol.makeConnection(self.transport)
self.protocol.dataReceived(b'* OK [IMAP4rev1]\r\n')
def test_simpleAtoms(self):
"""
A capability response consisting only of atoms without C{'='} in them
should result in a dict mapping those atoms to L{None}.
"""
capabilitiesResult = self.protocol.getCapabilities(useCache=False)
self.protocol.dataReceived(b'* CAPABILITY IMAP4rev1 LOGINDISABLED\r\n')
self.protocol.dataReceived(b'0001 OK Capability completed.\r\n')
def gotCapabilities(capabilities):
self.assertEqual(
capabilities, {b'IMAP4rev1': None, b'LOGINDISABLED': None})
capabilitiesResult.addCallback(gotCapabilities)
return capabilitiesResult
def test_categoryAtoms(self):
"""
A capability response consisting of atoms including C{'='} should have
those atoms split on that byte and have capabilities in the same
category aggregated into lists in the resulting dictionary.
(n.b. - I made up the word "category atom"; the protocol has no notion
of structure here, but rather allows each capability to define the
semantics of its entry in the capability response in a freeform manner.
If I had realized this earlier, the API for capabilities would look
different. As it is, we can hope that no one defines any crazy
semantics which are incompatible with this API, or try to figure out a
better API when someone does. -exarkun)
"""
capabilitiesResult = self.protocol.getCapabilities(useCache=False)
self.protocol.dataReceived(b'* CAPABILITY IMAP4rev1 AUTH=LOGIN AUTH=PLAIN\r\n')
self.protocol.dataReceived(b'0001 OK Capability completed.\r\n')
def gotCapabilities(capabilities):
self.assertEqual(
capabilities, {b'IMAP4rev1': None, b'AUTH': [b'LOGIN', b'PLAIN']})
capabilitiesResult.addCallback(gotCapabilities)
return capabilitiesResult
def test_mixedAtoms(self):
"""
A capability response consisting of both simple and category atoms of
the same type should result in a list containing L{None} as well as the
values for the category.
"""
capabilitiesResult = self.protocol.getCapabilities(useCache=False)
# Exercise codepath for both orderings of =-having and =-missing
# capabilities.
self.protocol.dataReceived(
b'* CAPABILITY IMAP4rev1 FOO FOO=BAR BAR=FOO BAR\r\n')
self.protocol.dataReceived(b'0001 OK Capability completed.\r\n')
def gotCapabilities(capabilities):
self.assertEqual(capabilities, {b'IMAP4rev1': None,
b'FOO': [None, b'BAR'],
b'BAR': [b'FOO', None]})
capabilitiesResult.addCallback(gotCapabilities)
return capabilitiesResult
class StillSimplerClient(imap4.IMAP4Client):
"""
An IMAP4 client which keeps track of unsolicited flag changes.
"""
def __init__(self):
imap4.IMAP4Client.__init__(self)
self.flags = {}
def flagsChanged(self, newFlags):
self.flags.update(newFlags)
class HandCraftedTests(IMAP4HelperMixin, unittest.TestCase):
def testTrailingLiteral(self):
transport = StringTransport()
c = imap4.IMAP4Client()
c.makeConnection(transport)
c.lineReceived(b'* OK [IMAP4rev1]')
def cbSelect(ignored):
d = c.fetchMessage('1')
c.dataReceived(b'* 1 FETCH (RFC822 {10}\r\n0123456789\r\n RFC822.SIZE 10)\r\n')
c.dataReceived(b'0003 OK FETCH\r\n')
return d
def cbLogin(ignored):
d = c.select('inbox')
c.lineReceived(b'0002 OK SELECT')
d.addCallback(cbSelect)
return d
d = c.login('blah', 'blah')
c.dataReceived(b'0001 OK LOGIN\r\n')
d.addCallback(cbLogin)
return d
def testPathelogicalScatteringOfLiterals(self):
self.server.checker.addUser(b'testuser', b'password-test')
transport = StringTransport()
self.server.makeConnection(transport)
transport.clear()
self.server.dataReceived(b"01 LOGIN {8}\r\n")
self.assertEqual(transport.value(), b"+ Ready for 8 octets of text\r\n")
transport.clear()
self.server.dataReceived(b"testuser {13}\r\n")
self.assertEqual(transport.value(), b"+ Ready for 13 octets of text\r\n")
transport.clear()
self.server.dataReceived(b"password-test\r\n")
self.assertEqual(transport.value(), b"01 OK LOGIN succeeded\r\n")
self.assertEqual(self.server.state, 'auth')
self.server.connectionLost(error.ConnectionDone("Connection done."))
def test_unsolicitedResponseMixedWithSolicitedResponse(self):
"""
If unsolicited data is received along with solicited data in the
response to a I{FETCH} command issued by L{IMAP4Client.fetchSpecific},
the unsolicited data is passed to the appropriate callback and not
included in the result with which the L{Deferred} returned by
L{IMAP4Client.fetchSpecific} fires.
"""
transport = StringTransport()
c = StillSimplerClient()
c.makeConnection(transport)
c.lineReceived(b'* OK [IMAP4rev1]')
def login():
d = c.login('blah', 'blah')
c.dataReceived(b'0001 OK LOGIN\r\n')
return d
def select():
d = c.select('inbox')
c.lineReceived(b'0002 OK SELECT')
return d
def fetch():
d = c.fetchSpecific(b'1:*',
headerType = b'HEADER.FIELDS',
headerArgs = [b'SUBJECT'])
c.dataReceived(b'* 1 FETCH (BODY[HEADER.FIELDS ("SUBJECT")] {38}\r\n')
c.dataReceived(b'Subject: Suprise for your woman...\r\n')
c.dataReceived(b'\r\n')
c.dataReceived(b')\r\n')
c.dataReceived(b'* 1 FETCH (FLAGS (\Seen))\r\n')
c.dataReceived(b'* 2 FETCH (BODY[HEADER.FIELDS ("SUBJECT")] {75}\r\n')
c.dataReceived(b'Subject: What you been doing. Order your meds here . ,. handcuff madsen\r\n')
c.dataReceived(b'\r\n')
c.dataReceived(b')\r\n')
c.dataReceived(b'0003 OK FETCH completed\r\n')
return d
def test(res):
self.assertEqual(res, {
1: [[b'BODY', ['HEADER.FIELDS', ['SUBJECT']],
b'Subject: Suprise for your woman...\r\n\r\n']],
2: [[b'BODY', [b'HEADER.FIELDS', [b'SUBJECT']],
b'Subject: What you been doing. Order your meds here . ,. handcuff madsen\r\n\r\n']]
})
self.assertEqual(c.flags, {1: ['\\Seen']})
return login(
).addCallback(strip(select)
).addCallback(strip(fetch)
).addCallback(test)
def test_literalWithoutPrecedingWhitespace(self):
"""
Literals should be recognized even when they are not preceded by
whitespace.
"""
transport = StringTransport()
protocol = imap4.IMAP4Client()
protocol.makeConnection(transport)
protocol.lineReceived(b'* OK [IMAP4rev1]')
def login():
d = protocol.login(b'blah', b'blah')
protocol.dataReceived(b'0001 OK LOGIN\r\n')
return d
def select():
d = protocol.select(b'inbox')
protocol.lineReceived(b'0002 OK SELECT')
return d
def fetch():
d = protocol.fetchSpecific(b'1:*',
headerType=b'HEADER.FIELDS',
headerArgs=[b'SUBJECT'])
protocol.dataReceived(
b'* 1 FETCH (BODY[HEADER.FIELDS ({7}\r\nSUBJECT)] "Hello")\r\n')
protocol.dataReceived(b'0003 OK FETCH completed\r\n')
return d
def test(result):
self.assertEqual(
result, {1: [[b'BODY', [b'HEADER.FIELDS', [b'SUBJECT']], b'Hello']]})
d = login()
d.addCallback(strip(select))
d.addCallback(strip(fetch))
d.addCallback(test)
return d
def test_nonIntegerLiteralLength(self):
"""
If the server sends a literal length which cannot be parsed as an
integer, L{IMAP4Client.lineReceived} should cause the protocol to be
disconnected by raising L{imap4.IllegalServerResponse}.
"""
transport = StringTransport()
protocol = imap4.IMAP4Client()
protocol.makeConnection(transport)
protocol.lineReceived(b'* OK [IMAP4rev1]')
def login():
d = protocol.login('blah', 'blah')
protocol.dataReceived(b'0001 OK LOGIN\r\n')
return d
def select():
d = protocol.select('inbox')
protocol.lineReceived(b'0002 OK SELECT')
return d
def fetch():
protocol.fetchSpecific(
b'1:*',
headerType=b'HEADER.FIELDS',
headerArgs=[b'SUBJECT'])
self.assertRaises(
imap4.IllegalServerResponse,
protocol.dataReceived,
b'* 1 FETCH {xyz}\r\n...')
d = login()
d.addCallback(strip(select))
d.addCallback(strip(fetch))
return d
def test_flagsChangedInsideFetchSpecificResponse(self):
"""
Any unrequested flag information received along with other requested
information in an untagged I{FETCH} received in response to a request
issued with L{IMAP4Client.fetchSpecific} is passed to the
C{flagsChanged} callback.
"""
transport = StringTransport()
c = StillSimplerClient()
c.makeConnection(transport)
c.lineReceived(b'* OK [IMAP4rev1]')
def login():
d = c.login('blah', 'blah')
c.dataReceived(b'0001 OK LOGIN\r\n')
return d
def select():
d = c.select('inbox')
c.lineReceived(b'0002 OK SELECT')
return d
def fetch():
d = c.fetchSpecific(b'1:*',
headerType=b'HEADER.FIELDS',
headerArgs=[b'SUBJECT'])
# This response includes FLAGS after the requested data.
c.dataReceived(b'* 1 FETCH (BODY[HEADER.FIELDS ("SUBJECT")] {22}\r\n')
c.dataReceived(b'Subject: subject one\r\n')
c.dataReceived(b' FLAGS (\\Recent))\r\n')
# And this one includes it before! Either is possible.
c.dataReceived(b'* 2 FETCH (FLAGS (\\Seen) BODY[HEADER.FIELDS ("SUBJECT")] {22}\r\n')
c.dataReceived(b'Subject: subject two\r\n')
c.dataReceived(b')\r\n')
c.dataReceived(b'0003 OK FETCH completed\r\n')
return d
def test(res):
self.assertEqual(res, {
1: [['BODY', ['HEADER.FIELDS', ['SUBJECT']],
'Subject: subject one\r\n']],
2: [['BODY', ['HEADER.FIELDS', ['SUBJECT']],
'Subject: subject two\r\n']]
})
self.assertEqual(c.flags, {1: ['\\Recent'], 2: ['\\Seen']})
return login(
).addCallback(strip(select)
).addCallback(strip(fetch)
).addCallback(test)
def test_flagsChangedInsideFetchMessageResponse(self):
"""
Any unrequested flag information received along with other requested
information in an untagged I{FETCH} received in response to a request
issued with L{IMAP4Client.fetchMessage} is passed to the
C{flagsChanged} callback.
"""
transport = StringTransport()
c = StillSimplerClient()
c.makeConnection(transport)
c.lineReceived(b'* OK [IMAP4rev1]')
def login():
d = c.login('blah', 'blah')
c.dataReceived(b'0001 OK LOGIN\r\n')
return d
def select():
d = c.select('inbox')
c.lineReceived(b'0002 OK SELECT')
return d
def fetch():
d = c.fetchMessage('1:*')
c.dataReceived(b'* 1 FETCH (RFC822 {24}\r\n')
c.dataReceived(b'Subject: first subject\r\n')
c.dataReceived(b' FLAGS (\Seen))\r\n')
c.dataReceived(b'* 2 FETCH (FLAGS (\Recent \Seen) RFC822 {25}\r\n')
c.dataReceived(b'Subject: second subject\r\n')
c.dataReceived(b')\r\n')
c.dataReceived(b'0003 OK FETCH completed\r\n')
return d
def test(res):
self.assertEqual(res, {
1: {'RFC822': 'Subject: first subject\r\n'},
2: {'RFC822': 'Subject: second subject\r\n'}})
self.assertEqual(
c.flags, {1: ['\\Seen'], 2: ['\\Recent', '\\Seen']})
return login(
).addCallback(strip(select)
).addCallback(strip(fetch)
).addCallback(test)
def test_authenticationChallengeDecodingException(self):
"""
When decoding a base64 encoded authentication message from the server,
decoding errors are logged and then the client closes the connection.
"""
transport = StringTransportWithDisconnection()
protocol = imap4.IMAP4Client()
transport.protocol = protocol
protocol.makeConnection(transport)
protocol.lineReceived(
b'* OK [CAPABILITY IMAP4rev1 IDLE NAMESPACE AUTH=CRAM-MD5] '
b'Twisted IMAP4rev1 Ready')
cAuth = imap4.CramMD5ClientAuthenticator(b'testuser')
protocol.registerAuthenticator(cAuth)
d = protocol.authenticate('secret')
# Should really be something describing the base64 decode error. See
# #6021.
self.assertFailure(d, error.ConnectionDone)
protocol.dataReceived(b'+ Something bad! and bad\r\n')
# This should not really be logged. See #6021.
logged = self.flushLoggedErrors(imap4.IllegalServerResponse)
self.assertEqual(len(logged), 1)
self.assertEqual(logged[0].value.args[0], "Something bad! and bad")
return d
class PreauthIMAP4ClientMixin:
"""
Mixin for L{unittest.TestCase} subclasses which provides a C{setUp} method
which creates an L{IMAP4Client} connected to a L{StringTransport} and puts
it into the I{authenticated} state.
@ivar transport: A L{StringTransport} to which C{client} is connected.
@ivar client: An L{IMAP4Client} which is connected to C{transport}.
"""
clientProtocol = imap4.IMAP4Client
def setUp(self):
"""
Create an IMAP4Client connected to a fake transport and in the
authenticated state.
"""
self.transport = StringTransport()
self.client = self.clientProtocol()
self.client.makeConnection(self.transport)
self.client.dataReceived(b'* PREAUTH Hello unittest\r\n')
def _extractDeferredResult(self, d):
"""
Synchronously extract the result of the given L{Deferred}. Fail the
test if that is not possible.
"""
result = []
error = []
d.addCallbacks(result.append, error.append)
if result:
return result[0]
elif error:
error[0].raiseException()
else:
self.fail("Expected result not available")
class SelectionTestsMixin(PreauthIMAP4ClientMixin):
"""
Mixin for test cases which defines tests which apply to both I{EXAMINE} and
I{SELECT} support.
"""
def _examineOrSelect(self):
"""
Issue either an I{EXAMINE} or I{SELECT} command (depending on
C{self.method}), assert that the correct bytes are written to the
transport, and return the L{Deferred} returned by whichever method was
called.
"""
d = getattr(self.client, self.method)('foobox')
self.assertEqual(
self.transport.value(), b'0001 ' + self.command + b' foobox\r\n')
return d
def _response(self, *lines):
"""
Deliver the given (unterminated) response lines to C{self.client} and
then deliver a tagged SELECT or EXAMINE completion line to finish the
SELECT or EXAMINE response.
"""
for line in lines:
self.client.dataReceived(line + b'\r\n')
self.client.dataReceived(
b'0001 OK [READ-ONLY] %s completed\r\n' % (self.command,))
def test_exists(self):
"""
If the server response to a I{SELECT} or I{EXAMINE} command includes an
I{EXISTS} response, the L{Deferred} return by L{IMAP4Client.select} or
L{IMAP4Client.examine} fires with a C{dict} including the value
associated with the C{'EXISTS'} key.
"""
d = self._examineOrSelect()
self._response('* 3 EXISTS')
self.assertEqual(
self._extractDeferredResult(d),
{'READ-WRITE': False, 'EXISTS': 3})
def test_nonIntegerExists(self):
"""
If the server returns a non-integer EXISTS value in its response to a
I{SELECT} or I{EXAMINE} command, the L{Deferred} returned by
L{IMAP4Client.select} or L{IMAP4Client.examine} fails with
L{IllegalServerResponse}.
"""
d = self._examineOrSelect()
self._response('* foo EXISTS')
self.assertRaises(
imap4.IllegalServerResponse, self._extractDeferredResult, d)
def test_recent(self):
"""
If the server response to a I{SELECT} or I{EXAMINE} command includes an
I{RECENT} response, the L{Deferred} return by L{IMAP4Client.select} or
L{IMAP4Client.examine} fires with a C{dict} including the value
associated with the C{'RECENT'} key.
"""
d = self._examineOrSelect()
self._response('* 5 RECENT')
self.assertEqual(
self._extractDeferredResult(d),
{'READ-WRITE': False, 'RECENT': 5})
def test_nonIntegerRecent(self):
"""
If the server returns a non-integer RECENT value in its response to a
I{SELECT} or I{EXAMINE} command, the L{Deferred} returned by
L{IMAP4Client.select} or L{IMAP4Client.examine} fails with
L{IllegalServerResponse}.
"""
d = self._examineOrSelect()
self._response('* foo RECENT')
self.assertRaises(
imap4.IllegalServerResponse, self._extractDeferredResult, d)
def test_unseen(self):
"""
If the server response to a I{SELECT} or I{EXAMINE} command includes an
I{UNSEEN} response, the L{Deferred} returned by L{IMAP4Client.select} or
L{IMAP4Client.examine} fires with a C{dict} including the value
associated with the C{'UNSEEN'} key.
"""
d = self._examineOrSelect()
self._response('* OK [UNSEEN 8] Message 8 is first unseen')
self.assertEqual(
self._extractDeferredResult(d),
{'READ-WRITE': False, 'UNSEEN': 8})
def test_nonIntegerUnseen(self):
"""
If the server returns a non-integer UNSEEN value in its response to a
I{SELECT} or I{EXAMINE} command, the L{Deferred} returned by
L{IMAP4Client.select} or L{IMAP4Client.examine} fails with
L{IllegalServerResponse}.
"""
d = self._examineOrSelect()
self._response('* OK [UNSEEN foo] Message foo is first unseen')
self.assertRaises(
imap4.IllegalServerResponse, self._extractDeferredResult, d)
def test_uidvalidity(self):
"""
If the server response to a I{SELECT} or I{EXAMINE} command includes an
I{UIDVALIDITY} response, the L{Deferred} returned by
L{IMAP4Client.select} or L{IMAP4Client.examine} fires with a C{dict}
including the value associated with the C{'UIDVALIDITY'} key.
"""
d = self._examineOrSelect()
self._response('* OK [UIDVALIDITY 12345] UIDs valid')
self.assertEqual(
self._extractDeferredResult(d),
{'READ-WRITE': False, 'UIDVALIDITY': 12345})
def test_nonIntegerUIDVALIDITY(self):
"""
If the server returns a non-integer UIDVALIDITY value in its response to
a I{SELECT} or I{EXAMINE} command, the L{Deferred} returned by
L{IMAP4Client.select} or L{IMAP4Client.examine} fails with
L{IllegalServerResponse}.
"""
d = self._examineOrSelect()
self._response('* OK [UIDVALIDITY foo] UIDs valid')
self.assertRaises(
imap4.IllegalServerResponse, self._extractDeferredResult, d)
def test_uidnext(self):
"""
If the server response to a I{SELECT} or I{EXAMINE} command includes an
I{UIDNEXT} response, the L{Deferred} returned by L{IMAP4Client.select}
or L{IMAP4Client.examine} fires with a C{dict} including the value
associated with the C{'UIDNEXT'} key.
"""
d = self._examineOrSelect()
self._response('* OK [UIDNEXT 4392] Predicted next UID')
self.assertEqual(
self._extractDeferredResult(d),
{'READ-WRITE': False, 'UIDNEXT': 4392})
def test_nonIntegerUIDNEXT(self):
"""
If the server returns a non-integer UIDNEXT value in its response to a
I{SELECT} or I{EXAMINE} command, the L{Deferred} returned by
L{IMAP4Client.select} or L{IMAP4Client.examine} fails with
L{IllegalServerResponse}.
"""
d = self._examineOrSelect()
self._response('* OK [UIDNEXT foo] Predicted next UID')
self.assertRaises(
imap4.IllegalServerResponse, self._extractDeferredResult, d)
def test_flags(self):
"""
If the server response to a I{SELECT} or I{EXAMINE} command includes an
I{FLAGS} response, the L{Deferred} returned by L{IMAP4Client.select} or
L{IMAP4Client.examine} fires with a C{dict} including the value
associated with the C{'FLAGS'} key.
"""
d = self._examineOrSelect()
self._response(
'* FLAGS (\\Answered \\Flagged \\Deleted \\Seen \\Draft)')
self.assertEqual(
self._extractDeferredResult(d), {
'READ-WRITE': False,
'FLAGS': ('\\Answered', '\\Flagged', '\\Deleted', '\\Seen',
'\\Draft')})
def test_permanentflags(self):
"""
If the server response to a I{SELECT} or I{EXAMINE} command includes an
I{FLAGS} response, the L{Deferred} returned by L{IMAP4Client.select} or
L{IMAP4Client.examine} fires with a C{dict} including the value
associated with the C{'FLAGS'} key.
"""
d = self._examineOrSelect()
self._response(
'* OK [PERMANENTFLAGS (\\Starred)] Just one permanent flag in '
'that list up there')
self.assertEqual(
self._extractDeferredResult(d), {
'READ-WRITE': False,
'PERMANENTFLAGS': ('\\Starred',)})
def test_unrecognizedOk(self):
"""
If the server response to a I{SELECT} or I{EXAMINE} command includes an
I{OK} with unrecognized response code text, parsing does not fail.
"""
d = self._examineOrSelect()
self._response(
'* OK [X-MADE-UP] I just made this response text up.')
# The value won't show up in the result. It would be okay if it did
# someday, perhaps. This shouldn't ever happen, though.
self.assertEqual(
self._extractDeferredResult(d), {'READ-WRITE': False})
def test_bareOk(self):
"""
If the server response to a I{SELECT} or I{EXAMINE} command includes an
I{OK} with no response code text, parsing does not fail.
"""
d = self._examineOrSelect()
self._response('* OK')
self.assertEqual(
self._extractDeferredResult(d), {'READ-WRITE': False})
class IMAP4ClientExamineTests(SelectionTestsMixin, unittest.TestCase):
"""
Tests for the L{IMAP4Client.examine} method.
An example of usage of the EXAMINE command from RFC 3501, section 6.3.2::
S: * 17 EXISTS
S: * 2 RECENT
S: * OK [UNSEEN 8] Message 8 is first unseen
S: * OK [UIDVALIDITY 3857529045] UIDs valid
S: * OK [UIDNEXT 4392] Predicted next UID
S: * FLAGS (\\Answered \\Flagged \\Deleted \\Seen \\Draft)
S: * OK [PERMANENTFLAGS ()] No permanent flags permitted
S: A932 OK [READ-ONLY] EXAMINE completed
"""
method = 'examine'
command = 'EXAMINE'
class IMAP4ClientSelectTests(SelectionTestsMixin, unittest.TestCase):
"""
Tests for the L{IMAP4Client.select} method.
An example of usage of the SELECT command from RFC 3501, section 6.3.1::
C: A142 SELECT INBOX
S: * 172 EXISTS
S: * 1 RECENT
S: * OK [UNSEEN 12] Message 12 is first unseen
S: * OK [UIDVALIDITY 3857529045] UIDs valid
S: * OK [UIDNEXT 4392] Predicted next UID
S: * FLAGS (\Answered \Flagged \Deleted \Seen \Draft)
S: * OK [PERMANENTFLAGS (\Deleted \Seen \*)] Limited
S: A142 OK [READ-WRITE] SELECT completed
"""
method = 'select'
command = 'SELECT'
class IMAP4ClientExpungeTests(PreauthIMAP4ClientMixin, unittest.TestCase):
"""
Tests for the L{IMAP4Client.expunge} method.
An example of usage of the EXPUNGE command from RFC 3501, section 6.4.3::
C: A202 EXPUNGE
S: * 3 EXPUNGE
S: * 3 EXPUNGE
S: * 5 EXPUNGE
S: * 8 EXPUNGE
S: A202 OK EXPUNGE completed
"""
def _expunge(self):
d = self.client.expunge()
self.assertEqual(self.transport.value(), b'0001 EXPUNGE\r\n')
self.transport.clear()
return d
def _response(self, sequenceNumbers):
for number in sequenceNumbers:
self.client.lineReceived(b'* %s EXPUNGE' % (number,))
self.client.lineReceived(b'0001 OK EXPUNGE COMPLETED')
def test_expunge(self):
"""
L{IMAP4Client.expunge} sends the I{EXPUNGE} command and returns a
L{Deferred} which fires with a C{list} of message sequence numbers
given by the server's response.
"""
d = self._expunge()
self._response([3, 3, 5, 8])
self.assertEqual(self._extractDeferredResult(d), [3, 3, 5, 8])
def test_nonIntegerExpunged(self):
"""
If the server responds with a non-integer where a message sequence
number is expected, the L{Deferred} returned by L{IMAP4Client.expunge}
fails with L{IllegalServerResponse}.
"""
d = self._expunge()
self._response([3, 3, 'foo', 8])
self.assertRaises(
imap4.IllegalServerResponse, self._extractDeferredResult, d)
class IMAP4ClientSearchTests(PreauthIMAP4ClientMixin, unittest.TestCase):
"""
Tests for the L{IMAP4Client.search} method.
An example of usage of the SEARCH command from RFC 3501, section 6.4.4::
C: A282 SEARCH FLAGGED SINCE 1-Feb-1994 NOT FROM "Smith"
S: * SEARCH 2 84 882
S: A282 OK SEARCH completed
C: A283 SEARCH TEXT "string not in mailbox"
S: * SEARCH
S: A283 OK SEARCH completed
C: A284 SEARCH CHARSET UTF-8 TEXT {6}
C: XXXXXX
S: * SEARCH 43
S: A284 OK SEARCH completed
"""
def _search(self):
d = self.client.search(imap4.Query(text="ABCDEF"))
self.assertEqual(
self.transport.value(), b'0001 SEARCH (TEXT "ABCDEF")\r\n')
return d
def _response(self, messageNumbers):
self.client.lineReceived(
b"* SEARCH " + b" ".join(map(str, messageNumbers)))
self.client.lineReceived(b"0001 OK SEARCH completed")
def test_search(self):
"""
L{IMAP4Client.search} sends the I{SEARCH} command and returns a
L{Deferred} which fires with a C{list} of message sequence numbers
given by the server's response.
"""
d = self._search()
self._response([2, 5, 10])
self.assertEqual(self._extractDeferredResult(d), [2, 5, 10])
def test_nonIntegerFound(self):
"""
If the server responds with a non-integer where a message sequence
number is expected, the L{Deferred} returned by L{IMAP4Client.search}
fails with L{IllegalServerResponse}.
"""
d = self._search()
self._response([2, "foo", 10])
self.assertRaises(
imap4.IllegalServerResponse, self._extractDeferredResult, d)
class IMAP4ClientFetchTests(PreauthIMAP4ClientMixin, unittest.TestCase):
"""
Tests for the L{IMAP4Client.fetch} method.
See RFC 3501, section 6.4.5.
"""
def test_fetchUID(self):
"""
L{IMAP4Client.fetchUID} sends the I{FETCH UID} command and returns a
L{Deferred} which fires with a C{dict} mapping message sequence numbers
to C{dict}s mapping C{'UID'} to that message's I{UID} in the server's
response.
"""
d = self.client.fetchUID(b'1:7')
self.assertEqual(self.transport.value(), b'0001 FETCH 1:7 (UID)\r\n')
self.client.lineReceived(b'* 2 FETCH (UID 22)')
self.client.lineReceived(b'* 3 FETCH (UID 23)')
self.client.lineReceived(b'* 4 FETCH (UID 24)')
self.client.lineReceived(b'* 5 FETCH (UID 25)')
self.client.lineReceived(b'0001 OK FETCH completed')
self.assertEqual(
self._extractDeferredResult(d), {
2: {'UID': '22'},
3: {'UID': '23'},
4: {'UID': '24'},
5: {'UID': '25'}})
def test_fetchUIDNonIntegerFound(self):
"""
If the server responds with a non-integer where a message sequence
number is expected, the L{Deferred} returned by L{IMAP4Client.fetchUID}
fails with L{IllegalServerResponse}.
"""
d = self.client.fetchUID(b'1')
self.assertEqual(self.transport.value(), b'0001 FETCH 1 (UID)\r\n')
self.client.lineReceived('* foo FETCH (UID 22)')
self.client.lineReceived('0001 OK FETCH completed')
self.assertRaises(
imap4.IllegalServerResponse, self._extractDeferredResult, d)
def test_incompleteFetchUIDResponse(self):
"""
If the server responds with an incomplete I{FETCH} response line, the
L{Deferred} returned by L{IMAP4Client.fetchUID} fails with
L{IllegalServerResponse}.
"""
d = self.client.fetchUID(b'1:7')
self.assertEqual(self.transport.value(), b'0001 FETCH 1:7 (UID)\r\n')
self.client.lineReceived(b'* 2 FETCH (UID 22)')
self.client.lineReceived(b'* 3 FETCH (UID)')
self.client.lineReceived(b'* 4 FETCH (UID 24)')
self.client.lineReceived(b'0001 OK FETCH completed')
self.assertRaises(
imap4.IllegalServerResponse, self._extractDeferredResult, d)
def test_fetchBody(self):
"""
L{IMAP4Client.fetchBody} sends the I{FETCH BODY} command and returns a
L{Deferred} which fires with a C{dict} mapping message sequence numbers
to C{dict}s mapping C{'RFC822.TEXT'} to that message's body as given in
the server's response.
"""
d = self.client.fetchBody(b'3')
self.assertEqual(
self.transport.value(), b'0001 FETCH 3 (RFC822.TEXT)\r\n')
self.client.lineReceived(b'* 3 FETCH (RFC822.TEXT "Message text")')
self.client.lineReceived(b'0001 OK FETCH completed')
self.assertEqual(
self._extractDeferredResult(d),
{3: {b'RFC822.TEXT': b'Message text'}})
def test_fetchSpecific(self):
"""
L{IMAP4Client.fetchSpecific} sends the I{BODY[]} command if no
parameters beyond the message set to retrieve are given. It returns a
L{Deferred} which fires with a C{dict} mapping message sequence numbers
to C{list}s of corresponding message data given by the server's
response.
"""
d = self.client.fetchSpecific(b'7')
self.assertEqual(
self.transport.value(), b'0001 FETCH 7 BODY[]\r\n')
self.client.lineReceived(b'* 7 FETCH (BODY[] "Some body")')
self.client.lineReceived(b'0001 OK FETCH completed')
self.assertEqual(
self._extractDeferredResult(d), {7: [['BODY', [], "Some body"]]})
def test_fetchSpecificPeek(self):
"""
L{IMAP4Client.fetchSpecific} issues a I{BODY.PEEK[]} command if passed
C{True} for the C{peek} parameter.
"""
d = self.client.fetchSpecific(b'6', peek=True)
self.assertEqual(
self.transport.value(), b'0001 FETCH 6 BODY.PEEK[]\r\n')
# BODY.PEEK responses are just BODY
self.client.lineReceived(b'* 6 FETCH (BODY[] "Some body")')
self.client.lineReceived(b'0001 OK FETCH completed')
self.assertEqual(
self._extractDeferredResult(d), {6: [['BODY', [], "Some body"]]})
def test_fetchSpecificNumbered(self):
"""
L{IMAP4Client.fetchSpecific}, when passed a sequence for
C{headerNumber}, sends the I{BODY[N.M]} command. It returns a
L{Deferred} which fires with a C{dict} mapping message sequence numbers
to C{list}s of corresponding message data given by the server's
response.
"""
d = self.client.fetchSpecific(b'7', headerNumber=(1, 2, 3))
self.assertEqual(
self.transport.value(), b'0001 FETCH 7 BODY[1.2.3]\r\n')
self.client.lineReceived(b'* 7 FETCH (BODY[1.2.3] "Some body")')
self.client.lineReceived(b'0001 OK FETCH completed')
self.assertEqual(
self._extractDeferredResult(d),
{7: [['BODY', ['1.2.3'], "Some body"]]})
def test_fetchSpecificText(self):
"""
L{IMAP4Client.fetchSpecific}, when passed C{'TEXT'} for C{headerType},
sends the I{BODY[TEXT]} command. It returns a L{Deferred} which fires
with a C{dict} mapping message sequence numbers to C{list}s of
corresponding message data given by the server's response.
"""
d = self.client.fetchSpecific(b'8', headerType=b'TEXT')
self.assertEqual(
self.transport.value(), b'0001 FETCH 8 BODY[TEXT]\r\n')
self.client.lineReceived(b'* 8 FETCH (BODY[TEXT] "Some body")')
self.client.lineReceived(b'0001 OK FETCH completed')
self.assertEqual(
self._extractDeferredResult(d),
{8: [[b'BODY', [b'TEXT'], b"Some body"]]})
def test_fetchSpecificNumberedText(self):
"""
If passed a value for the C{headerNumber} parameter and C{'TEXT'} for
the C{headerType} parameter, L{IMAP4Client.fetchSpecific} sends a
I{BODY[number.TEXT]} request and returns a L{Deferred} which fires with
a C{dict} mapping message sequence numbers to C{list}s of message data
given by the server's response.
"""
d = self.client.fetchSpecific(b'4', headerType=b'TEXT', headerNumber=7)
self.assertEqual(
self.transport.value(), b'0001 FETCH 4 BODY[7.TEXT]\r\n')
self.client.lineReceived(b'* 4 FETCH (BODY[7.TEXT] "Some body")')
self.client.lineReceived(b'0001 OK FETCH completed')
self.assertEqual(
self._extractDeferredResult(d),
{4: [['BODY', ['7.TEXT'], "Some body"]]})
def test_incompleteFetchSpecificTextResponse(self):
"""
If the server responds to a I{BODY[TEXT]} request with a I{FETCH} line
which is truncated after the I{BODY[TEXT]} tokens, the L{Deferred}
returned by L{IMAP4Client.fetchUID} fails with
L{IllegalServerResponse}.
"""
d = self.client.fetchSpecific(b'8', headerType=b'TEXT')
self.assertEqual(
self.transport.value(), b'0001 FETCH 8 BODY[TEXT]\r\n')
self.client.lineReceived(b'* 8 FETCH (BODY[TEXT])')
self.client.lineReceived(b'0001 OK FETCH completed')
self.assertRaises(
imap4.IllegalServerResponse, self._extractDeferredResult, d)
def test_fetchSpecificMIME(self):
"""
L{IMAP4Client.fetchSpecific}, when passed C{'MIME'} for C{headerType},
sends the I{BODY[MIME]} command. It returns a L{Deferred} which fires
with a C{dict} mapping message sequence numbers to C{list}s of
corresponding message data given by the server's response.
"""
d = self.client.fetchSpecific(b'8', headerType=b'MIME')
self.assertEqual(
self.transport.value(), b'0001 FETCH 8 BODY[MIME]\r\n')
self.client.lineReceived(b'* 8 FETCH (BODY[MIME] "Some body")')
self.client.lineReceived(b'0001 OK FETCH completed')
self.assertEqual(
self._extractDeferredResult(d),
{8: [['BODY', ['MIME'], "Some body"]]})
def test_fetchSpecificPartial(self):
"""
L{IMAP4Client.fetchSpecific}, when passed C{offset} and C{length},
sends a partial content request (like I{BODY[TEXT]}).
It returns a L{Deferred} which fires with a C{dict} mapping message
sequence numbers to C{list}s of corresponding message data given by the
server's response.
"""
d = self.client.fetchSpecific(
b'9', headerType=b'TEXT', offset=17, length=3)
self.assertEqual(
self.transport.value(), b'0001 FETCH 9 BODY[TEXT]<17.3>\r\n')
self.client.lineReceived(b'* 9 FETCH (BODY[TEXT]<17> "foo")')
self.client.lineReceived(b'0001 OK FETCH completed')
self.assertEqual(
self._extractDeferredResult(d),
{9: [[b'BODY', [b'TEXT'], b'<17>', b'foo']]})
def test_incompleteFetchSpecificPartialResponse(self):
"""
If the server responds to a I{BODY[TEXT]} request with a I{FETCH} line
which is truncated after the I{BODY[TEXT]} tokens, the
L{Deferred} returned by L{IMAP4Client.fetchUID} fails with
L{IllegalServerResponse}.
"""
d = self.client.fetchSpecific(b'8', headerType=b'TEXT')
self.assertEqual(
self.transport.value(), b'0001 FETCH 8 BODY[TEXT]\r\n')
self.client.lineReceived(b'* 8 FETCH (BODY[TEXT]<17>)')
self.client.lineReceived(b'0001 OK FETCH completed')
self.assertRaises(
imap4.IllegalServerResponse, self._extractDeferredResult, d)
def test_fetchSpecificHTML(self):
"""
If the body of a message begins with I{<} and ends with I{>} (as,
for example, HTML bodies typically will), this is still interpreted
as the body by L{IMAP4Client.fetchSpecific} (and particularly, not
as a length indicator for a response to a request for a partial
body).
"""
d = self.client.fetchSpecific(b'7')
self.assertEqual(
self.transport.value(), b'0001 FETCH 7 BODY[]\r\n')
self.client.lineReceived(b'* 7 FETCH (BODY[] "test")')
self.client.lineReceived(b'0001 OK FETCH completed')
self.assertEqual(
self._extractDeferredResult(d), {7: [[b'BODY', [], b"test"]]})
class IMAP4ClientStoreTests(PreauthIMAP4ClientMixin, unittest.TestCase):
"""
Tests for the L{IMAP4Client.setFlags}, L{IMAP4Client.addFlags}, and
L{IMAP4Client.removeFlags} methods.
An example of usage of the STORE command, in terms of which these three
methods are implemented, from RFC 3501, section 6.4.6::
C: A003 STORE 2:4 +FLAGS (\Deleted)
S: * 2 FETCH (FLAGS (\Deleted \Seen))
S: * 3 FETCH (FLAGS (\Deleted))
S: * 4 FETCH (FLAGS (\Deleted \Flagged \Seen))
S: A003 OK STORE completed
"""
clientProtocol = StillSimplerClient
def _flagsTest(self, method, item):
"""
Test a non-silent flag modifying method. Call the method, assert that
the correct bytes are sent, deliver a I{FETCH} response, and assert
that the result of the Deferred returned by the method is correct.
@param method: The name of the method to test.
@param item: The data item which is expected to be specified.
"""
d = getattr(self.client, method)(b'3', (b'\\Read', b'\\Seen'), False)
self.assertEqual(
self.transport.value(),
b'0001 STORE 3 ' + item + b' (\\Read \\Seen)\r\n')
self.client.lineReceived(b'* 3 FETCH (FLAGS (\\Read \\Seen))')
self.client.lineReceived(b'0001 OK STORE completed')
self.assertEqual(
self._extractDeferredResult(d),
{3: {b'FLAGS': [b'\\Read', b'\\Seen']}})
def _flagsSilentlyTest(self, method, item):
"""
Test a silent flag modifying method. Call the method, assert that the
correct bytes are sent, deliver an I{OK} response, and assert that the
result of the Deferred returned by the method is correct.
@param method: The name of the method to test.
@param item: The data item which is expected to be specified.
"""
d = getattr(self.client, method)(b'3', (b'\\Read', b'\\Seen'), True)
self.assertEqual(
self.transport.value(),
b'0001 STORE 3 ' + item + b' (\\Read \\Seen)\r\n')
self.client.lineReceived(b'0001 OK STORE completed')
self.assertEqual(self._extractDeferredResult(d), {})
def _flagsSilentlyWithUnsolicitedDataTest(self, method, item):
"""
Test unsolicited data received in response to a silent flag modifying
method. Call the method, assert that the correct bytes are sent,
deliver the unsolicited I{FETCH} response, and assert that the result
of the Deferred returned by the method is correct.
@param method: The name of the method to test.
@param item: The data item which is expected to be specified.
"""
d = getattr(self.client, method)(b'3', (b'\\Read', b'\\Seen'), True)
self.assertEqual(
self.transport.value(),
b'0001 STORE 3 ' + item + b' (\\Read \\Seen)\r\n')
self.client.lineReceived(b'* 2 FETCH (FLAGS (\\Read \\Seen))')
self.client.lineReceived(b'0001 OK STORE completed')
self.assertEqual(self._extractDeferredResult(d), {})
self.assertEqual(self.client.flags, {2: [b'\\Read', b'\\Seen']})
def test_setFlags(self):
"""
When passed a C{False} value for the C{silent} parameter,
L{IMAP4Client.setFlags} sends the I{STORE} command with a I{FLAGS} data
item and returns a L{Deferred} which fires with a C{dict} mapping
message sequence numbers to C{dict}s mapping C{'FLAGS'} to the new
flags of those messages.
"""
self._flagsTest('setFlags', b'FLAGS')
def test_setFlagsSilently(self):
"""
When passed a C{True} value for the C{silent} parameter,
L{IMAP4Client.setFlags} sends the I{STORE} command with a
I{FLAGS.SILENT} data item and returns a L{Deferred} which fires with an
empty dictionary.
"""
self._flagsSilentlyTest('setFlags', b'FLAGS.SILENT')
def test_setFlagsSilentlyWithUnsolicitedData(self):
"""
If unsolicited flag data is received in response to a I{STORE}
I{FLAGS.SILENT} request, that data is passed to the C{flagsChanged}
callback.
"""
self._flagsSilentlyWithUnsolicitedDataTest('setFlags', b'FLAGS.SILENT')
def test_addFlags(self):
"""
L{IMAP4Client.addFlags} is like L{IMAP4Client.setFlags}, but sends
I{+FLAGS} instead of I{FLAGS}.
"""
self._flagsTest('addFlags', b'+FLAGS')
def test_addFlagsSilently(self):
"""
L{IMAP4Client.addFlags} with a C{True} value for C{silent} behaves like
L{IMAP4Client.setFlags} with a C{True} value for C{silent}, but it
sends I{+FLAGS.SILENT} instead of I{FLAGS.SILENT}.
"""
self._flagsSilentlyTest('addFlags', b'+FLAGS.SILENT')
def test_addFlagsSilentlyWithUnsolicitedData(self):
"""
L{IMAP4Client.addFlags} behaves like L{IMAP4Client.setFlags} when used
in silent mode and unsolicited data is received.
"""
self._flagsSilentlyWithUnsolicitedDataTest('addFlags', b'+FLAGS.SILENT')
def test_removeFlags(self):
"""
L{IMAP4Client.removeFlags} is like L{IMAP4Client.setFlags}, but sends
I{-FLAGS} instead of I{FLAGS}.
"""
self._flagsTest('removeFlags', b'-FLAGS')
def test_removeFlagsSilently(self):
"""
L{IMAP4Client.removeFlags} with a C{True} value for C{silent} behaves
like L{IMAP4Client.setFlags} with a C{True} value for C{silent}, but it
sends I{-FLAGS.SILENT} instead of I{FLAGS.SILENT}.
"""
self._flagsSilentlyTest('removeFlags', b'-FLAGS.SILENT')
def test_removeFlagsSilentlyWithUnsolicitedData(self):
"""
L{IMAP4Client.removeFlags} behaves like L{IMAP4Client.setFlags} when
used in silent mode and unsolicited data is received.
"""
self._flagsSilentlyWithUnsolicitedDataTest('removeFlags', b'-FLAGS.SILENT')
class FakeyServer(imap4.IMAP4Server):
state = 'select'
timeout = None
def sendServerGreeting(self):
pass
@implementer(imap4.IMessage)
class FakeyMessage(util.FancyStrMixin):
showAttributes = ('headers', 'flags', 'date', '_body', 'uid')
def __init__(self, headers, flags, date, body, uid, subpart):
self.headers = headers
self.flags = flags
self._body = body
self.size = len(body)
self.date = date
self.uid = uid
self.subpart = subpart
def getHeaders(self, negate, *names):
self.got_headers = negate, names
return self.headers
def getFlags(self):
return self.flags
def getInternalDate(self):
return self.date
def getBodyFile(self):
return BytesIO(self._body)
def getSize(self):
return self.size
def getUID(self):
return self.uid
def isMultipart(self):
return self.subpart is not None
def getSubPart(self, part):
self.got_subpart = part
return self.subpart[part]
class NewStoreTests(unittest.TestCase, IMAP4HelperMixin):
result = None
storeArgs = None
def setUp(self):
self.received_messages = self.received_uid = None
self.server = imap4.IMAP4Server()
self.server.state = 'select'
self.server.mbox = self
self.connected = defer.Deferred()
self.client = SimpleClient(self.connected)
def addListener(self, x):
pass
def removeListener(self, x):
pass
def store(self, *args, **kw):
self.storeArgs = args, kw
return self.response
def _storeWork(self):
def connected():
return self.function(self.messages, self.flags, self.silent, self.uid)
def result(R):
self.result = R
self.connected.addCallback(strip(connected)
).addCallback(result
).addCallback(self._cbStopClient
).addErrback(self._ebGeneral)
def check(ignored):
self.assertEqual(self.result, self.expected)
self.assertEqual(self.storeArgs, self.expectedArgs)
d = loopback.loopbackTCP(self.server, self.client, noisy=False)
d.addCallback(check)
return d
def testSetFlags(self, uid=0):
self.function = self.client.setFlags
self.messages = b'1,5,9'
self.flags = [b'\\A', b'\\B', b'C']
self.silent = False
self.uid = uid
self.response = {
1: [b'\\A', b'\\B', b'C'],
5: [b'\\A', b'\\B', b'C'],
9: [b'\\A', b'\\B', b'C'],
}
self.expected = {
1: {b'FLAGS': [b'\\A', b'\\B', b'C']},
5: {b'FLAGS': [b'\\A', b'\\B', b'C']},
9: {b'FLAGS': [b'\\A', b'\\B', b'C']},
}
msg = imap4.MessageSet()
msg.add(1)
msg.add(5)
msg.add(9)
self.expectedArgs = ((msg, [b'\\A', b'\\B', b'C'], 0), {b'uid': 0})
return self._storeWork()
class GetBodyStructureTests(unittest.TestCase):
"""
Tests for L{imap4.getBodyStructure}, a helper for constructing a list which
directly corresponds to the wire information needed for a I{BODY} or
I{BODYSTRUCTURE} response.
"""
def test_singlePart(self):
"""
L{imap4.getBodyStructure} accepts a L{IMessagePart} provider and returns
a list giving the basic fields for the I{BODY} response for that
message.
"""
body = b'hello, world'
major = 'image'
minor = 'jpeg'
charset = 'us-ascii'
identifier = 'some kind of id'
description = 'great justice'
encoding = 'maximum'
msg = FakeyMessage({
'content-type': major + '/' + minor +
'; charset=' + charset + '; x=y',
'content-id': identifier,
'content-description': description,
'content-transfer-encoding': encoding,
}, (), b'', body, 123, None)
structure = imap4.getBodyStructure(msg)
self.assertEqual(
[major, minor, ["charset", charset, 'x', 'y'], identifier,
description, encoding, len(body)],
structure)
def test_singlePartExtended(self):
"""
L{imap4.getBodyStructure} returns a list giving the basic and extended
fields for a I{BODYSTRUCTURE} response if passed C{True} for the
C{extended} parameter.
"""
body = b'hello, world'
major = 'image'
minor = 'jpeg'
charset = 'us-ascii'
identifier = 'some kind of id'
description = 'great justice'
encoding = 'maximum'
md5 = 'abcdefabcdef'
msg = FakeyMessage({
'content-type': major + '/' + minor +
'; charset=' + charset + '; x=y',
'content-id': identifier,
'content-description': description,
'content-transfer-encoding': encoding,
'content-md5': md5,
'content-disposition': 'attachment; name=foo; size=bar',
'content-language': 'fr',
'content-location': 'France',
}, (), '', body, 123, None)
structure = imap4.getBodyStructure(msg, extended=True)
self.assertEqual(
[major, minor, ["charset", charset, 'x', 'y'], identifier,
description, encoding, len(body), md5,
['attachment', ['name', 'foo', 'size', 'bar']], 'fr', 'France'],
structure)
def test_singlePartWithMissing(self):
"""
For fields with no information contained in the message headers,
L{imap4.getBodyStructure} fills in L{None} values in its result.
"""
major = 'image'
minor = 'jpeg'
body = b'hello, world'
msg = FakeyMessage({
'content-type': major + '/' + minor
}, (), b'', body, 123, None)
structure = imap4.getBodyStructure(msg, extended=True)
self.assertEqual(
[major, minor, None, None, None, None, len(body), None, None,
None, None],
structure)
def test_textPart(self):
"""
For a I{text/*} message, the number of lines in the message body are
included after the common single-part basic fields.
"""
body = b'hello, world\nhow are you?\ngoodbye\n'
major = 'text'
minor = 'jpeg'
charset = 'us-ascii'
identifier = 'some kind of id'
description = 'great justice'
encoding = 'maximum'
msg = FakeyMessage({
'content-type': major + '/' + minor +
'; charset=' + charset + '; x=y',
'content-id': identifier,
'content-description': description,
'content-transfer-encoding': encoding,
}, (), b'', body, 123, None)
structure = imap4.getBodyStructure(msg)
self.assertEqual(
[major, minor, ["charset", charset, 'x', 'y'], identifier,
description, encoding, len(body), len(body.splitlines())],
structure)
def test_rfc822Message(self):
"""
For a I{message/rfc822} message, the common basic fields are followed
by information about the contained message.
"""
body = b'hello, world\nhow are you?\ngoodbye\n'
major = 'text'
minor = 'jpeg'
charset = 'us-ascii'
identifier = 'some kind of id'
description = 'great justice'
encoding = 'maximum'
msg = FakeyMessage({
'content-type': major + '/' + minor +
'; charset=' + charset + '; x=y',
'from': 'Alice ',
'to': 'Bob ',
'content-id': identifier,
'content-description': description,
'content-transfer-encoding': encoding,
}, (), '', body, 123, None)
container = FakeyMessage({
'content-type': 'message/rfc822',
}, (), b'', b'', 123, [msg])
structure = imap4.getBodyStructure(container)
self.assertEqual(
['message', 'rfc822', None, None, None, None, 0,
imap4.getEnvelope(msg), imap4.getBodyStructure(msg), 3],
structure)
def test_multiPart(self):
"""
For a I{multipart/*} message, L{imap4.getBodyStructure} returns a list
containing the body structure information for each part of the message
followed by an element giving the MIME subtype of the message.
"""
oneSubPart = FakeyMessage({
'content-type': 'image/jpeg; x=y',
'content-id': 'some kind of id',
'content-description': 'great justice',
'content-transfer-encoding': 'maximum',
}, (), b'', b'hello world', 123, None)
anotherSubPart = FakeyMessage({
'content-type': 'text/plain; charset=us-ascii',
}, (), b'', b'some stuff', 321, None)
container = FakeyMessage({
'content-type': 'multipart/related',
}, (), b'', b'', 555, [oneSubPart, anotherSubPart])
self.assertEqual(
[imap4.getBodyStructure(oneSubPart),
imap4.getBodyStructure(anotherSubPart),
'related'],
imap4.getBodyStructure(container))
def test_multiPartExtended(self):
"""
When passed a I{multipart/*} message and C{True} for the C{extended}
argument, L{imap4.getBodyStructure} includes extended structure
information from the parts of the multipart message and extended
structure information about the multipart message itself.
"""
oneSubPart = FakeyMessage({
b'content-type': b'image/jpeg; x=y',
b'content-id': b'some kind of id',
b'content-description': b'great justice',
b'content-transfer-encoding': b'maximum',
}, (), b'', b'hello world', 123, None)
anotherSubPart = FakeyMessage({
b'content-type': b'text/plain; charset=us-ascii',
}, (), b'', b'some stuff', 321, None)
container = FakeyMessage({
'content-type': 'multipart/related; foo=bar',
'content-language': 'es',
'content-location': 'Spain',
'content-disposition': 'attachment; name=monkeys',
}, (), b'', b'', 555, [oneSubPart, anotherSubPart])
self.assertEqual(
[imap4.getBodyStructure(oneSubPart, extended=True),
imap4.getBodyStructure(anotherSubPart, extended=True),
'related', ['foo', 'bar'], ['attachment', ['name', 'monkeys']],
'es', 'Spain'],
imap4.getBodyStructure(container, extended=True))
class NewFetchTests(unittest.TestCase, IMAP4HelperMixin):
def setUp(self):
self.received_messages = self.received_uid = None
self.result = None
self.server = imap4.IMAP4Server()
self.server.state = 'select'
self.server.mbox = self
self.connected = defer.Deferred()
self.client = SimpleClient(self.connected)
def addListener(self, x):
pass
def removeListener(self, x):
pass
def fetch(self, messages, uid):
self.received_messages = messages
self.received_uid = uid
return iter(zip(range(len(self.msgObjs)), self.msgObjs))
def _fetchWork(self, uid):
if uid:
for (i, msg) in zip(range(len(self.msgObjs)), self.msgObjs):
self.expected[i]['UID'] = str(msg.getUID())
def result(R):
self.result = R
self.connected.addCallback(lambda _: self.function(self.messages, uid)
).addCallback(result
).addCallback(self._cbStopClient
).addErrback(self._ebGeneral)
d = loopback.loopbackTCP(self.server, self.client, noisy=False)
d.addCallback(lambda x : self.assertEqual(self.result, self.expected))
return d
def testFetchUID(self):
self.function = lambda m, u: self.client.fetchUID(m)
self.messages = '7'
self.msgObjs = [
FakeyMessage({}, (), '', '', 12345, None),
FakeyMessage({}, (), '', '', 999, None),
FakeyMessage({}, (), '', '', 10101, None),
]
self.expected = {
0: {'UID': '12345'},
1: {'UID': '999'},
2: {'UID': '10101'},
}
return self._fetchWork(0)
def testFetchFlags(self, uid=0):
self.function = self.client.fetchFlags
self.messages = '9'
self.msgObjs = [
FakeyMessage({}, ['FlagA', 'FlagB', '\\FlagC'], '', '', 54321, None),
FakeyMessage({}, ['\\FlagC', 'FlagA', 'FlagB'], '', '', 12345, None),
]
self.expected = {
0: {'FLAGS': ['FlagA', 'FlagB', '\\FlagC']},
1: {'FLAGS': ['\\FlagC', 'FlagA', 'FlagB']},
}
return self._fetchWork(uid)
def testFetchFlagsUID(self):
return self.testFetchFlags(1)
def testFetchInternalDate(self, uid=0):
self.function = self.client.fetchInternalDate
self.messages = '13'
self.msgObjs = [
FakeyMessage({}, (), 'Fri, 02 Nov 2003 21:25:10 GMT', '', 23232, None),
FakeyMessage({}, (), 'Thu, 29 Dec 2013 11:31:52 EST', '', 101, None),
FakeyMessage({}, (), 'Mon, 10 Mar 1992 02:44:30 CST', '', 202, None),
FakeyMessage({}, (), 'Sat, 11 Jan 2000 14:40:24 PST', '', 303, None),
]
self.expected = {
0: {'INTERNALDATE': '02-Nov-2003 21:25:10 +0000'},
1: {'INTERNALDATE': '29-Dec-2013 11:31:52 -0500'},
2: {'INTERNALDATE': '10-Mar-1992 02:44:30 -0600'},
3: {'INTERNALDATE': '11-Jan-2000 14:40:24 -0800'},
}
return self._fetchWork(uid)
def testFetchInternalDateUID(self):
return self.testFetchInternalDate(1)
def test_fetchInternalDateLocaleIndependent(self):
"""
The month name in the date is locale independent.
"""
# Fake that we're in a language where December is not Dec
currentLocale = locale.setlocale(locale.LC_ALL, None)
locale.setlocale(locale.LC_ALL, "es_AR.UTF8")
self.addCleanup(locale.setlocale, locale.LC_ALL, currentLocale)
return self.testFetchInternalDate(1)
# if alternate locale is not available, the previous test will be skipped,
# please install this locale for it to run. Avoid using locale.getlocale to
# learn the current locale; its values don't round-trip well on all
# platforms. Fortunately setlocale returns a value which does round-trip
# well.
currentLocale = locale.setlocale(locale.LC_ALL, None)
try:
locale.setlocale(locale.LC_ALL, "es_AR.UTF8")
except locale.Error:
test_fetchInternalDateLocaleIndependent.skip = (
"The es_AR.UTF8 locale is not installed.")
else:
locale.setlocale(locale.LC_ALL, currentLocale)
def testFetchEnvelope(self, uid=0):
self.function = self.client.fetchEnvelope
self.messages = '15'
self.msgObjs = [
FakeyMessage({
'from': 'user@domain', 'to': 'resu@domain',
'date': 'thursday', 'subject': 'it is a message',
'message-id': 'id-id-id-yayaya'}, (), '', '', 65656,
None),
]
self.expected = {
0: {'ENVELOPE':
['thursday', 'it is a message',
[[None, None, 'user', 'domain']],
[[None, None, 'user', 'domain']],
[[None, None, 'user', 'domain']],
[[None, None, 'resu', 'domain']],
None, None, None, 'id-id-id-yayaya']
}
}
return self._fetchWork(uid)
def testFetchEnvelopeUID(self):
return self.testFetchEnvelope(1)
def test_fetchBodyStructure(self, uid=0):
"""
L{IMAP4Client.fetchBodyStructure} issues a I{FETCH BODYSTRUCTURE}
command and returns a Deferred which fires with a structure giving the
result of parsing the server's response. The structure is a list
reflecting the parenthesized data sent by the server, as described by
RFC 3501, section 7.4.2.
"""
self.function = self.client.fetchBodyStructure
self.messages = b'3:9,10:*'
self.msgObjs = [FakeyMessage({
'content-type': 'text/plain; name=thing; key="value"',
'content-id': 'this-is-the-content-id',
'content-description': 'describing-the-content-goes-here!',
'content-transfer-encoding': '8BIT',
'content-md5': 'abcdef123456',
'content-disposition': 'attachment; filename=monkeys',
'content-language': 'es',
'content-location': 'http://example.com/monkeys',
}, (), '', b'Body\nText\nGoes\nHere\n', 919293, None)]
self.expected = {0: {'BODYSTRUCTURE': [
'text', 'plain', ['key', 'value', 'name', 'thing'],
'this-is-the-content-id', 'describing-the-content-goes-here!',
'8BIT', '20', '4', 'abcdef123456',
['attachment', ['filename', 'monkeys']], 'es',
'http://example.com/monkeys']}}
return self._fetchWork(uid)
def testFetchBodyStructureUID(self):
"""
If passed C{True} for the C{uid} argument, C{fetchBodyStructure} can
also issue a I{UID FETCH BODYSTRUCTURE} command.
"""
return self.test_fetchBodyStructure(1)
def test_fetchBodyStructureMultipart(self, uid=0):
"""
L{IMAP4Client.fetchBodyStructure} can also parse the response to a
I{FETCH BODYSTRUCTURE} command for a multipart message.
"""
self.function = self.client.fetchBodyStructure
self.messages = '3:9,10:*'
innerMessage = FakeyMessage({
'content-type': 'text/plain; name=thing; key="value"',
'content-id': 'this-is-the-content-id',
'content-description': 'describing-the-content-goes-here!',
'content-transfer-encoding': '8BIT',
'content-language': 'fr',
'content-md5': '123456abcdef',
'content-disposition': 'inline',
'content-location': 'outer space',
}, (), '', 'Body\nText\nGoes\nHere\n', 919293, None)
self.msgObjs = [FakeyMessage({
'content-type': 'multipart/mixed; boundary="xyz"',
'content-language': 'en',
'content-location': 'nearby',
}, (), '', '', 919293, [innerMessage])]
self.expected = {0: {'BODYSTRUCTURE': [
['text', 'plain', ['key', 'value', 'name', 'thing'],
'this-is-the-content-id', 'describing-the-content-goes-here!',
'8BIT', '20', '4', '123456abcdef', ['inline', None], 'fr',
'outer space'],
'mixed', ['boundary', 'xyz'], None, 'en', 'nearby'
]}}
return self._fetchWork(uid)
def testFetchSimplifiedBody(self, uid=0):
self.function = self.client.fetchSimplifiedBody
self.messages = '21'
self.msgObjs = [FakeyMessage({}, (), '', 'Yea whatever', 91825,
[FakeyMessage({'content-type': 'image/jpg'}, (), '',
'Body Body Body', None, None
)]
)]
self.expected = {0:
{'BODY':
[None, None, None, None, None, None,
'12'
]
}
}
return self._fetchWork(uid)
def testFetchSimplifiedBodyUID(self):
return self.testFetchSimplifiedBody(1)
def testFetchSimplifiedBodyText(self, uid=0):
self.function = self.client.fetchSimplifiedBody
self.messages = '21'
self.msgObjs = [FakeyMessage({'content-type': 'text/plain'},
(), '', 'Yea whatever', 91825, None)]
self.expected = {0:
{'BODY':
['text', 'plain', None, None, None, None,
'12', '1'
]
}
}
return self._fetchWork(uid)
def testFetchSimplifiedBodyTextUID(self):
return self.testFetchSimplifiedBodyText(1)
def testFetchSimplifiedBodyRFC822(self, uid=0):
self.function = self.client.fetchSimplifiedBody
self.messages = '21'
self.msgObjs = [FakeyMessage({'content-type': 'message/rfc822'},
(), '', 'Yea whatever', 91825,
[FakeyMessage({'content-type': 'image/jpg'}, (), '',
'Body Body Body', None, None
)]
)]
self.expected = {0:
{'BODY':
['message', 'rfc822', None, None, None, None,
'12', [None, None, [[None, None, None]],
[[None, None, None]], None, None, None,
None, None, None], ['image', 'jpg', None,
None, None, None, '14'], '1'
]
}
}
return self._fetchWork(uid)
def testFetchSimplifiedBodyRFC822UID(self):
return self.testFetchSimplifiedBodyRFC822(1)
def test_fetchSimplifiedBodyMultipart(self):
"""
L{IMAP4Client.fetchSimplifiedBody} returns a dictionary mapping message
sequence numbers to fetch responses for the corresponding messages. In
particular, for a multipart message, the value in the dictionary maps
the string C{"BODY"} to a list giving the body structure information for
that message, in the form of a list of subpart body structure
information followed by the subtype of the message (eg C{"alternative"}
for a I{multipart/alternative} message). This structure is self-similar
in the case where a subpart is itself multipart.
"""
self.function = self.client.fetchSimplifiedBody
self.messages = '21'
# A couple non-multipart messages to use as the inner-most payload
singles = [
FakeyMessage(
{'content-type': 'text/plain'},
(), 'date', 'Stuff', 54321, None),
FakeyMessage(
{'content-type': 'text/html'},
(), 'date', 'Things', 32415, None)]
# A multipart/alternative message containing the above non-multipart
# messages. This will be the payload of the outer-most message.
alternative = FakeyMessage(
{'content-type': 'multipart/alternative'},
(), '', 'Irrelevant', 12345, singles)
# The outer-most message, also with a multipart type, containing just
# the single middle message.
mixed = FakeyMessage(
# The message is multipart/mixed
{'content-type': 'multipart/mixed'},
(), '', 'RootOf', 98765, [alternative])
self.msgObjs = [mixed]
self.expected = {
0: {'BODY': [
[['text', 'plain', None, None, None, None, '5', '1'],
['text', 'html', None, None, None, None, '6', '1'],
'alternative'],
'mixed']}}
return self._fetchWork(False)
def testFetchMessage(self, uid=0):
self.function = self.client.fetchMessage
self.messages = '1,3,7,10101'
self.msgObjs = [
FakeyMessage({'Header': 'Value'}, (), '', 'BODY TEXT\r\n', 91, None),
]
self.expected = {
0: {'RFC822': 'Header: Value\r\n\r\nBODY TEXT\r\n'}
}
return self._fetchWork(uid)
def testFetchMessageUID(self):
return self.testFetchMessage(1)
def testFetchHeaders(self, uid=0):
self.function = self.client.fetchHeaders
self.messages = '9,6,2'
self.msgObjs = [
FakeyMessage({'H1': 'V1', 'H2': 'V2'}, (), '', '', 99, None),
]
self.expected = {
0: {'RFC822.HEADER': imap4._formatHeaders({'H1': 'V1', 'H2': 'V2'})},
}
return self._fetchWork(uid)
def testFetchHeadersUID(self):
return self.testFetchHeaders(1)
def testFetchBody(self, uid=0):
self.function = self.client.fetchBody
self.messages = '1,2,3,4,5,6,7'
self.msgObjs = [
FakeyMessage({'Header': 'Value'}, (), '', 'Body goes here\r\n', 171, None),
]
self.expected = {
0: {'RFC822.TEXT': 'Body goes here\r\n'},
}
return self._fetchWork(uid)
def testFetchBodyUID(self):
return self.testFetchBody(1)
def testFetchBodyParts(self):
"""
Test the server's handling of requests for specific body sections.
"""
self.function = self.client.fetchSpecific
self.messages = '1'
outerBody = ''
innerBody1 = 'Contained body message text. Squarge.'
innerBody2 = 'Secondary message text of squarge body.'
headers = OrderedDict()
headers['from'] = 'sender@host'
headers['to'] = 'recipient@domain'
headers['subject'] = 'booga booga boo'
headers['content-type'] = 'multipart/alternative; boundary="xyz"'
innerHeaders = OrderedDict()
innerHeaders['subject'] = 'this is subject text'
innerHeaders['content-type'] = 'text/plain'
innerHeaders2 = OrderedDict()
innerHeaders2['subject'] = 'this is subject'
innerHeaders2['content-type'] = 'text/html'
self.msgObjs = [FakeyMessage(
headers, (), None, outerBody, 123,
[FakeyMessage(innerHeaders, (), None, innerBody1, None, None),
FakeyMessage(innerHeaders2, (), None, innerBody2, None, None)])]
self.expected = {
0: [['BODY', ['1'], 'Contained body message text. Squarge.']]}
def result(R):
self.result = R
self.connected.addCallback(
lambda _: self.function(self.messages, headerNumber=1))
self.connected.addCallback(result)
self.connected.addCallback(self._cbStopClient)
self.connected.addErrback(self._ebGeneral)
d = loopback.loopbackTCP(self.server, self.client, noisy=False)
d.addCallback(lambda ign: self.assertEqual(self.result, self.expected))
return d
def test_fetchBodyPartOfNonMultipart(self):
"""
Single-part messages have an implicit first part which clients
should be able to retrieve explicitly. Test that a client
requesting part 1 of a text/plain message receives the body of the
text/plain part.
"""
self.function = self.client.fetchSpecific
self.messages = '1'
parts = [1]
outerBody = 'DA body'
headers = OrderedDict()
headers['from'] = 'sender@host'
headers['to'] = 'recipient@domain'
headers['subject'] = 'booga booga boo'
headers['content-type'] = 'text/plain'
self.msgObjs = [FakeyMessage(
headers, (), None, outerBody, 123, None)]
self.expected = {0: [['BODY', ['1'], 'DA body']]}
def result(R):
self.result = R
self.connected.addCallback(
lambda _: self.function(self.messages, headerNumber=parts))
self.connected.addCallback(result)
self.connected.addCallback(self._cbStopClient)
self.connected.addErrback(self._ebGeneral)
d = loopback.loopbackTCP(self.server, self.client, noisy=False)
d.addCallback(lambda ign: self.assertEqual(self.result, self.expected))
return d
def testFetchSize(self, uid=0):
self.function = self.client.fetchSize
self.messages = '1:100,2:*'
self.msgObjs = [
FakeyMessage({}, (), '', 'x' * 20, 123, None),
]
self.expected = {
0: {'RFC822.SIZE': '20'},
}
return self._fetchWork(uid)
def testFetchSizeUID(self):
return self.testFetchSize(1)
def testFetchFull(self, uid=0):
self.function = self.client.fetchFull
self.messages = '1,3'
self.msgObjs = [
FakeyMessage({}, ('\\XYZ', '\\YZX', 'Abc'),
'Sun, 25 Jul 2010 06:20:30 -0400 (EDT)',
'xyz' * 2, 654, None),
FakeyMessage({}, ('\\One', '\\Two', 'Three'),
'Mon, 14 Apr 2003 19:43:44 -0400',
'abc' * 4, 555, None),
]
self.expected = {
0: {'FLAGS': ['\\XYZ', '\\YZX', 'Abc'],
'INTERNALDATE': '25-Jul-2010 06:20:30 -0400',
'RFC822.SIZE': '6',
'ENVELOPE': [None, None, [[None, None, None]], [[None, None, None]], None, None, None, None, None, None],
'BODY': [None, None, None, None, None, None, '6']},
1: {'FLAGS': ['\\One', '\\Two', 'Three'],
'INTERNALDATE': '14-Apr-2003 19:43:44 -0400',
'RFC822.SIZE': '12',
'ENVELOPE': [None, None, [[None, None, None]], [[None, None, None]], None, None, None, None, None, None],
'BODY': [None, None, None, None, None, None, '12']},
}
return self._fetchWork(uid)
def testFetchFullUID(self):
return self.testFetchFull(1)
def testFetchAll(self, uid=0):
self.function = self.client.fetchAll
self.messages = '1,2:3'
self.msgObjs = [
FakeyMessage({}, (), 'Mon, 14 Apr 2003 19:43:44 +0400',
'Lalala', 10101, None),
FakeyMessage({}, (), 'Tue, 15 Apr 2003 19:43:44 +0200',
'Alalal', 20202, None),
]
self.expected = {
0: {'ENVELOPE': [None, None, [[None, None, None]], [[None, None, None]], None, None, None, None, None, None],
'RFC822.SIZE': '6',
'INTERNALDATE': '14-Apr-2003 19:43:44 +0400',
'FLAGS': []},
1: {'ENVELOPE': [None, None, [[None, None, None]], [[None, None, None]], None, None, None, None, None, None],
'RFC822.SIZE': '6',
'INTERNALDATE': '15-Apr-2003 19:43:44 +0200',
'FLAGS': []},
}
return self._fetchWork(uid)
def testFetchAllUID(self):
return self.testFetchAll(1)
def testFetchFast(self, uid=0):
self.function = self.client.fetchFast
self.messages = '1'
self.msgObjs = [
FakeyMessage({}, ('\\X',), '19 Mar 2003 19:22:21 -0500', '', 9, None),
]
self.expected = {
0: {'FLAGS': ['\\X'],
'INTERNALDATE': '19-Mar-2003 19:22:21 -0500',
'RFC822.SIZE': '0'},
}
return self._fetchWork(uid)
def testFetchFastUID(self):
return self.testFetchFast(1)
class DefaultSearchTests(IMAP4HelperMixin, unittest.TestCase):
"""
Test the behavior of the server's SEARCH implementation, particularly in
the face of unhandled search terms.
"""
def setUp(self):
self.server = imap4.IMAP4Server()
self.server.state = 'select'
self.server.mbox = self
self.connected = defer.Deferred()
self.client = SimpleClient(self.connected)
self.msgObjs = [
FakeyMessage({}, (), b'', b'', 999, None),
FakeyMessage({}, (), b'', b'', 10101, None),
FakeyMessage({}, (), b'', b'', 12345, None),
FakeyMessage({}, (), b'', b'', 20001, None),
FakeyMessage({}, (), b'', b'', 20002, None),
]
def fetch(self, messages, uid):
"""
Pretend to be a mailbox and let C{self.server} lookup messages on me.
"""
return list(zip(range(1, len(self.msgObjs) + 1), self.msgObjs))
def _messageSetSearchTest(self, queryTerms, expectedMessages):
"""
Issue a search with given query and verify that the returned messages
match the given expected messages.
@param queryTerms: A string giving the search query.
@param expectedMessages: A list of the message sequence numbers
expected as the result of the search.
@return: A L{Deferred} which fires when the test is complete.
"""
def search():
return self.client.search(queryTerms)
d = self.connected.addCallback(strip(search))
def searched(results):
self.assertEqual(results, expectedMessages)
d.addCallback(searched)
d.addCallback(self._cbStopClient)
d.addErrback(self._ebGeneral)
self.loopback()
return d
def test_searchMessageSet(self):
"""
Test that a search which starts with a message set properly limits
the search results to messages in that set.
"""
return self._messageSetSearchTest(b'1', [1])
def test_searchMessageSetWithStar(self):
"""
If the search filter ends with a star, all the message from the
starting point are returned.
"""
return self._messageSetSearchTest(b'2:*', [2, 3, 4, 5])
def test_searchMessageSetWithStarFirst(self):
"""
If the search filter starts with a star, the result should be identical
with if the filter would end with a star.
"""
return self._messageSetSearchTest(b'*:2', [2, 3, 4, 5])
def test_searchMessageSetUIDWithStar(self):
"""
If the search filter ends with a star, all the message from the
starting point are returned (also for the SEARCH UID case).
"""
return self._messageSetSearchTest(b'UID 10000:*', [2, 3, 4, 5])
def test_searchMessageSetUIDWithStarFirst(self):
"""
If the search filter starts with a star, the result should be identical
with if the filter would end with a star (also for the SEARCH UID case).
"""
return self._messageSetSearchTest(b'UID *:10000', [2, 3, 4, 5])
def test_searchMessageSetUIDWithStarAndHighStart(self):
"""
A search filter of 1234:* should include the UID of the last message in
the mailbox, even if its UID is less than 1234.
"""
# in our fake mbox the highest message UID is 20002
return self._messageSetSearchTest(b'UID 30000:*', [5])
def test_searchMessageSetWithList(self):
"""
If the search filter contains nesting terms, one of which includes a
message sequence set with a wildcard, IT ALL WORKS GOOD.
"""
# 6 is bigger than the biggest message sequence number, but that's
# okay, because N:* includes the biggest message sequence number even
# if N is bigger than that (read the rfc nub).
return self._messageSetSearchTest(b'(6:*)', [5])
def test_searchOr(self):
"""
If the search filter contains an I{OR} term, all messages
which match either subexpression are returned.
"""
return self._messageSetSearchTest(b'OR 1 2', [1, 2])
def test_searchOrMessageSet(self):
"""
If the search filter contains an I{OR} term with a
subexpression which includes a message sequence set wildcard,
all messages in that set are considered for inclusion in the
results.
"""
return self._messageSetSearchTest(b'OR 2:* 2:*', [2, 3, 4, 5])
def test_searchNot(self):
"""
If the search filter contains a I{NOT} term, all messages
which do not match the subexpression are returned.
"""
return self._messageSetSearchTest(b'NOT 3', [1, 2, 4, 5])
def test_searchNotMessageSet(self):
"""
If the search filter contains a I{NOT} term with a
subexpression which includes a message sequence set wildcard,
no messages in that set are considered for inclusion in the
result.
"""
return self._messageSetSearchTest(b'NOT 2:*', [1])
def test_searchAndMessageSet(self):
"""
If the search filter contains multiple terms implicitly
conjoined with a message sequence set wildcard, only the
intersection of the results of each term are returned.
"""
return self._messageSetSearchTest(b'2:* 3', [3])
def test_searchInvalidCriteria(self):
"""
If the search criteria is not a valid key, a NO result is returned to
the client (resulting in an error callback), and an IllegalQueryError is
logged on the server side.
"""
queryTerms = b'FOO'
def search():
return self.client.search(queryTerms)
d = self.connected.addCallback(strip(search))
d = self.assertFailure(d, imap4.IMAP4Exception)
def errorReceived(results):
"""
Verify that the server logs an IllegalQueryError and the
client raises an IMAP4Exception with 'Search failed:...'
"""
self.client.transport.loseConnection()
self.server.transport.loseConnection()
# Check what the server logs
errors = self.flushLoggedErrors(imap4.IllegalQueryError)
self.assertEqual(len(errors), 1)
# Verify exception given to client has the correct message
self.assertEqual(
b"SEARCH failed: Invalid search command FOO", networkString(str(results)))
d.addCallback(errorReceived)
d.addErrback(self._ebGeneral)
self.loopback()
return d
@implementer(imap4.ISearchableMailbox)
class FetchSearchStoreTests(unittest.TestCase, IMAP4HelperMixin):
def setUp(self):
self.expected = self.result = None
self.server_received_query = None
self.server_received_uid = None
self.server_received_parts = None
self.server_received_messages = None
self.server = imap4.IMAP4Server()
self.server.state = 'select'
self.server.mbox = self
self.connected = defer.Deferred()
self.client = SimpleClient(self.connected)
def search(self, query, uid):
# Look for a specific bad query, so we can verify we handle it properly
if query == [b'FOO']:
raise imap4.IllegalQueryError("FOO is not a valid search criteria")
self.server_received_query = query
self.server_received_uid = uid
return self.expected
def addListener(self, *a, **kw):
pass
removeListener = addListener
def _searchWork(self, uid):
def search():
return self.client.search(self.query, uid=uid)
def result(R):
self.result = R
self.connected.addCallback(strip(search)
).addCallback(result
).addCallback(self._cbStopClient
).addErrback(self._ebGeneral)
def check(ignored):
# Ensure no short-circuiting weirdness is going on
self.assertFalse(self.result is self.expected)
self.assertEqual(self.result, self.expected)
self.assertEqual(self.uid, self.server_received_uid)
self.assertEqual(
imap4.parseNestedParens(self.query),
self.server_received_query
)
d = loopback.loopbackTCP(self.server, self.client, noisy=False)
d.addCallback(check)
return d
def testSearch(self):
self.query = imap4.Or(
imap4.Query(header=('subject', 'substring')),
imap4.Query(larger=1024, smaller=4096),
)
self.expected = [1, 4, 5, 7]
self.uid = 0
return self._searchWork(0)
def testUIDSearch(self):
self.query = imap4.Or(
imap4.Query(header=('subject', 'substring')),
imap4.Query(larger=1024, smaller=4096),
)
self.uid = 1
self.expected = [1, 2, 3]
return self._searchWork(1)
def getUID(self, msg):
try:
return self.expected[msg]['UID']
except (TypeError, IndexError):
return self.expected[msg-1]
except KeyError:
return 42
def fetch(self, messages, uid):
self.server_received_uid = uid
self.server_received_messages = str(messages)
return self.expected
def _fetchWork(self, fetch):
def result(R):
self.result = R
self.connected.addCallback(strip(fetch)
).addCallback(result
).addCallback(self._cbStopClient
).addErrback(self._ebGeneral)
def check(ignored):
# Ensure no short-circuiting weirdness is going on
self.assertFalse(self.result is self.expected)
self.parts and self.parts.sort()
self.server_received_parts and self.server_received_parts.sort()
if self.uid:
for (k, v) in self.expected.items():
v['UID'] = str(k)
self.assertEqual(self.result, self.expected)
self.assertEqual(self.uid, self.server_received_uid)
self.assertEqual(self.parts, self.server_received_parts)
self.assertEqual(imap4.parseIdList(self.messages),
imap4.parseIdList(self.server_received_messages))
d = loopback.loopbackTCP(self.server, self.client, noisy=False)
d.addCallback(check)
return d
def test_invalidTerm(self):
"""
If, as part of a search, an ISearchableMailbox raises an
IllegalQueryError (e.g. due to invalid search criteria), client sees a
failure response, and an IllegalQueryError is logged on the server.
"""
query = b'FOO'
def search():
return self.client.search(query)
d = self.connected.addCallback(strip(search))
d = self.assertFailure(d, imap4.IMAP4Exception)
def errorReceived(results):
"""
Verify that the server logs an IllegalQueryError and the
client raises an IMAP4Exception with 'Search failed:...'
"""
self.client.transport.loseConnection()
self.server.transport.loseConnection()
# Check what the server logs
errors = self.flushLoggedErrors(imap4.IllegalQueryError)
self.assertEqual(len(errors), 1)
# Verify exception given to client has the correct message
self.assertEqual(
"SEARCH failed: FOO is not a valid search criteria",
str(results))
d.addCallback(errorReceived)
d.addErrback(self._ebGeneral)
self.loopback()
return d
class FakeMailbox:
def __init__(self):
self.args = []
def addMessage(self, body, flags, date):
self.args.append((body, flags, date))
return defer.succeed(None)
@implementer(imap4.IMessageFile)
class FeaturefulMessage:
def getFlags(self):
return 'flags'
def getInternalDate(self):
return 'internaldate'
def open(self):
return BytesIO(b"open")
@implementer(imap4.IMessageCopier)
class MessageCopierMailbox:
def __init__(self):
self.msgs = []
def copy(self, msg):
self.msgs.append(msg)
return len(self.msgs)
class CopyWorkerTests(unittest.TestCase):
def testFeaturefulMessage(self):
s = imap4.IMAP4Server()
# Yes. I am grabbing this uber-non-public method to test it.
# It is complex. It needs to be tested directly!
# Perhaps it should be refactored, simplified, or split up into
# not-so-private components, but that is a task for another day.
# Ha ha! Addendum! Soon it will be split up, and this test will
# be re-written to just use the default adapter for IMailbox to
# IMessageCopier and call .copy on that adapter.
f = s._IMAP4Server__cbCopy
m = FakeMailbox()
d = f([(i, FeaturefulMessage()) for i in range(1, 11)], 'tag', m)
def cbCopy(results):
for a in m.args:
self.assertEqual(a[0].read(), b"open")
self.assertEqual(a[1], "flags")
self.assertEqual(a[2], "internaldate")
for (status, result) in results:
self.assertTrue(status)
self.assertEqual(result, None)
return d.addCallback(cbCopy)
def testUnfeaturefulMessage(self):
s = imap4.IMAP4Server()
# See above comment
f = s._IMAP4Server__cbCopy
m = FakeMailbox()
msgs = [FakeyMessage({b'Header-Counter': intToBytes(i)}, (), b'Date', b'Body ' + intToBytes(i), i + 10, None) for i in range(1, 11)]
d = f([im for im in zip(range(1, 11), msgs)], 'tag', m)
def cbCopy(results):
seen = []
for a in m.args:
seen.append(a[0].read())
self.assertEqual(a[1], ())
self.assertEqual(a[2], b"Date")
seen.sort()
exp = [b"Header-Counter: " + intToBytes(i) + b"\r\n\r\nBody " +intToBytes(i) for i in range(1, 11)]
exp.sort()
self.assertEqual(seen, exp)
for (status, result) in results:
self.assertTrue(status)
self.assertEqual(result, None)
return d.addCallback(cbCopy)
def testMessageCopier(self):
s = imap4.IMAP4Server()
# See above comment
f = s._IMAP4Server__cbCopy
m = MessageCopierMailbox()
msgs = [object() for i in range(1, 11)]
d = f([im for im in zip(range(1, 11), msgs)], b'tag', m)
def cbCopy(results):
self.assertEqual(results, zip([1] * 10, range(1, 11)))
for (orig, new) in zip(msgs, m.msgs):
self.assertIdentical(orig, new)
return d.addCallback(cbCopy)
class TLSTests(IMAP4HelperMixin, unittest.TestCase):
serverCTX = ServerTLSContext and ServerTLSContext()
clientCTX = ClientTLSContext and ClientTLSContext()
def loopback(self):
return loopback.loopbackTCP(self.server, self.client, noisy=False)
def testAPileOfThings(self):
SimpleServer.theAccount.addMailbox(b'inbox')
called = []
def login():
called.append(None)
return self.client.login(b'testuser', b'password-test')
def list():
called.append(None)
return self.client.list(b'inbox', b'%')
def status():
called.append(None)
return self.client.status(b'inbox', b'UIDNEXT')
def examine():
called.append(None)
return self.client.examine(b'inbox')
def logout():
called.append(None)
return self.client.logout()
self.client.requireTransportSecurity = True
methods = [login, list, status, examine, logout]
map(self.connected.addCallback, map(strip, methods))
self.connected.addCallbacks(self._cbStopClient, self._ebGeneral)
def check(ignored):
self.assertEqual(self.server.startedTLS, True)
self.assertEqual(self.client.startedTLS, True)
self.assertEqual(len(called), len(methods))
d = self.loopback()
d.addCallback(check)
return d
def testLoginLogin(self):
self.server.checker.addUser(b'testuser', b'password-test')
success = []
self.client.registerAuthenticator(imap4.LOGINAuthenticator(b'testuser'))
self.connected.addCallback(
lambda _: self.client.authenticate(b'password-test')
).addCallback(
lambda _: self.client.logout()
).addCallback(success.append
).addCallback(self._cbStopClient
).addErrback(self._ebGeneral)
d = self.loopback()
d.addCallback(lambda x : self.assertEqual(len(success), 1))
return d
def test_startTLS(self):
"""
L{IMAP4Client.startTLS} triggers TLS negotiation and returns a
L{Deferred} which fires after the client's transport is using
encryption.
"""
success = []
self.connected.addCallback(lambda _: self.client.startTLS())
def checkSecure(ignored):
self.assertTrue(
interfaces.ISSLTransport.providedBy(self.client.transport))
self.connected.addCallback(checkSecure)
self.connected.addCallback(self._cbStopClient)
self.connected.addCallback(success.append)
self.connected.addErrback(self._ebGeneral)
d = self.loopback()
d.addCallback(lambda x : self.assertTrue(success))
return defer.gatherResults([d, self.connected])
def testFailedStartTLS(self):
failures = []
def breakServerTLS(ign):
self.server.canStartTLS = False
self.connected.addCallback(breakServerTLS)
self.connected.addCallback(lambda ign: self.client.startTLS())
self.connected.addErrback(
lambda err: failures.append(err.trap(imap4.IMAP4Exception)))
self.connected.addCallback(self._cbStopClient)
self.connected.addErrback(self._ebGeneral)
def check(ignored):
self.assertTrue(failures)
self.assertIdentical(failures[0], imap4.IMAP4Exception)
return self.loopback().addCallback(check)
class SlowMailbox(SimpleMailbox):
howSlow = 2
callLater = None
fetchDeferred = None
# Not a very nice implementation of fetch(), but it'll
# do for the purposes of testing.
def fetch(self, messages, uid):
d = defer.Deferred()
self.callLater(self.howSlow, d.callback, ())
self.fetchDeferred.callback(None)
return d
class TimeoutTests(IMAP4HelperMixin, unittest.TestCase):
def test_serverTimeout(self):
"""
The *client* has a timeout mechanism which will close connections that
are inactive for a period.
"""
c = Clock()
self.server.timeoutTest = True
self.client.timeout = 5 #seconds
self.client.callLater = c.callLater
self.selectedArgs = None
def login():
d = self.client.login(b'testuser', b'password-test')
c.advance(5)
d.addErrback(timedOut)
return d
def timedOut(failure):
self._cbStopClient(None)
failure.trap(error.TimeoutError)
d = self.connected.addCallback(strip(login))
d.addErrback(self._ebGeneral)
return defer.gatherResults([d, self.loopback()])
def test_longFetchDoesntTimeout(self):
"""
The connection timeout does not take effect during fetches.
"""
c = Clock()
SlowMailbox.callLater = c.callLater
SlowMailbox.fetchDeferred = defer.Deferred()
self.server.callLater = c.callLater
SimpleServer.theAccount.mailboxFactory = SlowMailbox
SimpleServer.theAccount.addMailbox('mailbox-test')
self.server.setTimeout(1)
def login():
return self.client.login(b'testuser', b'password-test')
def select():
self.server.setTimeout(1)
return self.client.select('mailbox-test')
def fetch():
return self.client.fetchUID(b'1:*')
def stillConnected():
self.assertNotEqual(self.server.state, 'timeout')
def cbAdvance(ignored):
for i in range(4):
c.advance(.5)
SlowMailbox.fetchDeferred.addCallback(cbAdvance)
d1 = self.connected.addCallback(strip(login))
d1.addCallback(strip(select))
d1.addCallback(strip(fetch))
d1.addCallback(strip(stillConnected))
d1.addCallback(self._cbStopClient)
d1.addErrback(self._ebGeneral)
d = defer.gatherResults([d1, self.loopback()])
return d
def test_idleClientDoesDisconnect(self):
"""
The *server* has a timeout mechanism which will close connections that
are inactive for a period.
"""
c = Clock()
# Hook up our server protocol
transport = StringTransportWithDisconnection()
transport.protocol = self.server
self.server.callLater = c.callLater
self.server.makeConnection(transport)
# Make sure we can notice when the connection goes away
lost = []
connLost = self.server.connectionLost
self.server.connectionLost = lambda reason: (lost.append(None), connLost(reason))[1]
# 2/3rds of the idle timeout elapses...
c.pump([0.0] + [self.server.timeOut / 3.0] * 2)
self.assertFalse(lost, lost)
# Now some more
c.pump([0.0, self.server.timeOut / 2.0])
self.assertTrue(lost)
class DisconnectionTests(unittest.TestCase):
def testClientDisconnectFailsDeferreds(self):
c = imap4.IMAP4Client()
t = StringTransportWithDisconnection()
c.makeConnection(t)
d = self.assertFailure(c.login(b'testuser', 'example.com'), error.ConnectionDone)
c.connectionLost(error.ConnectionDone("Connection closed"))
return d
class SynchronousMailbox(object):
"""
Trivial, in-memory mailbox implementation which can produce a message
synchronously.
"""
def __init__(self, messages):
self.messages = messages
def fetch(self, msgset, uid):
assert not uid, "Cannot handle uid requests."
for msg in msgset:
yield msg, self.messages[msg - 1]
class StringTransportConsumer(StringTransport):
producer = None
streaming = None
def registerProducer(self, producer, streaming):
self.producer = producer
self.streaming = streaming
class PipeliningTests(unittest.TestCase):
"""
Tests for various aspects of the IMAP4 server's pipelining support.
"""
messages = [
FakeyMessage({}, [], b'', b'0', None, None),
FakeyMessage({}, [], b'', b'1', None, None),
FakeyMessage({}, [], b'', b'2', None, None),
]
def setUp(self):
self.iterators = []
self.transport = StringTransportConsumer()
self.server = imap4.IMAP4Server(None, None, self.iterateInReactor)
self.server.makeConnection(self.transport)
def iterateInReactor(self, iterator):
d = defer.Deferred()
self.iterators.append((iterator, d))
return d
def tearDown(self):
self.server.connectionLost(failure.Failure(error.ConnectionDone()))
def test_synchronousFetch(self):
"""
Test that pipelined FETCH commands which can be responded to
synchronously are responded to correctly.
"""
mailbox = SynchronousMailbox(self.messages)
# Skip over authentication and folder selection
self.server.state = 'select'
self.server.mbox = mailbox
# Get rid of any greeting junk
self.transport.clear()
# Here's some pipelined stuff
self.server.dataReceived(
b'01 FETCH 1 BODY[]\r\n'
b'02 FETCH 2 BODY[]\r\n'
b'03 FETCH 3 BODY[]\r\n')
# Flush anything the server has scheduled to run
while self.iterators:
for e in self.iterators[0][0]:
break
else:
self.iterators.pop(0)[1].callback(None)
# The bodies are empty because we aren't simulating a transport
# exactly correctly (we have StringTransportConsumer but we never
# call resumeProducing on its producer). It doesn't matter: just
# make sure the surrounding structure is okay, and that no
# exceptions occurred.
self.assertEqual(
self.transport.value(),
b'* 1 FETCH (BODY[] )\r\n'
b'01 OK FETCH completed\r\n'
b'* 2 FETCH (BODY[] )\r\n'
b'02 OK FETCH completed\r\n'
b'* 3 FETCH (BODY[] )\r\n'
b'03 OK FETCH completed\r\n')
if ClientTLSContext is None:
for case in (TLSTests,):
case.skip = "OpenSSL not present"
elif interfaces.IReactorSSL(reactor, None) is None:
for case in (TLSTests,):
case.skip = "Reactor doesn't support SSL"
class IMAP4ServerFetchTests(unittest.TestCase):
"""
This test case is for the FETCH tests that require
a C{StringTransport}.
"""
def setUp(self):
self.transport = StringTransport()
self.server = imap4.IMAP4Server()
self.server.state = 'select'
self.server.makeConnection(self.transport)
def test_fetchWithPartialValidArgument(self):
"""
If by any chance, extra bytes got appended at the end of a valid
FETCH arguments, the client should get a BAD - arguments invalid
response.
See U{RFC 3501},
section 6.4.5,
"""
# We need to clear out the welcome message.
self.transport.clear()
# Let's send out the faulty command.
self.server.dataReceived(b"0001 FETCH 1 FULLL\r\n")
expected = b"0001 BAD Illegal syntax: Invalid Argument\r\n"
self.assertEqual(self.transport.value(), expected)
self.transport.clear()
self.server.connectionLost(error.ConnectionDone("Connection closed"))