summaryrefslogtreecommitdiff
path: root/tests/integration/mail/imap/regressions_mime_struct
diff options
context:
space:
mode:
authorKali Kaneko (leap communications) <kali@leap.se>2016-09-01 00:06:52 -0400
committerKali Kaneko (leap communications) <kali@leap.se>2016-09-01 00:06:52 -0400
commitf826bc473a0c50fcf55f4e8609aa07622814f902 (patch)
tree32665c6608c536c3b3db5b3fa504567043171c91 /tests/integration/mail/imap/regressions_mime_struct
parentc74c51f9fc753c6a870f7c14d5fdd10b152e0991 (diff)
[tests] move tests to root folder
Diffstat (limited to 'tests/integration/mail/imap/regressions_mime_struct')
-rwxr-xr-xtests/integration/mail/imap/regressions_mime_struct461
1 files changed, 461 insertions, 0 deletions
diff --git a/tests/integration/mail/imap/regressions_mime_struct b/tests/integration/mail/imap/regressions_mime_struct
new file mode 100755
index 00000000..03326646
--- /dev/null
+++ b/tests/integration/mail/imap/regressions_mime_struct
@@ -0,0 +1,461 @@
+#!/usr/bin/env python
+
+# -*- coding: utf-8 -*-
+# regression_mime_struct
+# Copyright (C) 2014 LEAP
+# Copyright (c) Twisted Matrix Laboratories.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+"""
+Simple Regression Tests for checking MIME struct handling using IMAP4 client.
+
+Iterates trough all mails under a given folder and tries to APPEND them to
+the server being tested. After FETCHING the pushed message, it compares
+the received version with the one that was saved, and exits with an error
+code if they do not match.
+"""
+import os
+import StringIO
+import sys
+
+from email.parser import Parser
+
+from twisted.internet import protocol
+from twisted.internet import ssl
+from twisted.internet import defer
+from twisted.internet import stdio
+from twisted.mail import imap4
+from twisted.protocols import basic
+from twisted.python import log
+
+
+REGRESSIONS_FOLDER = os.environ.get(
+ "REGRESSIONS_FOLDER", "regressions_test")
+print "[+] Using regressions folder:", REGRESSIONS_FOLDER
+
+parser = Parser()
+
+
+def get_msg_parts(raw):
+ """
+ Return a representation of the parts of a message suitable for
+ comparison.
+
+ :param raw: string for the message
+ :type raw: str
+ """
+ m = parser.parsestr(raw)
+ return [dict(part.items())
+ if part.is_multipart()
+ else part.get_payload()
+ for part in m.walk()]
+
+
+def compare_msg_parts(a, b):
+ """
+ Compare two sequences of parts of messages.
+
+ :param a: part sequence for message a
+ :param b: part sequence for message b
+
+ :return: True if both message sequences are equivalent.
+ :rtype: bool
+ """
+ # XXX This could be smarter and show the differences in the
+ # different parts when/where they differ.
+ #import pprint; pprint.pprint(a[0])
+ #import pprint; pprint.pprint(b[0])
+
+ def lowerkey(d):
+ return dict((k.lower(), v.replace('\r', ''))
+ for k, v in d.iteritems())
+
+ def eq(x, y):
+ # For dicts, we compare a variation with their keys
+ # in lowercase, and \r removed from their values
+ if all(map(lambda i: isinstance(i, dict), (x, y))):
+ x, y = map(lowerkey, (x, y))
+ return x == y
+
+ compare_vector = map(lambda tup: eq(tup[0], tup[1]), zip(a, b))
+ all_match = all(compare_vector)
+
+ if not all_match:
+ print "PARTS MISMATCH!"
+ print "vector: ", compare_vector
+ index = compare_vector.index(False)
+ from pprint import pprint
+ print "Expected:"
+ pprint(a[index])
+ print ("***")
+ print "Found:"
+ pprint(b[index])
+ print
+
+ return all_match
+
+
+def get_fd(string):
+ """
+ Return a file descriptor with the passed string
+ as content.
+ """
+ fd = StringIO.StringIO()
+ fd.write(string)
+ fd.seek(0)
+ return fd
+
+
+class TrivialPrompter(basic.LineReceiver):
+ promptDeferred = None
+
+ def prompt(self, msg):
+ assert self.promptDeferred is None
+ self.display(msg)
+ self.promptDeferred = defer.Deferred()
+ return self.promptDeferred
+
+ def display(self, msg):
+ self.transport.write(msg)
+
+ def lineReceived(self, line):
+ if self.promptDeferred is None:
+ return
+ d, self.promptDeferred = self.promptDeferred, None
+ d.callback(line)
+
+
+class SimpleIMAP4Client(imap4.IMAP4Client):
+ """
+ A client with callbacks for greeting messages from an IMAP server.
+ """
+ greetDeferred = None
+
+ def serverGreeting(self, caps):
+ self.serverCapabilities = caps
+ if self.greetDeferred is not None:
+ d, self.greetDeferred = self.greetDeferred, None
+ d.callback(self)
+
+
+class SimpleIMAP4ClientFactory(protocol.ClientFactory):
+ usedUp = False
+ protocol = SimpleIMAP4Client
+
+ def __init__(self, username, onConn):
+ self.ctx = ssl.ClientContextFactory()
+
+ self.username = username
+ self.onConn = onConn
+
+ def buildProtocol(self, addr):
+ """
+ Initiate the protocol instance. Since we are building a simple IMAP
+ client, we don't bother checking what capabilities the server has. We
+ just add all the authenticators twisted.mail has. Note: Gmail no
+ longer uses any of the methods below, it's been using XOAUTH since
+ 2010.
+ """
+ assert not self.usedUp
+ self.usedUp = True
+
+ p = self.protocol(self.ctx)
+ p.factory = self
+ p.greetDeferred = self.onConn
+
+ p.registerAuthenticator(imap4.PLAINAuthenticator(self.username))
+ p.registerAuthenticator(imap4.LOGINAuthenticator(self.username))
+ p.registerAuthenticator(
+ imap4.CramMD5ClientAuthenticator(self.username))
+
+ return p
+
+ def clientConnectionFailed(self, connector, reason):
+ d, self.onConn = self.onConn, None
+ d.errback(reason)
+
+
+def cbServerGreeting(proto, username, password):
+ """
+ Initial callback - invoked after the server sends us its greet message.
+ """
+ # Hook up stdio
+ tp = TrivialPrompter()
+ stdio.StandardIO(tp)
+
+ # And make it easily accessible
+ proto.prompt = tp.prompt
+ proto.display = tp.display
+
+ # Try to authenticate securely
+ return proto.authenticate(
+ password).addCallback(
+ cbAuthentication,
+ proto).addErrback(
+ ebAuthentication, proto, username, password
+ )
+
+
+def ebConnection(reason):
+ """
+ Fallback error-handler. If anything goes wrong, log it and quit.
+ """
+ log.startLogging(sys.stdout)
+ log.err(reason)
+ return reason
+
+
+def cbAuthentication(result, proto):
+ """
+ Callback after authentication has succeeded.
+
+ Lists a bunch of mailboxes.
+ """
+ return proto.select(
+ REGRESSIONS_FOLDER
+ ).addCallback(
+ cbSelectMbox, proto
+ ).addErrback(
+ ebSelectMbox, proto, REGRESSIONS_FOLDER)
+
+
+def ebAuthentication(failure, proto, username, password):
+ """
+ Errback invoked when authentication fails.
+
+ If it failed because no SASL mechanisms match, offer the user the choice
+ of logging in insecurely.
+
+ If you are trying to connect to your Gmail account, you will be here!
+ """
+ failure.trap(imap4.NoSupportedAuthentication)
+ return InsecureLogin(proto, username, password)
+
+
+def InsecureLogin(proto, username, password):
+ """
+ Raise insecure-login error.
+ """
+ return proto.login(
+ username, password
+ ).addCallback(
+ cbAuthentication, proto)
+
+
+def cbSelectMbox(result, proto):
+ """
+ Callback invoked when select command finishes successfully.
+
+ If any message is in the test folder, it will flag them as deleted and
+ expunge.
+ If no messages found, it will start with the APPEND tests.
+ """
+ print "SELECT: %s EXISTS " % result.get("EXISTS", "??")
+
+ if result["EXISTS"] != 0:
+ # Flag as deleted, expunge, and do an examine again.
+ print "There is mail here, will delete..."
+ return cbDeleteAndExpungeTestFolder(proto)
+
+ else:
+ return cbAppendNextMessage(proto)
+
+
+def ebSelectMbox(failure, proto, folder):
+ """
+ Errback invoked when the examine command fails.
+
+ Creates the folder.
+ """
+ log.err(failure)
+ log.msg("Folder %r does not exist. Creating..." % (folder,))
+ return proto.create(folder).addCallback(cbAuthentication, proto)
+
+
+def ebExpunge(failure):
+ log.err(failure)
+
+
+def cbDeleteAndExpungeTestFolder(proto):
+ """
+ Callback invoked fom cbExamineMbox when the number of messages in the
+ mailbox is not zero. It flags all messages as deleted and expunge the
+ mailbox.
+ """
+ return proto.setFlags(
+ "1:*", ("\\Deleted",)
+ ).addCallback(
+ lambda r: proto.expunge()
+ ).addCallback(
+ cbExpunge, proto
+ ).addErrback(
+ ebExpunge)
+
+
+def cbExpunge(result, proto):
+ return proto.select(
+ REGRESSIONS_FOLDER
+ ).addCallback(
+ cbSelectMbox, proto
+ ).addErrback(ebSettingDeleted, proto)
+
+
+def ebSettingDeleted(failure, proto):
+ """
+ Report errors during deletion of messages in the mailbox.
+ """
+ print failure.getTraceback()
+
+
+def cbAppendNextMessage(proto):
+ """
+ Appends the next message in the global queue to the test folder.
+ """
+ # 1. Get the next test message from global tuple.
+ try:
+ next_sample = SAMPLES.pop()
+ except IndexError:
+ # we're done!
+ return proto.logout()
+
+ print "\nAPPEND %s" % (next_sample,)
+ raw = open(next_sample).read()
+ msg = get_fd(raw)
+ return proto.append(
+ REGRESSIONS_FOLDER, msg
+ ).addCallback(
+ lambda r: proto.select(REGRESSIONS_FOLDER)
+ ).addCallback(
+ cbAppend, proto, raw
+ ).addErrback(
+ ebAppend, proto, raw)
+
+
+def cbAppend(result, proto, orig_msg):
+ """
+ Fetches the message right after an append.
+ """
+ # XXX keep account of highest UID
+ uid = "1:*"
+
+ return proto.fetchSpecific(
+ '%s' % uid,
+ headerType='',
+ headerArgs=['BODY.PEEK[]'],
+ ).addCallback(
+ cbCompareMessage, proto, orig_msg
+ ).addErrback(ebAppend, proto, orig_msg)
+
+
+def ebAppend(failure, proto, raw):
+ """
+ Errorback for the append operation
+ """
+ print "ERROR WHILE APPENDING!"
+ print failure.getTraceback()
+
+
+def cbPickMessage(result, proto):
+ """
+ Pick a message.
+ """
+ return proto.fetchSpecific(
+ '%s' % result,
+ headerType='',
+ headerArgs=['BODY.PEEK[]'],
+ ).addCallback(cbCompareMessage, proto)
+
+
+def cbCompareMessage(result, proto, raw):
+ """
+ Display message and compare it with the original one.
+ """
+ parts_orig = get_msg_parts(raw)
+
+ if result:
+ keys = result.keys()
+ keys.sort()
+ else:
+ print "[-] GOT NO RESULT"
+ return proto.logout()
+
+ latest = max(keys)
+
+ fetched_msg = result[latest][0][2]
+ parts_fetched = get_msg_parts(fetched_msg)
+
+ equal = compare_msg_parts(
+ parts_orig,
+ parts_fetched)
+
+ if equal:
+ print "[+] MESSAGES MATCH"
+ return cbAppendNextMessage(proto)
+ else:
+ print "[-] ERROR: MESSAGES DO NOT MATCH !!!"
+ print " ABORTING COMPARISON..."
+ # FIXME logout and print the subject ...
+ return proto.logout()
+
+
+def cbClose(result):
+ """
+ Close the connection when we finish everything.
+ """
+ from twisted.internet import reactor
+ reactor.stop()
+
+
+def main():
+ import glob
+ import sys
+
+ if len(sys.argv) != 4:
+ print "Usage: regressions <user> <pass> <samples-folder>"
+ sys.exit()
+
+ hostname = "localhost"
+ port = "1984"
+ username = sys.argv[1]
+ password = sys.argv[2]
+
+ samplesdir = sys.argv[3]
+
+ if not os.path.isdir(samplesdir):
+ print ("Could not find samples folder! "
+ "Make sure of copying mail_breaker contents there.")
+ sys.exit()
+
+ samples = glob.glob(samplesdir + '/*')
+
+ global SAMPLES
+ SAMPLES = []
+ SAMPLES += samples
+
+ onConn = defer.Deferred(
+ ).addCallback(
+ cbServerGreeting, username, password
+ ).addErrback(
+ ebConnection
+ ).addBoth(cbClose)
+
+ factory = SimpleIMAP4ClientFactory(username, onConn)
+
+ from twisted.internet import reactor
+ reactor.connectTCP(hostname, int(port), factory)
+ reactor.run()
+
+
+if __name__ == '__main__':
+ main()