← Back to branch summary

~launchpad-pqm/launchpad/devel

« back to all changes in this revision

Viewing changes to lib/canonical/archivepublisher/ftests/test_uploadprocessor.py

  • Committer: Christian Reis
  • Date: 2007-05-04 17:48:03 UTC
  • mto: This revision was merged to the branch mainline in revision 4206.
  • Revision ID: christian.reis@canonical.com-20070504174803-ij77q4jwkm5l8mbk
Break apart upload code into canonical/archiveuploader

Show diffs side-by-side

added added

removed removed

Lines of Context:
lib/canonical/archivepublisher/ftests/test_uploadprocessor.py
1
 
# Copyright 2006 Canonical Ltd.  All rights reserved.
2
 
 
3
 
"""Functional tests for uploadprocessor.py."""
4
 
 
5
 
__metaclass__ = type
6
 
 
7
 
import os
8
 
from shutil import rmtree
9
 
from tempfile import mkdtemp
10
 
import unittest
11
 
 
12
 
from zope.component import getUtility
13
 
 
14
 
from canonical.archivepublisher.tests.test_uploadprocessor import (
15
 
    MockOptions, MockLogger)
16
 
from canonical.archivepublisher.uploadpolicy import AbstractUploadPolicy
17
 
from canonical.archivepublisher.uploadprocessor import UploadProcessor
18
 
from canonical.config import config
19
 
from canonical.database.constants import UTC_NOW
20
 
from canonical.launchpad.ftests import (
21
 
    import_public_test_keys, syncUpdate)
22
 
from canonical.launchpad.interfaces import (
23
 
    IDistributionSet, IDistroReleaseSet)
24
 
from canonical.launchpad.mail import stub
25
 
from canonical.lp.dbschema import (
26
 
    DistributionReleaseStatus, DistroReleaseQueueStatus,
27
 
    PackagePublishingStatus, PackagePublishingPocket)
28
 
from canonical.testing import LaunchpadZopelessLayer
29
 
 
30
 
 
31
 
class BrokenUploadPolicy(AbstractUploadPolicy):
32
 
    """A broken upload policy, to test error handling."""
33
 
 
34
 
    def __init__(self):
35
 
        AbstractUploadPolicy.__init__(self)
36
 
        self.name = "broken"
37
 
        self.unsigned_changes_ok = True
38
 
        self.unsigned_dsc_ok = True
39
 
 
40
 
    def checkUpload(self, upload):
41
 
        """Raise an exception upload processing is not expecting."""
42
 
        raise Exception("Exception raised by BrokenUploadPolicy for testing.")
43
 
 
44
 
 
45
 
class TestUploadProcessorBase(unittest.TestCase):
46
 
    """Base class for functional tests over uploadprocessor.py."""
47
 
    layer = LaunchpadZopelessLayer
48
 
 
49
 
    def setUp(self):
50
 
        self.queue_folder = mkdtemp()
51
 
        os.makedirs(os.path.join(self.queue_folder, "incoming"))
52
 
 
53
 
        self.test_files_dir = os.path.join(config.root,
54
 
            "lib/canonical/archivepublisher/tests/data/suite")
55
 
 
56
 
        import_public_test_keys()
57
 
 
58
 
        self.options = MockOptions()
59
 
        self.options.base_fsroot = self.queue_folder
60
 
        self.options.leafname = None
61
 
        self.options.distro = "ubuntu"
62
 
        self.options.distrorelease = None
63
 
        self.options.nomails = False
64
 
        self.options.context = 'insecure'
65
 
 
66
 
        self.log = MockLogger()
67
 
 
68
 
    def tearDown(self):
69
 
        rmtree(self.queue_folder)
70
 
 
71
 
    def assertLogContains(self, line):
72
 
        """Assert if a given line is present in the log messages."""
73
 
        self.assertTrue(line in self.log.lines)
74
 
 
75
 
    def setupBreezy(self):
76
 
        """Create a fresh distrorelease in ubuntu.
77
 
 
78
 
        Use *initialiseFromParent* procedure to create 'breezy'
79
 
        on ubuntu based on the last 'breezy-autotest'.
80
 
 
81
 
        Also sets 'changeslist' and 'nominatedarchindep' properly.
82
 
        """
83
 
        ubuntu = getUtility(IDistributionSet).getByName('ubuntu')
84
 
        bat = ubuntu['breezy-autotest']
85
 
        dr_set = getUtility(IDistroReleaseSet)
86
 
        breezy = dr_set.new(
87
 
            ubuntu, 'breezy', 'Breezy Badger',
88
 
            'The Breezy Badger', 'Black and White', 'Someone',
89
 
            '5.10', bat, bat.owner)
90
 
        breezy_i386 = breezy.newArch('i386', bat['i386'].processorfamily,
91
 
                                     True, breezy.owner)
92
 
        breezy.nominatedarchindep = breezy_i386
93
 
        breezy.changeslist = 'breezy-changes@ubuntu.com'
94
 
        breezy.initialiseFromParent()
95
 
        self.breezy = breezy
96
 
 
97
 
 
98
 
class TestUploadProcessor(TestUploadProcessorBase):
99
 
    """Basic tests on uploadprocessor class.
100
 
 
101
 
    * Check if the rejection message is send even when an unexpected
102
 
      exception occur when processing the upload.
103
 
    * Check if known uploads targeted to a FROZEN distrorelease
104
 
      end up in UNAPPROVED queue.
105
 
 
106
 
    This test case is able to setup a fresh distrorelease in Ubuntu.
107
 
    """
108
 
 
109
 
    def testRejectionEmailForUnhandledException(self):
110
 
        """Test there's a rejection email when nascentupload breaks.
111
 
 
112
 
        If a developer makes an upload which finds a bug in nascentupload,
113
 
        and an unhandled exception occurs, we should try to send a
114
 
        rejection email. We'll test that this works, in a case where we
115
 
        will have the right information to send the email before the
116
 
        error occurs.
117
 
 
118
 
        If we haven't extracted enough information to send a rejection
119
 
        email when things break, trying to send one will raise a new
120
 
        exception, and the upload will fail silently as before. We don't
121
 
        test this case.
122
 
 
123
 
        See bug 35965.
124
 
        """
125
 
        # Register our broken upload policy
126
 
        AbstractUploadPolicy._registerPolicy(BrokenUploadPolicy)
127
 
        self.options.context = 'broken'
128
 
        uploadprocessor = UploadProcessor(
129
 
            self.options, self.layer.txn, self.log)
130
 
 
131
 
        # Place a suitable upload in the queue. This one is one of
132
 
        # Daniel's.
133
 
        os.system("cp -a %s %s" %
134
 
            (os.path.join(self.test_files_dir, "baz_1.0-1"),
135
 
             os.path.join(self.queue_folder, "incoming")))
136
 
 
137
 
        # Try to process it
138
 
        uploadprocessor.processChangesFile(
139
 
            os.path.join(self.queue_folder, "incoming", "baz_1.0-1"),
140
 
            "baz_1.0-1_source.changes")
141
 
 
142
 
        # Check the mailer stub has a rejection email for Daniel
143
 
        from_addr, to_addrs, raw_msg = stub.test_emails.pop()
144
 
        daniel = "Daniel Silverstone <daniel.silverstone@canonical.com>"
145
 
        self.assertEqual(to_addrs, [daniel])
146
 
        self.assertTrue("Unhandled exception processing upload: Exception "
147
 
                        "raised by BrokenUploadPolicy for testing." in raw_msg)
148
 
 
149
 
    def testUploadToFrozenDistro(self):
150
 
        """Uploads to a frozen distrorelease should work, but be unapproved.
151
 
 
152
 
        The rule for a frozen distrorelease is that uploads should still
153
 
        be permitted, but that the usual rule for auto-accepting uploads
154
 
        of existing packages should be suspended. New packages will still
155
 
        go into NEW, but new versions will be UNAPPROVED, rather than
156
 
        ACCEPTED.
157
 
 
158
 
        To test this, we will upload two versions of the same package,
159
 
        accepting and publishing the first, and freezing the distrorelease
160
 
        before the second. If all is well, the second upload should go
161
 
        through ok, but end up in status UNAPPROVED, and with the
162
 
        appropriate email contents.
163
 
 
164
 
        See bug 58187.
165
 
        """
166
 
        # Extra setup for breezy
167
 
        self.setupBreezy()
168
 
        self.layer.txn.commit()
169
 
 
170
 
        # Set up the uploadprocessor with appropriate options and logger
171
 
        uploadprocessor = UploadProcessor(
172
 
            self.options, self.layer.txn, self.log)
173
 
 
174
 
        # Place a suitable upload in the queue. This is a source upload
175
 
        # for breezy.
176
 
        os.system("cp -a %s %s" %
177
 
            (os.path.join(self.test_files_dir, "bar_1.0-1"),
178
 
             os.path.join(self.queue_folder, "incoming")))
179
 
 
180
 
        # Process
181
 
        uploadprocessor.processChangesFile(
182
 
            os.path.join(self.queue_folder, "incoming", "bar_1.0-1"),
183
 
            "bar_1.0-1_source.changes")
184
 
 
185
 
        # Check it went ok to the NEW queue and all is going well so far.
186
 
        from_addr, to_addrs, raw_msg = stub.test_emails.pop()
187
 
        foo_bar = "Foo Bar <foo.bar@canonical.com>"
188
 
        daniel = "Daniel Silverstone <daniel.silverstone@canonical.com>"
189
 
        self.assertEqual([e.strip() for e in to_addrs], [foo_bar, daniel])
190
 
        self.assertTrue(
191
 
            "NEW" in raw_msg, "Expected email containing NEW: %s" % raw_msg)
192
 
 
193
 
        # Accept and publish the upload.
194
 
        # This is required so that the next upload of a later version of
195
 
        # the same package will work correctly.
196
 
        queue_items = self.breezy.getQueueItems(
197
 
            status=DistroReleaseQueueStatus.NEW, name="bar",
198
 
            version="1.0-1", exact_match=True)
199
 
        self.assertEqual(queue_items.count(), 1)
200
 
        queue_item = queue_items[0]
201
 
 
202
 
        queue_item.setAccepted()
203
 
        pubrec = queue_item.sources[0].publish(self.log)
204
 
        pubrec.status = PackagePublishingStatus.PUBLISHED
205
 
        pubrec.datepublished = UTC_NOW
206
 
 
207
 
        # Make ubuntu/breezy a frozen distro, so a source upload for an
208
 
        # existing package will be allowed, but unapproved.
209
 
        self.breezy.releasestatus = DistributionReleaseStatus.FROZEN
210
 
        self.layer.txn.commit()
211
 
 
212
 
        # Place a newer version of bar into the queue.
213
 
        os.system("cp -a %s %s" %
214
 
            (os.path.join(self.test_files_dir, "bar_1.0-2"),
215
 
             os.path.join(self.queue_folder, "incoming")))
216
 
 
217
 
        # Try to process it
218
 
        uploadprocessor.processChangesFile(
219
 
            os.path.join(self.queue_folder, "incoming", "bar_1.0-2"),
220
 
            "bar_1.0-2_source.changes")
221
 
 
222
 
        # Verify we get an email talking about awaiting approval.
223
 
        from_addr, to_addrs, raw_msg = stub.test_emails.pop()
224
 
        daniel = "Daniel Silverstone <daniel.silverstone@canonical.com>"
225
 
        foo_bar = "Foo Bar <foo.bar@canonical.com>"
226
 
        self.assertEqual([e.strip() for e in to_addrs], [foo_bar, daniel])
227
 
        self.assertTrue("This upload awaits approval" in raw_msg,
228
 
                        "Expected an 'upload awaits approval' email.\n"
229
 
                        "Got:\n%s" % raw_msg)
230
 
 
231
 
        # And verify that the queue item is in the unapproved state.
232
 
        queue_items = self.breezy.getQueueItems(
233
 
            status=DistroReleaseQueueStatus.UNAPPROVED, name="bar",
234
 
            version="1.0-2", exact_match=True)
235
 
        self.assertEqual(queue_items.count(), 1)
236
 
        queue_item = queue_items[0]
237
 
        self.assertEqual(
238
 
            queue_item.status, DistroReleaseQueueStatus.UNAPPROVED,
239
 
            "Expected queue item to be in UNAPPROVED status.")
240
 
 
241
 
 
242
 
def test_suite():
243
 
    return unittest.TestLoader().loadTestsFromName(__name__)
244
 
 
245