~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-31 00:00:38 UTC
  • mfrom: (13139.1.1 rollback-13139)
  • Revision ID: launchpad@pqm.canonical.com-20110531000038-s0tqzkfwiymxeez0
[rs=wgrant][rollback=13139] Revert r13139. Messages are not unique
 over all of LP.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright 2009, 2011 Canonical Ltd.  This software is licensed under the
 
1
# Copyright 2009 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 (
10
 
    Equals,
11
 
    Is,
12
 
    MatchesRegex,
13
 
    )
 
9
from testtools.matchers import Equals, Is
14
10
import transaction
15
11
from zope.security.management import setSecurityPolicy
16
12
 
28
24
    authenticateEmail,
29
25
    extract_addresses,
30
26
    handleMail,
31
 
    handle_one_mail,
32
27
    MailErrorUtility,
33
28
    ORIGINAL_TO_HEADER,
34
29
    )
35
30
from lp.services.mail.sendmail import MailController
36
31
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
 
    )
42
32
from lp.testing import TestCaseWithFactory
43
33
from lp.testing.factory import GPGSigningContext
44
34
from lp.testing.mail_helpers import pop_notifications
132
122
        mail = self.factory.makeSignedMessage(email_address=email)
133
123
        self.assertThat(authenticateEmail(mail), Is(None))
134
124
 
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
 
 
154
125
 
155
126
class TestExtractAddresses(TestCaseWithFactory):
156
127