test_sb_imapfilter.py

用python实现的邮件过滤器

                if uid:
                    msg_uid = int(msg)
                else:
                    msg_uid = IMAP_UIDS[int(msg)]
                msg_text = IMAP_MESSAGES[msg_uid]
                headers, unused = msg_text.split('\r\n\r\n', 1)
                response[msg].append(("FETCH (RFC822.HEADER {%s}" %
                                      (len(headers),), headers))
        if msg_parts.find("FLAGS INTERNALDATE") != -1:
            # We make up flags & dates.
            for msg in msg_nums:
                response[msg].append('FETCH (FLAGS (\Seen \Deleted) '
                                     'INTERNALDATE "27-Jul-2004 13:1'
                                     '1:56 +1200')
        for msg in msg_nums:
            try:
                simple = " ".join(response[msg])
            except TypeError:
                simple = []
                for part in response[msg]:
                    if isinstance(part, types.StringTypes):
                        simple.append(part)
                    else:
                        simple.append('%s\r\n%s)' % (part[0], part[1]))
                simple = " ".join(simple)
            response[msg] = "* %s %s" % (msg, simple)
        response_text = "\r\n".join(response.values())
        return "%s\r\n%s OK FETCH completed\r\n" % (response_text, id)

    def onUID(self, id, command, args, uid=False):
        actual_command, args = args.split(None, 1)
        handler = self.handlers.get(actual_command, self.onUnknown)
        return handler(id, actual_command, args, uid=True)

    def onUnknown(self, id, command, args, uid=False):
        """Unknown IMAP4 command."""
        return "%s BAD Command unrecognised: %s\r\n" % (id, repr(command))


class BaseIMAPFilterTest(unittest.TestCase):
    def setUp(self):
        self.imap = IMAPSession("localhost", IMAP_PORT)

    def tearDown(self):
        try:
            self.imap.logout()
        except imaplib.error:
            pass


class IMAPSessionTest(BaseIMAPFilterTest):
    def testConnection(self):
        # Connection is made in setup, just need to check
        # that it worked.
        self.assert_(self.imap.connected)
        
    def testGoodLogin(self):
        self.imap.login(IMAP_USERNAME, IMAP_PASSWORD)
        self.assert_(self.imap.logged_in)

    def testBadLogin(self):
        print "\nYou should see a message indicating that login failed."
        self.assertRaises(LoginFailure, self.imap.login, IMAP_USERNAME,
                          "wrong password")

    def test_check_response(self):
        test_data = "IMAP response data"
        response = ("OK", test_data)
        data = self.imap.check_response("", response)
        self.assertEqual(data, test_data)
        response = ("NO", test_data)
        self.assertRaises(BadIMAPResponseError, self.imap.check_response,
                          "", response)

    def testSelectFolder(self):
        # This test will fail if testGoodLogin fails.
        self.imap.login(IMAP_USERNAME, IMAP_PASSWORD)
        
        # Check handling of Python (not SpamBayes) bug #845560.
        self.assertRaises(BadIMAPResponseError, self.imap.SelectFolder, "")

        # Check selection.
        self.imap.SelectFolder("Inbox")
        response = self.imap.response('OK')
        self.assertEquals(response[0], "OK")
        self.assert_(response[1] != [None])

        # Check that we don't reselect if we are already in that folder.
        self.imap.SelectFolder("Inbox")
        response = self.imap.response('OK')
        self.assertEquals(response[0], "OK")
        self.assertEquals(response[1], [None])

    def test_folder_list(self):
        global FAIL_NEXT

        # This test will fail if testGoodLogin fails.
        self.imap.login(IMAP_USERNAME, IMAP_PASSWORD)

        # Everything working.        
        folders = self.imap.folder_list()
        correct = IMAP_FOLDER_LIST[:]
        correct.sort()
        self.assertEqual(folders, correct)

        # Bad command.
        print "\nYou should see a message indicating that getting the " \
              "folder list failed."
        FAIL_NEXT = True
        self.assertEqual(self.imap.folder_list(), [])

        # Literals in response.
        # XXX TO DO!
        
    def test_extract_fetch_data(self):
        response = "bad response"
        self.assertRaises(BadIMAPResponseError,
                          self.imap.extract_fetch_data, response)

        # Check UID and message_number.
        message_number = "123"
        uid = "5432"
        response = ("%s (UID %s)" % (message_number, uid),)
        data = self.imap.extract_fetch_data(response)
        self.assertEqual(data[message_number]["message_number"],
                         message_number)
        self.assertEqual(data[message_number]["UID"], uid)

        # Check INTERNALDATE, FLAGS.
        flags = r"(\Seen \Deleted)"
        date = '"27-Jul-2004 13:11:56 +1200"'
        response = ("%s (FLAGS %s INTERNALDATE %s)" % \
                   (message_number, flags, date),)
        data = self.imap.extract_fetch_data(response)
        self.assertEqual(data[message_number]["message_number"],
                         message_number)
        self.assertEqual(data[message_number]["FLAGS"], flags)
        self.assertEqual(data[message_number]["INTERNALDATE"], date)

        # Check RFC822 and literals.
        rfc = "Subject: Test\r\n\r\nThis is a test message."
        response = (("%s (RFC822 {%s}" % (message_number, len(rfc)), rfc),)
        data = self.imap.extract_fetch_data(response)
        self.assertEqual(data[message_number]["message_number"],
                         message_number)
        self.assertEqual(data[message_number]["RFC822"], rfc)

        # Check RFC822.HEADER.
        headers = "Subject: Foo\r\nX-SpamBayes-ID: 1231-1\r\n"
        response = (("%s (RFC822.HEADER {%s}" % (message_number,
                                                len(headers)), headers),)
        data = self.imap.extract_fetch_data(response)
        self.assertEqual(data[message_number]["RFC822.HEADER"], headers)

        # Check BODY.PEEK.
        peek = "Subject: Test2\r\n\r\nThis is another test message."
        response = (("%s (BODY[] {%s}" % (message_number, len(peek)),
                    peek),)
        data = self.imap.extract_fetch_data(response)
        self.assertEqual(data[message_number]["BODY[]"], peek)

        # A more complcated test with more than one message number.
        uid = '3018'
        flags = '(\\Seen \\Deleted)'
        headers = "Return-Path: <[e-mail]>\r\nX-Original-To" \
                  ": [e-mail]\r\nDelivered-To: dleinbac@ma" \
                  "il2.majro.dhs.org\r\nReceived: from its-mail1.mass" \
                  "ey.ac.nz (its-mail1.massey.ac.nz [130.123.128.11])" \
                  "\r\n\tby mail2.majro.dhs.org (Postfix) with ESMTP " \
                  "id 7BC5018FE22\r\n\tfor <[e-mail]>; Mon" \
                  ", 13 Dec 2004 22:46:05 -0800 (PST)\r\nReceived: fr" \
                  "om its-mm1.massey.ac.nz (its-mm1 [130.123.128.45])" \
                  "\r\n\tby its-mail1.massey.ac.nz (8.9.3/8.9.3) with" \
                  "ESMTP id TAA12081;\r\n\tTue, 14 Dec 2004 19:45:56 " \
                  "+1300 (NZDT)\r\nReceived: from its-campus2.massey." \
                  "ac.nz (Not Verified[130.123.48.254]) by its-mm1.ma" \
                  "ssey.ac.nz with NetIQ MailMarshal\r\n\tid <B003745" \
                  "788>; Tue, 14 Dec 2004 19:45:56 +1300\r\nReceived:" \
                  "from it029048 (it029048.massey.ac.nz [130.123.238." \
                  "51])\r\n\tby its-campus2.massey.ac.nz (8.9.3/8.9.3" \
                  ") with ESMTP id TAA05881;\r\n\tTue, 14 Dec 2004 19" \
                  ':45:55 +1300 (NZDT)\r\nFrom: "Tony Meyer"  <tameye' \
                  '[e-mail]>\r\nTo: "\'David Leinbach\'" <david@l' \
                  "einbach.name>, <[e-mail]>\r\nSubject: " \
                  "RE: [Spambayes] KeyError in sp_imapfilter\r\nDate:" \
                  "Tue, 14 Dec 2004 19:45:25 +1300\r\nMessage-ID: <EC" \
                  "[e-mail]" \
                  "sey.ac.nz>\r\nMIME-Version: 1.0\r\nContent-Type: t" \
                  'ext/plain;\r\n\tcharset="us-ascii"\r\nContent-Tran' \
                  "sfer-Encoding: quoted-printable\r\nX-Priority: 3 (" \
                  "Normal)\r\nX-MSMail-Priority: Normal\r\nX-Mailer: " \
                  "Microsoft Outlook, Build 10.0.4510\r\nIn-Reply-To:" \
                  "<ECBA357DDED63B4995F5C1F5CBE5B1E801A49168@its-xchg" \
                  ".massey.ac.nz>\r\nX-Habeas-SWE-1: winter into spri" \
                  "ng\r\nX-Habeas-SWE-2: brightly anticipated\r\nX-Ha" \
                  "beas-SWE-3: like Habeas SWE (tm)\r\nX-Habeas-SWE-4" \
                  ": Copyright 2002 Habeas (tm)\r\nX-Habeas-SWE-5: Se" \
                  "nder Warranted Email (SWE) (tm). The sender of thi" \
                  "s\r\nX-Habeas-SWE-6: email in exchange for a licen" \
                  "se for this Habeas\r\nX-Habeas-SWE-7: warrant mark" \
                  "warrants that this is a Habeas Compliant\r\nX-Habe" \
                  "as-SWE-8: Message (HCM) and  not spam. Please repo" \
                  "rt use of this\r\nX-Habeas-SWE-9: mark in spam to " \
                  "<http://www.habeas.com/report/>.\r\nX-MimeOLE: Pro" \
                  "duced By Microsoft MimeOLE V6.00.2900.2180\r\nImpo" \
                  "rtance: Normal\r\n\r\n"
        response = ['1 (FLAGS %s)' % flags,
                    ('2 (UID %s RFC822.HEADER {%d}' % (uid, len(headers)),
                     headers), ')'] 
        data = self.imap.extract_fetch_data(response)
        self.assertEqual(data['1']["message_number"], '1')
        self.assertEqual(data['2']["message_number"], '2')
        self.assertEqual(data['1']["FLAGS"], flags)
        self.assertEqual(data['2']["UID"], uid)
        self.assertEqual(data['2']["RFC822.HEADER"], headers)

        # Another complicated one (like the previous, but the two message
        # numbers are the same).
        headers = 'Return-Path: <[e-mail]>\r\n' \
                  'X-Original-To: [e-mail]\r\n' \
                  'Delivered-To: [e-mail]\r\n' \
                  'Received: from cressida (unknown [70.70.206.137])' \
                  '\r\n\tby mail2.majro.dhs.org (Postfix) with ESMTP' \
                  'id AAD451CD28E\r\n\tfor <[e-mail]>; Mo' \
                  'n,  6 Dec 2004 11:49:41 -0800 (PST)\r\n' \
                  'From: "Erin Leinbach" <[e-mail]>\r\n' \
                  'To: "Dave" <[e-mail]>\r\n' \
                  'Subject: Goo goo dolls songs\r\n' \
                  'Date: Mon, 6 Dec 2004 11:51:07 -0800\r\n' \
                  'Message-ID: <000001c4dbcc$ed6188a0$6801a8c0@cre' \
                  'ssida>\r\nMIME-Version: 1.0\r\n' \
                  'Content-Type: text/plain;\r\n' \
                  '\tcharset="Windows-1252"\r\n' \
                  'Content-Transfer-Encoding: 7bit\r\n' \
                  'X-Priority: 3 (Normal)\r\n' \
                  'X-MSMail-Priority: Normal\r\n' \
                  'X-Mailer: Microsoft Outlook, Build 10.0.2616\r\n' \
                  'Importance: Normal\r\n' \
                  'X-MimeOLE: Produced By Microsoft MimeOLE V6.00.29' \
                  '00.2180\r\nX-Spambayes-Classification: ham\r\n' \
                  'X-Spambayes-MailId: 000001\r\n\r\n'
        uid = '3086'
        flags = '(\\Seen \\Deleted)'
        response = [('5 (UID %s RFC822.HEADER {839}' % (uid,), headers),
                    ')', '5 (FLAGS %s)' % (flags,)]
        data = self.imap.extract_fetch_data(response)
        self.assertEqual(data['5']["message_number"], '5')
        self.assertEqual(data['5']["FLAGS"], flags)
        self.assertEqual(data['5']["UID"], uid)
        self.assertEqual(data['5']["RFC822.HEADER"], headers)

    def _counter(self, size):
        self._count += 1
        return self._imap_file_read(size)
    def test_safe_read(self):
        # Ensure that safe_read only gets self.imap.MAXIMUM_SAFE_READ bytes
        # at a time, and that it does collect everything.
        # Setup a fake file to read from.
        saved_file = self.imap.file
        self.imap.file = StringIO.StringIO()
        self.imap.file.write("".join(IMAP_MESSAGES.values()*10))
        self.imap.file.seek(0)
        try:
            # First check when the size is less than the maximum.
            self.assertEqual(len(self.imap.read(\
                self.imap.MAXIMUM_SAFE_READ-1)),
                             self.imap.MAXIMUM_SAFE_READ-1)
            # Check when the size is more than the maximum.
            self.assertEqual(len(self.imap.read(\
                self.imap.MAXIMUM_SAFE_READ+1)),
                             self.imap.MAXIMUM_SAFE_READ+1)
            # Check that the read is called once when the size is smaller.
            self._count = 0
            self._imap_file_read = self.imap.file.read
            self.imap.file.read = self._counter
            self.imap.read(self.imap.MAXIMUM_SAFE_READ-1)
            self.assertEqual(self._count, 1)
            # Check that the read is called twice when the size is larger.
            self._count = 0
            self.imap.read(self.imap.MAXIMUM_SAFE_READ+1)
            self.assertEqual(self._count, 2)