~launchpad-pqm/launchpad/devel
» Revision
12989
: lib/lp/archivepublisher/tests/test_createdistroseriesindexesjob.py
« back to all changes in this revision
Viewing changes to lib/lp/archivepublisher/tests/test_createdistroseriesindexesjob.py
- browse files at revision 12989
- compare with another revision
- compare with revision 13600
- download diff
- download tarball
- view history from revision 12989
- Committer: Launchpad Patch Queue Manager
- Date: 2011-05-06 06:57:37 UTC
- mfrom: (12981.2.7 bug-777941)
- Revision ID: launchpad@pqm.canonical.com-20110506065737-1vb8pmgtuia2ubqc
[r=jcsackett][bug=55211,
777941] Create new distroseries indexes from publish-ftpmaster script.
777941] Create new distroseries indexes from publish-ftpmaster script.
- files modified:
- lib/canonical/config/schema-lazr.conf
added
removed
lib/lp/archivepublisher/tests/test_createdistroseriesindexesjob.py
1
# Copyright 2011 Canonical Ltd. This software is licensed under the
2
# GNU Affero General Public License version 3 (see the file LICENSE).
3
4
"""Tests for `CreateDistroSeriesIndexesJob`."""
5
6
__metaclass__ = type
7
8
from logging import (
9
FATAL,
10
getLogger,
11
)
12
import os.path
13
from zope.component import getUtility
14
from zope.security.proxy import removeSecurityProxy
15
16
from canonical.config import config
17
from canonical.launchpad.interfaces.lpstorm import IMasterStore
18
from canonical.launchpad.webapp.testing import verifyObject
19
from canonical.testing.layers import LaunchpadZopelessLayer
20
from lp.archivepublisher.config import getPubConfig
21
from lp.archivepublisher.interfaces.createdistroseriesindexesjob import (
22
ICreateDistroSeriesIndexesJobSource,
23
)
24
from lp.archivepublisher.interfaces.publisherconfig import IPublisherConfigSet
25
from lp.archivepublisher.model.createdistroseriesindexesjob import (
26
FEATURE_FLAG_ENABLE_MODULE,
27
CreateDistroSeriesIndexesJob,
28
)
29
from lp.registry.interfaces.pocket import pocketsuffix
30
from lp.services.job.model.job import Job
31
from lp.services.features.testing import FeatureFixture
32
from lp.services.job.interfaces.job import (
33
IRunnableJob,
34
JobStatus,
35
)
36
from lp.services.job.runner import JobCronScript
37
from lp.services.log.logger import DevNullLogger
38
from lp.services.mail import stub
39
from lp.services.mail.sendmail import format_address_for_person
40
from lp.services.utils import file_exists
41
from lp.soyuz.enums import ArchivePurpose
42
from lp.soyuz.interfaces.distributionjob import IDistributionJob
43
from lp.soyuz.model.distributionjob import DistributionJob
44
from lp.testing import TestCaseWithFactory
45
from lp.testing.fakemethod import FakeMethod
46
from lp.testing.mail_helpers import run_mail_jobs
47
48
49
def silence_publisher_logger():
50
"""Silence the logger that `run_publisher` creates."""
51
getLogger("publish-distro").setLevel(FATAL)
52
53
54
class TestCreateDistroSeriesIndexesJobSource(TestCaseWithFactory):
55
"""Test utility."""
56
57
layer = LaunchpadZopelessLayer
58
59
def setUp(self):
60
super(TestCreateDistroSeriesIndexesJobSource, self).setUp()
61
self.useFixture(FeatureFixture({FEATURE_FLAG_ENABLE_MODULE: u'on'}))
62
63
def removePublisherConfig(self, distribution):
64
"""Strip `distribution` of its publisher configuration."""
65
publisher_config = getUtility(IPublisherConfigSet).getByDistribution(
66
distribution)
67
IMasterStore(publisher_config).remove(publisher_config)
68
69
def test_baseline(self):
70
# The utility conforms to the interfaces it claims to implement.
71
jobsource = getUtility(ICreateDistroSeriesIndexesJobSource)
72
self.assertTrue(
73
verifyObject(ICreateDistroSeriesIndexesJobSource, jobsource))
74
75
def test_creates_job_for_distro_with_publisher_config(self):
76
# The utility can create a job if the distribution has a
77
# publisher configuration.
78
distroseries = self.factory.makeDistroSeries()
79
jobset = getUtility(ICreateDistroSeriesIndexesJobSource)
80
job = jobset.makeFor(distroseries)
81
self.assertIsInstance(job, CreateDistroSeriesIndexesJob)
82
83
def test_does_not_create_job_for_distro_without_publisher_config(self):
84
# If the distribution has no publisher configuration, the
85
# utility creates no job for it.
86
distroseries = self.factory.makeDistroSeries()
87
self.removePublisherConfig(distroseries.distribution)
88
jobset = getUtility(ICreateDistroSeriesIndexesJobSource)
89
job = jobset.makeFor(distroseries)
90
self.assertIs(None, job)
91
92
def test_feature_flag_disables_feature(self):
93
# The creation of jobs is controlled by a feature flag.
94
self.useFixture(FeatureFixture({FEATURE_FLAG_ENABLE_MODULE: u''}))
95
jobset = getUtility(ICreateDistroSeriesIndexesJobSource)
96
self.assertIs(None, jobset.makeFor(self.factory.makeDistroSeries()))
97
98
99
class HorribleFailure(Exception):
100
"""A sample error for testing purposes."""
101
102
103
class TestCreateDistroSeriesIndexesJob(TestCaseWithFactory):
104
"""Test job class."""
105
106
layer = LaunchpadZopelessLayer
107
108
def setUp(self):
109
super(TestCreateDistroSeriesIndexesJob, self).setUp()
110
self.useFixture(FeatureFixture({FEATURE_FLAG_ENABLE_MODULE: u'on'}))
111
112
def getJobSource(self):
113
"""Shorthand for getting at the job-source utility."""
114
return getUtility(ICreateDistroSeriesIndexesJobSource)
115
116
def makeJob(self, distroseries=None):
117
"""Create an `CreateDistroSeriesIndexesJob`."""
118
if distroseries is None:
119
distroseries = self.factory.makeDistroSeries()
120
return self.getJobSource().makeFor(distroseries)
121
122
def getDistsRoot(self, distribution):
123
"""Get distsroot directory for `distribution`."""
124
archive = removeSecurityProxy(distribution.main_archive)
125
pub_config = getPubConfig(archive)
126
return pub_config.distsroot
127
128
def makeDistsDirs(self, distroseries):
129
"""Create dists directories in `distsroot` for `distroseries`."""
130
distsroot = self.getDistsRoot(distroseries.distribution)
131
base = os.path.join(distsroot, distroseries.name)
132
for suffix in pocketsuffix.itervalues():
133
os.makedirs(base + suffix)
134
135
def makeCredibleJob(self):
136
"""Create a job with fixtures required for running it."""
137
silence_publisher_logger()
138
distro = self.factory.makeDistribution(
139
publish_root_dir=unicode(self.makeTemporaryDirectory()))
140
distroseries = self.factory.makeDistroSeries(distribution=distro)
141
self.makeDistsDirs(distroseries)
142
return self.makeJob(distroseries)
143
144
def becomeArchivePublisher(self):
145
"""Become the archive publisher database user (and clean up later)."""
146
self.becomeDbUser(config.archivepublisher.dbuser)
147
self.addCleanup(self.becomeDbUser, 'launchpad')
148
149
def getSuites(self, distroseries):
150
"""Get the list of suites for `distroseries`."""
151
return [
152
distroseries.name + suffix
153
for suffix in pocketsuffix.itervalues()]
154
155
def test_baseline(self):
156
# The job class conforms to the interfaces it claims to implement.
157
job = self.makeJob()
158
self.assertTrue(verifyObject(IRunnableJob, job))
159
self.assertTrue(verifyObject(IDistributionJob, job))
160
161
def test_getSuites_identifies_distroseries_suites(self):
162
# getSuites lists all suites in the distroseries.
163
job = self.makeJob()
164
self.assertContentEqual(
165
self.getSuites(job.distroseries),
166
removeSecurityProxy(job).getSuites())
167
168
def test_getSuites_ignores_suites_for_other_distroseries(self):
169
# getSuites does not list suites in the distribution that do not
170
# belong to the right distroseries.
171
job = self.makeJob()
172
self.assertContentEqual(
173
self.getSuites(job.distroseries),
174
removeSecurityProxy(job).getSuites())
175
176
def test_job_runs_publish_distro_for_main(self):
177
# The job always runs publish_distro for the distribution's main
178
# archive.
179
job = self.makeJob()
180
naked_job = removeSecurityProxy(job)
181
naked_job.runPublishDistro = FakeMethod()
182
job.run()
183
args, kwargs = naked_job.runPublishDistro.calls[-1]
184
self.assertEqual((), args)
185
186
def test_job_runs_publish_distro_for_partner_if_present(self):
187
# If the distribution has a partner archive, the job will run
188
# publish_distro for it. This differs from the run for the main
189
# archive in that publish_distro receives the --partner option.
190
distroseries = self.factory.makeDistroSeries()
191
self.factory.makeArchive(
192
distribution=distroseries.distribution,
193
purpose=ArchivePurpose.PARTNER)
194
job = self.makeJob(distroseries)
195
naked_job = removeSecurityProxy(job)
196
naked_job.runPublishDistro = FakeMethod()
197
job.run()
198
self.assertIn(
199
('--partner', ),
200
[args for args, kwargs in naked_job.runPublishDistro.calls])
201
202
def test_job_does_not_run_publish_distro_for_partner_if_not_present(self):
203
# If the distribution does not have a partner archive,
204
# publish_distro is not run for the partner archive.
205
job = self.makeJob()
206
naked_job = removeSecurityProxy(job)
207
naked_job.runPublishDistro = FakeMethod()
208
job.run()
209
self.assertEqual(1, naked_job.runPublishDistro.call_count)
210
211
def test_job_notifies_if_successful(self):
212
# Once the indexes have been created, the job calls its
213
# notifySuccess method to let stakeholders know that they may
214
# proceed with their release process.
215
job = self.makeJob()
216
naked_job = removeSecurityProxy(job)
217
naked_job.runPublishDistro = FakeMethod()
218
naked_job.notifySuccess = FakeMethod()
219
job.run()
220
self.assertEqual(1, naked_job.notifySuccess.call_count)
221
222
def test_failure_notifies_recipients(self):
223
# Failure notices are sent to the addresses returned by
224
# getMailRecipients.
225
job = self.makeJob()
226
removeSecurityProxy(job).getMailRecipients = FakeMethod(
227
result=["foo@example.com"])
228
job.notifyUserError(HorribleFailure("Boom!"))
229
run_mail_jobs()
230
sender, recipients, body = stub.test_emails.pop()
231
self.assertIn("foo@example.com", recipients)
232
233
def test_success_notifies_recipients(self):
234
# Success notices are sent to the addresses returned by
235
# getMailRecipients.
236
job = self.makeJob()
237
naked_job = removeSecurityProxy(job)
238
naked_job.getMailRecipients = FakeMethod(result=["bar@example.com"])
239
naked_job.notifySuccess()
240
run_mail_jobs()
241
sender, recipients, body = stub.test_emails.pop()
242
self.assertIn("bar@example.com", recipients)
243
244
def test_notifySuccess_sends_email(self):
245
# notifySuccess sends out a success notice by email.
246
job = self.makeJob()
247
removeSecurityProxy(job).notifySuccess()
248
run_mail_jobs()
249
sender, recipients, body = stub.test_emails.pop()
250
self.assertIn("success", body)
251
252
def test_release_manager_gets_notified(self):
253
# The release manager gets notified. This role is represented
254
# by the driver for the distroseries.
255
distroseries = self.factory.makeDistroSeries()
256
distroseries.driver = self.factory.makePerson()
257
job = self.makeJob(distroseries)
258
self.assertIn(
259
format_address_for_person(distroseries.driver),
260
removeSecurityProxy(job).getMailRecipients())
261
262
def test_distribution_owner_gets_notified_if_no_release_manager(self):
263
# If no release manager is available, the distribution owners
264
# are notified.
265
distroseries = self.factory.makeDistroSeries()
266
distroseries.driver = None
267
job = self.makeJob(distroseries)
268
self.assertIn(
269
format_address_for_person(distroseries.distribution.owner),
270
removeSecurityProxy(job).getMailRecipients())
271
272
def test_destroySelf_destroys_job(self):
273
job = self.makeJob()
274
job_id = removeSecurityProxy(job).job.id
275
self.becomeArchivePublisher()
276
job.destroySelf()
277
store = IMasterStore(Job)
278
self.assertIs(None, store.find(Job, Job.id == job_id).one())
279
280
def test_destroySelf_destroys_DistributionJob(self):
281
job = self.makeJob()
282
job_id = job.id
283
self.becomeArchivePublisher()
284
job.destroySelf()
285
store = IMasterStore(DistributionJob)
286
self.assertIs(
287
None,
288
store.find(DistributionJob, DistributionJob.id == job_id).one())
289
290
def test_run_does_the_job(self):
291
# The job runs publish_distro and generates the expected output
292
# files.
293
job = self.makeCredibleJob()
294
self.becomeArchivePublisher()
295
job.run()
296
distsroot = self.getDistsRoot(job.distribution)
297
output = os.path.join(distsroot, job.distroseries.name, "Release")
298
self.assertTrue(file_exists(output))
299
300
def test_job_runner_runs_jobs(self):
301
# The generic job runner can set itself up to run these jobs.
302
job = self.makeCredibleJob()
303
script = JobCronScript(
304
test_args=["create_distroseries_indexes"],
305
commandline_config=True)
306
script.logger = DevNullLogger()
307
script.main()
308
self.assertEqual(
309
JobStatus.COMPLETED, removeSecurityProxy(job).context.job.status)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1