← Back to branch summary

~launchpad-pqm/launchpad/devel

« back to all changes in this revision

Viewing changes to lib/lp/archiveuploader/tests/test_sync_notification.py

  • Committer: Launchpad Patch Queue Manager
  • Date: 2012-01-06 07:15:43 UTC
  • mfrom: (14634.1.6 bug-876594)
  • Revision ID: launchpad@pqm.canonical.com-20120106071543-ym23hw45nfuskgre
[r=jcsackett][bug=876594] Don't notify upstream authors of build
 failures of synced sources.

Show diffs side-by-side

added added

removed removed

Lines of Context:
lib/lp/archiveuploader/tests/test_sync_notification.py
 
1
# Copyright 2012 Canonical Ltd.  This software is licensed under the
 
2
# GNU Affero General Public License version 3 (see the file LICENSE).
 
3
 
 
4
"""Test notification behaviour for cross-distro package syncs."""
 
5
 
 
6
__metaclass__ = type
 
7
 
 
8
import os.path
 
9
 
 
10
from zope.component import getUtility
 
11
 
 
12
from lp.archiveuploader.nascentupload import (
 
13
    NascentUpload,
 
14
    UploadError,
 
15
    )
 
16
from lp.registry.interfaces.pocket import PackagePublishingPocket
 
17
from lp.services.log.logger import DevNullLogger
 
18
from lp.soyuz.enums import (
 
19
    ArchivePermissionType,
 
20
    SourcePackageFormat,
 
21
    )
 
22
from lp.soyuz.interfaces.sourcepackageformat import (
 
23
    ISourcePackageFormatSelectionSet,
 
24
    )
 
25
from lp.soyuz.model.archivepermission import ArchivePermission
 
26
from lp.soyuz.scripts.packagecopier import do_copy
 
27
from lp.testing import (
 
28
    login,
 
29
    TestCaseWithFactory,
 
30
    )
 
31
from lp.testing.fakemethod import FakeMethod
 
32
from lp.testing.layers import LaunchpadZopelessLayer
 
33
from lp.testing.mail_helpers import pop_notifications
 
34
 
 
35
 
 
36
class FakeUploadPolicy:
 
37
    def __init__(self, spph):
 
38
        self.distroseries = spph.distroseries
 
39
        self.archive = spph.distroseries.main_archive
 
40
        self.pocket = spph.pocket
 
41
 
 
42
    setDistroSeriesAndPocket = FakeMethod()
 
43
    validateUploadType = FakeMethod()
 
44
    checkUpload = FakeMethod()
 
45
 
 
46
 
 
47
class FakeChangesFile:
 
48
    def __init__(self, spph, file_path):
 
49
        self.files = []
 
50
        self.filepath = file_path
 
51
        self.filename = os.path.basename(file_path)
 
52
        self.architectures = ['i386']
 
53
        self.suite_name = '-'.join([spph.distroseries.name, spph.pocket.name])
 
54
        self.raw_content = open(file_path).read()
 
55
        self.signingkey = None
 
56
 
 
57
    checkFileName = FakeMethod([])
 
58
    processAddresses = FakeMethod([])
 
59
    processFiles = FakeMethod([])
 
60
    verify = FakeMethod([UploadError("Deliberately broken")])
 
61
 
 
62
 
 
63
class TestSyncNotification(TestCaseWithFactory):
 
64
 
 
65
    layer = LaunchpadZopelessLayer
 
66
 
 
67
    def makePersonWithEmail(self):
 
68
        """Create a person; return (person, email)."""
 
69
        address = "%s@example.com" % self.factory.getUniqueString()
 
70
        person = self.factory.makePerson(email=address)
 
71
        return person, address
 
72
 
 
73
    def makeSPPH(self, distroseries, maintainer_address):
 
74
        """Create a `SourcePackagePublishingHistory`."""
 
75
        return self.factory.makeSourcePackagePublishingHistory(
 
76
            distroseries=distroseries, pocket=PackagePublishingPocket.RELEASE,
 
77
            dsc_maintainer_rfc822=maintainer_address)
 
78
 
 
79
    def makeUploader(self, person, archive, component):
 
80
        """Grant a person upload privileges for archive/component."""
 
81
        ArchivePermission(
 
82
            person=person, archive=archive, component=component,
 
83
            permission=ArchivePermissionType.UPLOAD)
 
84
 
 
85
    def syncSource(self, spph, target_distroseries, requester):
 
86
        """Sync `spph` into `target_distroseries`."""
 
87
        getUtility(ISourcePackageFormatSelectionSet).add(
 
88
            target_distroseries, SourcePackageFormat.FORMAT_1_0)
 
89
        target_archive = target_distroseries.main_archive
 
90
        self.makeUploader(requester, target_archive, spph.component)
 
91
        [synced_spph] = do_copy(
 
92
            [spph], target_archive, target_distroseries,
 
93
            pocket=spph.pocket, person=requester, allow_delayed_copies=False,
 
94
            close_bugs=False)
 
95
        return synced_spph
 
96
 
 
97
    def makeChangesFile(self, spph, maintainer, maintainer_address,
 
98
                        changer, changer_address):
 
99
        temp_dir = self.makeTemporaryDirectory()
 
100
        changes_file = os.path.join(
 
101
            temp_dir, "%s.changes" % spph.source_package_name)
 
102
        with open(changes_file, 'w') as changes:
 
103
            changes.write(
 
104
                "Maintainer: %s <%s>\n"
 
105
                "Changed-By: %s <%s>\n"
 
106
                % (
 
107
                    maintainer.name,
 
108
                    maintainer_address,
 
109
                    changer.name,
 
110
                    changer_address,
 
111
                    ))
 
112
        return FakeChangesFile(spph, changes_file)
 
113
 
 
114
    def makeNascentUpload(self, spph, maintainer, maintainer_address,
 
115
                          changer, changer_address):
 
116
        """Create a `NascentUpload` for `spph`."""
 
117
        changes = self.makeChangesFile(
 
118
            spph, maintainer, maintainer_address, changer, changer_address)
 
119
        upload = NascentUpload(
 
120
            changes, FakeUploadPolicy(spph), DevNullLogger())
 
121
        upload.queue_root = upload._createQueueEntry()
 
122
        das = self.factory.makeDistroArchSeries(
 
123
            distroseries=spph.distroseries)
 
124
        bpb = self.factory.makeBinaryPackageBuild(
 
125
            source_package_release=spph.sourcepackagerelease,
 
126
            archive=spph.archive, distroarchseries=das, pocket=spph.pocket,
 
127
            sourcepackagename=spph.sourcepackagename)
 
128
        upload.queue_root.addBuild(bpb)
 
129
        return upload
 
130
 
 
131
    def processAndRejectUpload(self, nascent_upload):
 
132
        nascent_upload.process()
 
133
        # Obtain the required privileges for do_reject.
 
134
        login('foo.bar@canonical.com')
 
135
        nascent_upload.do_reject(notify=True)
 
136
 
 
137
    def getNotifiedAddresses(self):
 
138
        """Get email addresses that were notified."""
 
139
        return [message['to'] for message in pop_notifications()]
 
140
 
 
141
    def test_failed_copy_builds_do_not_spam_upstream(self):
 
142
        """Failed builds do not spam people who are not responsible for them.
 
143
 
 
144
        We import Debian source packages, then sync them into Ubuntu (and
 
145
        from there, into Ubuntu-derived distros).  Those syncs then trigger
 
146
        builds that the original Debian maintainers and last-change authors
 
147
        are not responsible for.
 
148
 
 
149
        In a situation like that, we should not bother those people with the
 
150
        failure.  We notify the person who requested the sync instead.
 
151
 
 
152
        (The logic in lp.soyuz.adapters.notification may still notify the
 
153
        author of the last change, if that person is also an uploader for the
 
154
        archive that the failure happened in.  For this particular situation
 
155
        we consider that not so much an intended behaviour, as an emergent one
 
156
        that does not seem inappropriate.  It'd be hard to change if we wanted
 
157
        to.)
 
158
 
 
159
        This test guards against bug 876594.
 
160
        """
 
161
        maintainer, maintainer_address = self.makePersonWithEmail()
 
162
        changer, changer_address = self.makePersonWithEmail()
 
163
        dsp = self.factory.makeDistroSeriesParent()
 
164
        original_spph = self.makeSPPH(dsp.parent_series, maintainer_address)
 
165
        sync_requester, syncer_address = self.makePersonWithEmail()
 
166
        synced_spph = self.syncSource(
 
167
            original_spph, dsp.derived_series, sync_requester)
 
168
        nascent_upload = self.makeNascentUpload(
 
169
            synced_spph, maintainer, maintainer_address,
 
170
            changer, changer_address)
 
171
        pop_notifications()
 
172
        self.processAndRejectUpload(nascent_upload)
 
173
 
 
174
        notified_addresses = '\n'.join(self.getNotifiedAddresses())
 
175
 
 
176
        self.assertNotIn(maintainer_address, notified_addresses)
 
177
        self.assertNotIn(changer_address, notified_addresses)
 
178
        self.assertIn(syncer_address, notified_addresses)