1
# Copyright 2012 Canonical Ltd. This software is licensed under the
2
# GNU Affero General Public License version 3 (see the file LICENSE).
4
"""Test notification behaviour for cross-distro package syncs."""
10
from zope.component import getUtility
12
from lp.archiveuploader.nascentupload import (
16
from lp.registry.interfaces.pocket import PackagePublishingPocket
17
from lp.services.log.logger import DevNullLogger
18
from lp.soyuz.enums import (
19
ArchivePermissionType,
22
from lp.soyuz.interfaces.sourcepackageformat import (
23
ISourcePackageFormatSelectionSet,
25
from lp.soyuz.model.archivepermission import ArchivePermission
26
from lp.soyuz.scripts.packagecopier import do_copy
27
from lp.testing import (
31
from lp.testing.fakemethod import FakeMethod
32
from lp.testing.layers import LaunchpadZopelessLayer
33
from lp.testing.mail_helpers import pop_notifications
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
42
setDistroSeriesAndPocket = FakeMethod()
43
validateUploadType = FakeMethod()
44
checkUpload = FakeMethod()
47
class FakeChangesFile:
48
def __init__(self, spph, file_path):
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
57
checkFileName = FakeMethod([])
58
processAddresses = FakeMethod([])
59
processFiles = FakeMethod([])
60
verify = FakeMethod([UploadError("Deliberately broken")])
63
class TestSyncNotification(TestCaseWithFactory):
65
layer = LaunchpadZopelessLayer
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
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)
79
def makeUploader(self, person, archive, component):
80
"""Grant a person upload privileges for archive/component."""
82
person=person, archive=archive, component=component,
83
permission=ArchivePermissionType.UPLOAD)
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,
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:
104
"Maintainer: %s <%s>\n"
105
"Changed-By: %s <%s>\n"
112
return FakeChangesFile(spph, changes_file)
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)
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)
137
def getNotifiedAddresses(self):
138
"""Get email addresses that were notified."""
139
return [message['to'] for message in pop_notifications()]
141
def test_failed_copy_builds_do_not_spam_upstream(self):
142
"""Failed builds do not spam people who are not responsible for them.
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.
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.
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
159
This test guards against bug 876594.
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)
172
self.processAndRejectUpload(nascent_upload)
174
notified_addresses = '\n'.join(self.getNotifiedAddresses())
176
self.assertNotIn(maintainer_address, notified_addresses)
177
self.assertNotIn(changer_address, notified_addresses)
178
self.assertIn(syncer_address, notified_addresses)
added
removed
lib/lp/archiveuploader/tests/test_sync_notification.py
