test_sb_imapfilter.py

用python实现的邮件过滤器

        finally:
            self.imap.file = saved_file


class IMAPMessageTest(BaseIMAPFilterTest):
    def setUp(self):
        BaseIMAPFilterTest.setUp(self)
        self.msg = IMAPMessage()
        self.msg.imap_server = self.imap

    # These tests might fail if more than one second passes
    # between the call and the assert.  We could make it more robust,
    # or you could just run this on a faster machine, like me <wink>.
    def test_extract_time_no_date(self):
        date = self.msg.extractTime()
        self.assertEqual(date, imaplib.Time2Internaldate(time.time()))
    def test_extract_time_date(self):
        self.msg["Date"] = "Wed, 19 May 2004 20:05:15 +1200"
        date = self.msg.extractTime()
        self.assertEqual(date, '"19-May-2004 20:05:15 +1200"')
    def test_extract_time_bad_date(self):
        self.msg["Date"] = "Mon, 06 May 0102 10:51:16 -0100"
        date = self.msg.extractTime()
        self.assertEqual(date, imaplib.Time2Internaldate(time.time()))

    def test_as_string_invalid(self):
        content = "This is example content.\nThis is more\r\n"
        self.msg.invalid = True
        self.msg.invalid_content = content
        as_string = self.msg.as_string()
        self.assertEqual(self.msg._force_CRLF(content), as_string)

    def testMoveTo(self):
        fol1 = "Folder1"
        fol2 = "Folder2"
        self.msg.MoveTo(fol1)
        self.assertEqual(self.msg.folder, fol1)
        self.msg.MoveTo(fol2)
        self.assertEqual(self.msg.previous_folder, fol1)
        self.assertEqual(self.msg.folder, fol2)

    def test_get_full_message(self):
        self.assertRaises(AssertionError, self.msg.get_full_message)
        self.msg.id = "unittest"
        self.assertRaises(AttributeError, self.msg.get_full_message)

        self.msg.imap_server.login(IMAP_USERNAME, IMAP_PASSWORD)
        self.msg.imap_server.select()
        response = self.msg.imap_server.fetch(1, "UID")
        self.assertEqual(response[0], "OK")
        self.msg.uid = response[1][0][7:-1]
        self.msg.folder = IMAPFolder("Inbox", self.msg.imap_server, None)

        new_msg = self.msg.get_full_message()
        self.assertEqual(new_msg.folder, self.msg.folder)
        self.assertEqual(new_msg.previous_folder, self.msg.previous_folder)
        self.assertEqual(new_msg.uid, self.msg.uid)
        self.assertEqual(new_msg.id, self.msg.id)
        self.assertEqual(new_msg.rfc822_key, self.msg.rfc822_key)
        self.assertEqual(new_msg.rfc822_command, self.msg.rfc822_command)
        self.assertEqual(new_msg.imap_server, self.msg.imap_server)
        id_header = options["Headers", "mailid_header_name"]
        self.assertEqual(new_msg[id_header], self.msg.id)

        new_msg2 = new_msg.get_full_message()
        # These should be the same object, not just equal.
        self.assert_(new_msg is new_msg2)

    def test_get_bad_message(self):
        self.msg.id = "unittest"
        self.msg.imap_server.login(IMAP_USERNAME, IMAP_PASSWORD)
        self.msg.imap_server.select()
        self.msg.uid = 103 # id of malformed message in dummy server
        self.msg.folder = IMAPFolder("Inbox", self.msg.imap_server, None)
        print "\nWith email package versions less than 3.0, you should " \
              "see an error parsing the message."
        new_msg = self.msg.get_full_message()
        # With Python < 2.4 (i.e. email < 3.0) we get an exception
        # header.  With more recent versions, we get a defects attribute.
        # XXX I can't find a message that generates a defect!  Until
        # message 103 is replaced with one that does, this will fail with
        # Python 2.4/email 3.0.
        has_header = new_msg.as_string().find("X-Spambayes-Exception: ") != -1
        has_defect = hasattr(new_msg, "defects") and len(new_msg.defects) > 0
        self.assert_(has_header or has_defect)

    def test_get_memory_error_message(self):
        # XXX Figure out a way to trigger a memory error - but not in
        # the fake IMAP server, in imaplib, or our IMAP class.
        pass

    def test_Save(self):
        # XXX To-do
        pass


class IMAPFolderTest(BaseIMAPFilterTest):
    def setUp(self):
        BaseIMAPFilterTest.setUp(self)
        self.imap.login(IMAP_USERNAME, IMAP_PASSWORD)
        self.folder = IMAPFolder("testfolder", self.imap, None)

    def test_cmp(self):
        folder2 = IMAPFolder("testfolder", self.imap, None)
        folder3 = IMAPFolder("testfolder2", self.imap, None)
        self.assertEqual(self.folder, folder2)
        self.assertNotEqual(self.folder, folder3)
        
    def test_iter(self):
        keys = self.folder.keys()
        for msg in self.folder:
            msg = msg.get_full_message()
            msg_correct = email.message_from_string(IMAP_MESSAGES[int(keys[0])],
                                                    _class=message.Message)
            id_header_name = options["Headers", "mailid_header_name"]
            if msg_correct[id_header_name] is None:
                msg_correct[id_header_name] = msg.id
            self.assertEqual(msg.as_string(), msg_correct.as_string())
            keys = keys[1:]

    def test_keys(self):
        keys = self.folder.keys()
        # We get back UIDs, not IDs, so convert to check.
        correct_keys = [str(IMAP_UIDS[id]) for id in UNDELETED_IDS]
        self.assertEqual(keys, correct_keys)

    def test_getitem_new_style(self):
        # 101 already has a suitable (new style) id, so it should
        # not be recreated.
        id_header_name = options["Headers", "mailid_header_name"]
        msg1 = self.folder[101]
        self.assertEqual(msg1.id, SB_ID_1)
        msg1 = msg1.get_full_message()
        msg1_correct = email.message_from_string(IMAP_MESSAGES[101],
                                                 message.Message)
        self.assertNotEqual(msg1[id_header_name], None)
        msg1_correct[id_header_name] = SB_ID_1
        self.assertEqual(msg1.as_string(), msg1_correct.as_string())

    def test_getitem_old_style(self):
        # 102 already has a suitable (old style) id, so it should
        # not be recreated.  We should be sure to use the old id,
        # rather than the new one, too, for backwards compatibility.
        id_header_name = options["Headers", "mailid_header_name"]
        msg2 = self.folder[102]
        self.assertEqual(msg2.id, SB_ID_2)
        msg2 = msg2.get_full_message()
        self.assertNotEqual(msg2[id_header_name], None)
        self.assertEqual(msg2.as_string(), IMAP_MESSAGES[102])

    def test_getitem_new_id(self):
        # 104 doesn't have an id, so should be recreated with one.
        id_header_name = options["Headers", "mailid_header_name"]
        msg3 = self.folder[104]
        self.assertNotEqual(msg3[id_header_name], None)
        msg_correct = email.message_from_string(IMAP_MESSAGES[104],
                                                message.Message)
        msg_correct[id_header_name] = msg3.id
        self.assertEqual(msg3.as_string(), msg_correct.as_string())

    def test_generate_id(self):
        print "\nThis test takes slightly over a second."
        id1 = self.folder._generate_id()
        id2 = self.folder._generate_id()
        id3 = self.folder._generate_id()
        # Need to wait at least one clock tick.
        time.sleep(1)
        id4 = self.folder._generate_id()
        self.assertEqual(id2, id1 + "-2")
        self.assertEqual(id3, id1 + "-3")
        self.assertNotEqual(id1, id4)
        self.assertNotEqual(id2, id4)
        self.assertNotEqual(id3, id4)
        self.assert_('-' not in id4)
        
    def test_Train(self):
        # XXX To-do
        pass
    def test_Filter(self):
        # XXX To-do
        pass


class IMAPFilterTest(BaseIMAPFilterTest):
    def setUp(self):
        BaseIMAPFilterTest.setUp(self)
        self.imap.login(IMAP_USERNAME, IMAP_PASSWORD)
        classifier = Classifier()
        self.filter = IMAPFilter(classifier, None)
        options["imap", "ham_train_folders"] = ("ham_to_train",)
        options["imap", "spam_train_folders"] = ("spam_to_train",)

    def test_Train(self):
        # XXX To-do
        pass
    def test_Filter(self):
        # XXX To-do
        pass


class SFBugsTest(BaseIMAPFilterTest):
    def test_802545(self):
        # Test that the filter selects each folder before expunging,
        # and that it was logged in in the first place.
        pass

    def test_816400(self):
        # Test that bad dates don't cause an error in appending.
        # (also sf #890645)
        # e.g. 31-Dec-1969 16:00:18 +0100
        #  Date: Mon, 06 May 0102 10:51:16 -0100
        # Date: Sat, 08 Jun 0102 19:44:54 -0700
        # Date: 16 Mar 80 8:16:44 AM
        pass

    def test_818552(self):
        # Test that, when saving, we remove the RECENT flag including
        # the space after it.
        pass

    def test_842984(self):
        # Confirm that if webbrowser.open_new() fails, we print a
        # message saying "Please point your web browser at
        # http://localhost:8880/" rather than bombing out.
        pass

    def test_886133(self):
        # Check that folder names with characters not allowed in XML
        # are correctly handled for the web interface.
        pass


class InterfaceTest(unittest.TestCase):
    def setUp(self):
        print "\nThis test takes slightly over one second."
        self.saved_server = options["imap", "server"]
        options["imap", "server"] = ""
        thread.start_new_thread(run, (True,))
        # Wait for it to be ready.
        time.sleep(1)

    def tearDown(self):
        options["imap", "server"] = self.saved_server
        # Shutdown as though through the web UI.
        from urllib import urlopen, urlencode
        urlopen('http://localhost:%d/save' % options["html_ui", "port"],
                urlencode({'how': _('Save & shutdown')})).read()

    def test_UI(self):
        # Smoke-test the HTML UI.
        httpServer = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        httpServer.connect(('localhost', options["html_ui", "port"]))
        httpServer.send("get / HTTP/1.0\r\n\r\n")
        response = ''
        while 1:
            packet = httpServer.recv(1024)
            if not packet: break
            response += packet
        self.assert_(re.search(r"(?s)<html>.*SpamBayes IMAP Filter.*</html>",
                               response))


def suite():
    suite = unittest.TestSuite()
    for cls in (IMAPSessionTest,
                IMAPMessageTest,
                IMAPFolderTest,
                IMAPFilterTest,
                SFBugsTest,
                InterfaceTest,
               ):
        suite.addTest(unittest.makeSuite(cls))
    return suite

if __name__=='__main__':
    def runTestServer():
        TestListener()
        asyncore.loop()
    thread.start_new_thread(runTestServer, ())
    sb_test_support.unittest_main(argv=sys.argv + ['suite'])