Logo Search packages:      
Sourcecode: mailping version File versions

test_incoming.py

from twisted.trial import unittest
import os, sets
from cStringIO import StringIO

from MailPing import maildir, incoming, fileutil

class TestIncoming(unittest.TestCase):
    def setUp(self):
        self.statedir = self.mktemp()
        os.mkdir(self.statedir)
        os.mkdir(os.path.join(self.statedir, 'pending'))

        for name in ['incoming', 'broken', 'junk']:
            path = os.path.join(self.statedir, name)
            os.mkdir(path)
            os.mkdir(os.path.join(path, 'cur'))
            os.mkdir(os.path.join(path, 'new'))
            os.mkdir(os.path.join(path, 'tmp'))

    def assert_empty_maildir(self, maildir):
        for subdir in ['cur', 'new', 'tmp']:
            path = os.path.join(self.statedir, maildir, subdir)
            contents = os.listdir(path)
            self.assertEquals(contents, [],
                                  'Maildir %r should be empty, but contains %r' % (path, contents))

    def assert_maildirs_empty(self):
        for maildir in ['incoming', 'broken', 'junk']:
            self.assert_empty_maildir(maildir)

    def assert_pending_empty(self):
        path = os.path.join(self.statedir, 'pending')
        contents = os.listdir(path)
        self.assertEquals(contents, [],
                          'Pending %r should be empty, but contains %r' % (path, contents))

    def assert_no_success(self):
        path = os.path.join(self.statedir, 'success')
        self.failIf(os.path.exists(path))

        path = os.path.join(self.statedir, 'success.msg')
        self.failIf(os.path.exists(path))

        path = os.path.join(self.statedir, 'latency')
        self.failIf(os.path.exists(path))

    def assert_success(self, timestamp, content=None, latency=None):
        path = os.path.join(self.statedir, 'success')

        lastSuccess = fileutil.getTime(path)
        self.assertEquals(lastSuccess, timestamp)

        if content is not None:
            text = file(path+'.msg').read()
            self.assertEquals(content, text)

        if latency is not None:
            got = fileutil.getTime(os.path.join(self.statedir, 'latency'))
            self.assertEquals(got, latency)

    def testProcessing_junk(self):
        text = '''\
From: Foo Bar <foo.bar@thud.invalid>
To: Mr. You <you@destination.invalid>
Subject: SPAM SPAM SPAM!
Reply-To: admin@foo.invalid

content is just junk
'''
        fileutil.writeFile(
            os.path.join(self.statedir, 'incoming', 'new',
                         '42.P34Q0.testhost'),
            text)
        incoming.process(self.statedir)
        filename = os.path.join(self.statedir, 'junk', 'new',
                                '42.P34Q0.testhost')
        self.failUnless(os.path.isfile(filename))
        content = file(filename).read()
        self.assertEquals(text, content)

        self.assert_empty_maildir('incoming')
        self.assert_empty_maildir('broken')
        self.assert_pending_empty()
        self.assert_no_success()

    def testProcessing_broken(self):
        text = '''\
From: Foo Bar <foo.bar@thud.invalid>
To: Mr. You <you@destination.invalid>
Subject: Mail ping f0e241677fa9f2e0d7884776e4a6846b19703e6d
Reply-To: admin@foo.invalid

content is just junk
'''
        fileutil.writeFile(
            os.path.join(self.statedir, 'incoming', 'new',
                         '42.P34Q0.testhost'),
            text)
        incoming.process(self.statedir)
        filename = os.path.join(self.statedir, 'broken', 'new',
                                '42.P34Q0.testhost')
        self.failUnless(os.path.isfile(filename))
        content = file(filename).read()
        self.assertEquals(text, content)

        self.assert_empty_maildir('incoming')
        self.assert_empty_maildir('junk')
        self.assert_pending_empty()
        self.assert_no_success()

    def testProcessing_older(self):
        text = '''\
From: Foo Bar <foo.bar@thud.invalid>
To: Mr. You <you@destination.invalid>
Subject: Mail ping f0e241677fa9f2e0d7884776e4a6846b19703e6d
Reply-To: admin@foo.invalid

content is just junk
'''

        fileutil.writeFile(os.path.join(self.statedir, 'success'),
                           '42\n')

        fileutil.writeFile(
            os.path.join(self.statedir, 'incoming', 'new',
                         '43.P34Q0.testhost'),
            text)

        fileutil.writeFile(
            os.path.join(self.statedir, 'pending',
                         'f0e241677fa9f2e0d7884776e4a6846b19703e6d'),
            '13\n')

        incoming.process(self.statedir)

        self.assert_maildirs_empty()
        self.assert_pending_empty()

        self.assert_success(content=None, timestamp=42, latency=-1)

    def testProcessing_newer(self):
        text = '''\
From: Foo Bar <foo.bar@thud.invalid>
To: Mr. You <you@destination.invalid>
Subject: Mail ping f0e241677fa9f2e0d7884776e4a6846b19703e6d
Reply-To: admin@foo.invalid

content is just junk
'''

        fileutil.writeFile(os.path.join(self.statedir, 'success'),
                           '42\n')

        fileutil.writeFile(
            os.path.join(self.statedir, 'incoming', 'new',
                         '45.P34Q0.testhost'),
            text)

        fileutil.writeFile(
            os.path.join(self.statedir, 'pending',
                         'f0e241677fa9f2e0d7884776e4a6846b19703e6d'),
            '43\n')

        incoming.process(self.statedir)

        self.assert_maildirs_empty()
        self.assert_pending_empty()

        self.assert_success(content=text,
                            timestamp=43,
                            latency=2)

    def testProcessing_Latency_BackInTime(self):
        text = '''\
From: Foo Bar <foo.bar@thud.invalid>
To: Mr. You <you@destination.invalid>
Subject: Mail ping f0e241677fa9f2e0d7884776e4a6846b19703e6d
Reply-To: admin@foo.invalid

content is just junk
'''

        fileutil.writeFile(os.path.join(self.statedir, 'success'),
                           '42\n')

        fileutil.writeFile(
            os.path.join(self.statedir, 'incoming', 'new',
                         '42.P34Q0.testhost'),
            text)

        fileutil.writeFile(
            os.path.join(self.statedir, 'pending',
                         'f0e241677fa9f2e0d7884776e4a6846b19703e6d'),
            '43\n')

        incoming.process(self.statedir)

        self.assert_maildirs_empty()
        self.assert_pending_empty()

        self.assert_success(content=text,
                            timestamp=43,
                            latency=-1)

    def testProcessing_Latency_Unparseable(self):
        text = '''\
From: Foo Bar <foo.bar@thud.invalid>
To: Mr. You <you@destination.invalid>
Subject: Mail ping f0e241677fa9f2e0d7884776e4a6846b19703e6d
Reply-To: admin@foo.invalid

content is just junk
'''

        fileutil.writeFile(os.path.join(self.statedir, 'success'),
                           '42\n')

        fileutil.writeFile(
            os.path.join(self.statedir, 'incoming', 'new',
                         'garbage-in-nothing-out'),
            text)

        fileutil.writeFile(
            os.path.join(self.statedir, 'pending',
                         'f0e241677fa9f2e0d7884776e4a6846b19703e6d'),
            '43\n')

        incoming.process(self.statedir)

        self.assert_maildirs_empty()
        self.assert_pending_empty()

        self.assert_success(content=text,
                            timestamp=43,
                            latency=-1)

    def testReceiveToNonExisting(self):
        text = "Message for you, Sir"
        msg = StringIO(text)
        path = os.path.join(self.statedir, 'nonexisting')
        maildir.deliverToMaildir(path, msg)

        new = os.path.join(path, 'new')
        self.failUnless(os.path.isdir(new))
        l = os.listdir(new)
        self.assertEquals(len(l), 1)
        f = file(os.path.join(new, l[0]))
        got = f.read()
        self.assertEquals(got, text)

Generated by  Doxygen 1.6.0   Back to index