← Back to branch summary

~launchpad-pqm/launchpad/devel

« back to all changes in this revision

Viewing changes to lib/lp/services/mail/tests/test_incoming.py

  • Committer: Launchpad Patch Queue Manager
  • Date: 2011-05-30 18:00:36 UTC
  • mfrom: (13125.3.4 duplicate-msgid)
  • Revision ID: launchpad@pqm.canonical.com-20110530180036-xkh210pbaxm8q3ss
[r=adeuring][bug=595166] Skip duplicate incoming messages.

Show diffs side-by-side

added added

removed removed

Lines of Context:
lib/lp/services/mail/tests/test_incoming.py
1
 
# Copyright 2009 Canonical Ltd.  This software is licensed under the
 
1
# Copyright 2009, 2011 Canonical Ltd.  This software is licensed under the
2
2
# GNU Affero General Public License version 3 (see the file LICENSE).
3
3
 
4
4
from doctest import DocTestSuite
6
6
import os
7
7
import unittest
8
8
 
9
 
from testtools.matchers import Equals, Is
 
9
from testtools.matchers import (
 
10
    Equals,
 
11
    Is,
 
12
    MatchesRegex,
 
13
    )
10
14
import transaction
11
15
from zope.security.management import setSecurityPolicy
12
16
 
24
28
    authenticateEmail,
25
29
    extract_addresses,
26
30
    handleMail,
 
31
    handle_one_mail,
27
32
    MailErrorUtility,
28
33
    ORIGINAL_TO_HEADER,
29
34
    )
30
35
from lp.services.mail.sendmail import MailController
31
36
from lp.services.mail.stub import TestMailer
 
37
from lp.services.messages.model.message import MessageSet
 
38
from lp.services.mail.tests import (
 
39
    active_handler,
 
40
    ListHandler,
 
41
    )
32
42
from lp.testing import TestCaseWithFactory
33
43
from lp.testing.factory import GPGSigningContext
34
44
from lp.testing.mail_helpers import pop_notifications
122
132
        mail = self.factory.makeSignedMessage(email_address=email)
123
133
        self.assertThat(authenticateEmail(mail), Is(None))
124
134
 
 
135
    def test_handle_one_mail_duplicate_message_id(self):
 
136
        # An incoming mail with the message-id of an already-processed mail is
 
137
        # skipped.
 
138
        orig_message = self.factory.makeMessage()
 
139
        handler_calls = []
 
140
        self.assertEqual([], handler_calls)
 
141
        duplicate_mail = self.factory.makeSignedMessage(
 
142
            msgid=orig_message.rfc822msgid,
 
143
            to_address='new@random.example.com')
 
144
        log = BufferLogger()
 
145
        with active_handler('random.example.com', ListHandler()) as handler:
 
146
            handle_one_mail(log, duplicate_mail, None, None, None)
 
147
        self.assertEqual([], handler.processed)
 
148
        messages = MessageSet().get(orig_message.rfc822msgid)
 
149
        self.assertEqual(1, len(messages))
 
150
        matches = MatchesRegex(
 
151
            '(.|\n)*WARNING Message with id .* already stored.  Skipping.')
 
152
        self.assertThat(log.getLogBuffer(), matches)
 
153
 
125
154
 
126
155
class TestExtractAddresses(TestCaseWithFactory):
127
156