« back to all changes in this revision
Viewing changes to lib/lp/soyuz/tests/test_packagecloner.py
- browse files at revision 11142
- compare with another revision
- download diff
- download tarball
- view history from revision 11142
- Committer: Launchpad Patch Queue Manager
- Date: 2010-07-16 17:15:56 UTC
- mfrom: (11017.2.12 test-package-cloner)
- Revision ID: launchpad@pqm.canonical.com-20100716171556-lqd8diiac9n493rq
[r=rockstar][ui=none] Make PackageCloner create builds,
and add some tests for it.
and add some tests for it.
added
removed
lib/lp/soyuz/tests/test_packagecloner.py
1
# Copyright 2009 Canonical Ltd. This software is licensed under the
2
# GNU Affero General Public License version 3 (see the file LICENSE).
3
4
__metaclass__ = type
5
6
from zope.component import getUtility
7
from zope.security.proxy import removeSecurityProxy
8
9
from canonical.testing import LaunchpadZopelessLayer
10
11
from lp.buildmaster.interfaces.buildbase import BuildStatus
12
from lp.registry.interfaces.pocket import PackagePublishingPocket
13
from lp.soyuz.adapters.packagelocation import PackageLocation
14
from lp.soyuz.interfaces.archive import ArchivePurpose
15
from lp.soyuz.interfaces.archivearch import IArchiveArchSet
16
from lp.soyuz.interfaces.component import IComponentSet
17
from lp.soyuz.interfaces.binarypackagebuild import IBinaryPackageBuildSet
18
from lp.soyuz.interfaces.packagecloner import IPackageCloner
19
from lp.soyuz.interfaces.publishing import (
20
IPublishingSet, PackagePublishingStatus)
21
from lp.soyuz.model.processor import ProcessorFamilySet
22
from lp.testing import TestCaseWithFactory
23
24
25
class PackageInfo:
26
27
def __init__(self, name, version,
28
status=PackagePublishingStatus.PUBLISHED, component="main"):
29
self.name = name
30
self.version = version
31
self.status = status
32
self.component = component
33
34
35
class PackageClonerTests(TestCaseWithFactory):
36
37
layer = LaunchpadZopelessLayer
38
39
def checkCopiedSources(self, archive, distroseries, expected):
40
"""Check the sources published in an archive against an expected set.
41
42
Given an archive and a target distroseries the sources published in
43
that distroseries are checked against a set of PackageInfo to
44
ensure that the correct package names and versions are published.
45
"""
46
expected_set = set([(info.name, info.version) for info in expected])
47
sources = archive.getPublishedSources(
48
distroseries=distroseries,
49
status=(PackagePublishingStatus.PENDING,
50
PackagePublishingStatus.PUBLISHED))
51
actual_set = set()
52
for source in sources:
53
source = removeSecurityProxy(source)
54
actual_set.add(
55
(source.source_package_name, source.source_package_version))
56
self.assertEqual(expected_set, actual_set)
57
58
def createSourceDistribution(self, package_infos):
59
"""Create a distribution to be the source of a copy archive."""
60
distroseries = self.createSourceDistroSeries()
61
self.createSourcePublications(package_infos, distroseries)
62
return distroseries
63
64
def createSourceDistroSeries(self):
65
"""Create a DistroSeries suitable for copying.
66
67
Creates a distroseries with a DistroArchSeries and nominatedarchindep,
68
which makes it suitable for copying because it will create some builds.
69
"""
70
distro_name = "foobuntu"
71
distro = self.factory.makeDistribution(name=distro_name)
72
distroseries_name = "maudlin"
73
distroseries = self.factory.makeDistroSeries(
74
distribution=distro, name=distroseries_name)
75
das = self.factory.makeDistroArchSeries(
76
distroseries=distroseries, architecturetag="i386",
77
processorfamily=ProcessorFamilySet().getByName("x86"),
78
supports_virtualized=True)
79
distroseries.nominatedarchindep = das
80
return distroseries
81
82
def getTargetArchive(self, distribution):
83
"""Get a target archive for copying in to."""
84
return self.factory.makeArchive(
85
name="test-copy-archive", purpose=ArchivePurpose.COPY,
86
distribution=distribution)
87
88
def createSourcePublication(self, info, distroseries):
89
"""Create a SourcePackagePublishingHistory based on a PackageInfo."""
90
archive = distroseries.distribution.main_archive
91
sources = archive.getPublishedSources(
92
distroseries=distroseries,
93
status=(PackagePublishingStatus.PENDING,
94
PackagePublishingStatus.PUBLISHED),
95
name=info.name, exact_match=True)
96
for src in sources:
97
src.supersede()
98
self.factory.makeSourcePackagePublishingHistory(
99
sourcepackagename=self.factory.getOrMakeSourcePackageName(
100
name=info.name),
101
distroseries=distroseries, component=self.factory.makeComponent(
102
info.component),
103
version=info.version, architecturehintlist='any',
104
archive=archive, status=info.status,
105
pocket=PackagePublishingPocket.RELEASE)
106
107
def createSourcePublications(self, package_infos, distroseries):
108
"""Create a source publication for each item in package_infos."""
109
for package_info in package_infos:
110
self.createSourcePublication(package_info, distroseries)
111
112
def makeCopyArchive(self, package_infos, component="main",
113
source_pocket=None, target_pocket=None,
114
proc_families=None):
115
"""Make a copy archive based on a new distribution."""
116
distroseries = self.createSourceDistribution(package_infos)
117
copy_archive = self.getTargetArchive(distroseries.distribution)
118
to_component = getUtility(IComponentSet).ensure(component)
119
self.copyArchive(
120
copy_archive, distroseries, from_pocket=source_pocket,
121
to_pocket=target_pocket, to_component=to_component,
122
proc_families=proc_families)
123
return (copy_archive, distroseries)
124
125
def checkBuilds(self, archive, package_infos):
126
"""Check the build records pending in an archive.
127
128
Given a set of PackageInfo objects check that each has a build
129
created for it.
130
"""
131
expected_builds = list(
132
[(info.name, info.version) for info in package_infos])
133
builds = list(
134
getUtility(IBinaryPackageBuildSet).getBuildsForArchive(
135
archive, status=BuildStatus.NEEDSBUILD))
136
actual_builds = list()
137
for build in builds:
138
naked_build = removeSecurityProxy(build)
139
spr = naked_build.source_package_release
140
actual_builds.append((spr.name, spr.version))
141
self.assertEqual(sorted(expected_builds), sorted(actual_builds))
142
143
def copyArchive(self, to_archive, to_distroseries, from_archive=None,
144
from_distroseries=None, from_pocket=None, to_pocket=None,
145
to_component=None, packagesets=None, proc_families=None):
146
"""Use a PackageCloner to copy an archive."""
147
if from_distroseries is None:
148
from_distroseries = to_distroseries
149
if from_archive is None:
150
from_archive = from_distroseries.distribution.main_archive
151
if from_pocket is None:
152
from_pocket = PackagePublishingPocket.RELEASE
153
if to_pocket is None:
154
to_pocket = PackagePublishingPocket.RELEASE
155
if packagesets is None:
156
packagesets = []
157
origin = PackageLocation(
158
from_archive, from_distroseries.distribution, from_distroseries,
159
from_pocket)
160
destination = PackageLocation(
161
to_archive, to_distroseries.distribution, to_distroseries,
162
to_pocket)
163
origin.packagesets = packagesets
164
if to_component is not None:
165
destination.component = to_component
166
cloner = getUtility(IPackageCloner)
167
cloner.clonePackages(
168
origin, destination, distroarchseries_list=None,
169
proc_families=proc_families)
170
return cloner
171
172
def testCopiesPublished(self):
173
"""Test that PUBLISHED sources are copied."""
174
package_info = PackageInfo(
175
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED)
176
copy_archive, distroseries = self.makeCopyArchive([package_info])
177
self.checkCopiedSources(
178
copy_archive, distroseries, [package_info])
179
180
def testCopiesPending(self):
181
"""Test that PENDING sources are copied."""
182
package_info = PackageInfo(
183
"bzr", "2.1", status=PackagePublishingStatus.PENDING)
184
copy_archive, distroseries = self.makeCopyArchive([package_info])
185
self.checkCopiedSources(
186
copy_archive, distroseries, [package_info])
187
188
def testDoesntCopySuperseded(self):
189
"""Test that SUPERSEDED sources are not copied."""
190
package_info = PackageInfo(
191
"bzr", "2.1", status=PackagePublishingStatus.SUPERSEDED)
192
copy_archive, distroseries = self.makeCopyArchive([package_info])
193
self.checkCopiedSources(
194
copy_archive, distroseries, [])
195
196
def testDoesntCopyDeleted(self):
197
"""Test that DELETED sources are not copied."""
198
package_info = PackageInfo(
199
"bzr", "2.1", status=PackagePublishingStatus.DELETED)
200
copy_archive, distroseries = self.makeCopyArchive([package_info])
201
self.checkCopiedSources(
202
copy_archive, distroseries, [])
203
204
def testDoesntCopyObsolete(self):
205
"""Test that OBSOLETE sources are not copied."""
206
package_info = PackageInfo(
207
"bzr", "2.1", status=PackagePublishingStatus.OBSOLETE)
208
copy_archive, distroseries = self.makeCopyArchive([package_info])
209
self.checkCopiedSources(
210
copy_archive, distroseries, [])
211
212
def testCopiesAllComponents(self):
213
"""Test that packages from all components are copied.
214
215
When copying you specify a component, but that component doesn't
216
limit the packages copied. We create a source in main and one in
217
universe, and then copy with --component main, and expect to see
218
both sources in the copy.
219
"""
220
package_infos = [
221
PackageInfo(
222
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED,
223
component="universe"),
224
PackageInfo(
225
"apt", "2.2", status=PackagePublishingStatus.PUBLISHED,
226
component="main")]
227
copy_archive, distroseries = self.makeCopyArchive(package_infos,
228
component="main")
229
self.checkCopiedSources(copy_archive, distroseries, package_infos)
230
231
def testSubsetsBasedOnPackageset(self):
232
"""Test that --package-set limits the sources copied."""
233
package_infos = [
234
PackageInfo(
235
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED),
236
PackageInfo(
237
"apt", "2.2", status=PackagePublishingStatus.PUBLISHED),
238
]
239
distroseries = self.createSourceDistribution(package_infos)
240
spn = self.factory.getOrMakeSourcePackageName(name="apt")
241
packageset = self.factory.makePackageset(
242
distroseries=distroseries, packages=(spn,))
243
copy_archive = self.getTargetArchive(distroseries.distribution)
244
self.copyArchive(copy_archive, distroseries, packagesets=[packageset])
245
self.checkCopiedSources(
246
copy_archive, distroseries, [package_infos[1]])
247
248
def testUnionsPackagesets(self):
249
"""Test that package sets are unioned when copying archives."""
250
package_infos = [
251
PackageInfo(
252
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED),
253
PackageInfo(
254
"apt", "2.2", status=PackagePublishingStatus.PUBLISHED),
255
PackageInfo(
256
"gcc", "4.5", status=PackagePublishingStatus.PUBLISHED),
257
]
258
distroseries = self.createSourceDistribution(package_infos)
259
apt_spn = self.factory.getOrMakeSourcePackageName(name="apt")
260
gcc_spn = self.factory.getOrMakeSourcePackageName(name="gcc")
261
apt_packageset = self.factory.makePackageset(
262
distroseries=distroseries, packages=(apt_spn,))
263
gcc_packageset = self.factory.makePackageset(
264
distroseries=distroseries, packages=(gcc_spn,))
265
copy_archive = self.getTargetArchive(distroseries.distribution)
266
self.copyArchive(
267
copy_archive, distroseries,
268
packagesets=[apt_packageset, gcc_packageset])
269
self.checkCopiedSources(
270
copy_archive, distroseries, package_infos[1:])
271
272
def testRecursivelyCopiesPackagesets(self):
273
"""Test that package set copies include subsets."""
274
package_infos = [
275
PackageInfo(
276
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED),
277
PackageInfo(
278
"apt", "2.2", status=PackagePublishingStatus.PUBLISHED),
279
PackageInfo(
280
"gcc", "4.5", status=PackagePublishingStatus.PUBLISHED),
281
]
282
distroseries = self.createSourceDistribution(package_infos)
283
apt_spn = self.factory.getOrMakeSourcePackageName(name="apt")
284
gcc_spn = self.factory.getOrMakeSourcePackageName(name="gcc")
285
apt_packageset = self.factory.makePackageset(
286
distroseries=distroseries, packages=(apt_spn,))
287
gcc_packageset = self.factory.makePackageset(
288
distroseries=distroseries, packages=(gcc_spn,))
289
apt_packageset.add((gcc_packageset,))
290
copy_archive = self.getTargetArchive(distroseries.distribution)
291
self.copyArchive(
292
copy_archive, distroseries, packagesets=[apt_packageset])
293
self.checkCopiedSources(
294
copy_archive, distroseries, package_infos[1:])
295
296
def testCloneFromPPA(self):
297
"""Test we can create a copy archive with a PPA as the source."""
298
distroseries = self.createSourceDistroSeries()
299
ppa = self.factory.makeArchive(
300
purpose=ArchivePurpose.PPA,
301
distribution=distroseries.distribution)
302
package_info = PackageInfo(
303
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED,
304
component="universe")
305
self.factory.makeSourcePackagePublishingHistory(
306
sourcepackagename=self.factory.getOrMakeSourcePackageName(
307
name=package_info.name),
308
distroseries=distroseries, component=self.factory.makeComponent(
309
package_info.component),
310
version=package_info.version, archive=ppa,
311
status=package_info.status, architecturehintlist='any',
312
pocket=PackagePublishingPocket.RELEASE)
313
copy_archive = self.getTargetArchive(distroseries.distribution)
314
self.copyArchive(copy_archive, distroseries, from_archive=ppa)
315
self.checkCopiedSources(
316
copy_archive, distroseries, [package_info])
317
318
def testCreatesNoBuildsWithNoProcFamilies(self):
319
"""Test that no builds are created if we specify no proc families."""
320
package_info = PackageInfo(
321
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED)
322
copy_archive, distroseries = self.makeCopyArchive([package_info])
323
self.checkBuilds(copy_archive, [])
324
325
def testCreatesBuilds(self):
326
"""Test that a copy archive creates builds for the copied packages."""
327
package_info = PackageInfo(
328
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED)
329
# This is the processor family for the DAS that the source has,
330
# so we expect to get builds.
331
proc_families = [ProcessorFamilySet().getByName("x86")]
332
copy_archive, distroseries = self.makeCopyArchive(
333
[package_info], proc_families=proc_families)
334
self.checkBuilds(copy_archive, [package_info])
335
336
def testNoBuildsIfProcFamilyNotInSource(self):
337
"""Test that no builds are created for a proc family without a DAS."""
338
package_info = PackageInfo(
339
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED)
340
# This is a processor family without a DAS in the source, so
341
# we expect no builds.
342
family = self.factory.makeProcessorFamily(name="armel")
343
self.factory.makeProcessor(family=family, name="armel")
344
proc_families = [family]
345
copy_archive, distroseries = self.makeCopyArchive(
346
[package_info], proc_families=proc_families)
347
self.checkBuilds(copy_archive, [])
348
349
def testBuildsOnlyForProcFamiliesInSource(self):
350
"""Test that builds are only created for proc families in source."""
351
package_info = PackageInfo(
352
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED)
353
# One of these processor families has a DAS in the source, so
354
# we expect one set of builds
355
family = self.factory.makeProcessorFamily(name="armel")
356
self.factory.makeProcessor(family=family, name="armel")
357
proc_families = [family, ProcessorFamilySet().getByName("x86")]
358
copy_archive, distroseries = self.makeCopyArchive(
359
[package_info], proc_families=proc_families)
360
self.checkBuilds(copy_archive, [package_info])
361
362
def testCreatesSubsetOfBuilds(self):
363
"""Test that builds are only created for requested families."""
364
package_info = PackageInfo(
365
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED)
366
distroseries = self.createSourceDistribution([package_info])
367
# Create a DAS for a second family
368
self.factory.makeDistroArchSeries(
369
distroseries=distroseries, architecturetag="amd64",
370
processorfamily=ProcessorFamilySet().getByName("amd64"),
371
supports_virtualized=True)
372
# The request builds for only one of the families, so we
373
# expect just one build for each source
374
proc_families = [ProcessorFamilySet().getByName("x86")]
375
copy_archive = self.getTargetArchive(distroseries.distribution)
376
self.copyArchive(
377
copy_archive, distroseries, proc_families=proc_families)
378
self.checkBuilds(copy_archive, [package_info])
379
380
def testCreatesMultipleBuilds(self):
381
"""Test that multiple families result in mutiple builds."""
382
package_info = PackageInfo(
383
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED)
384
distroseries = self.createSourceDistribution([package_info])
385
# Create a DAS for a second family
386
amd64_family = ProcessorFamilySet().getByName("amd64")
387
self.factory.makeDistroArchSeries(
388
distroseries=distroseries, architecturetag="amd64",
389
processorfamily=amd64_family, supports_virtualized=True)
390
# The request builds for both families, so we expect two builds
391
# per source.
392
proc_families = [ProcessorFamilySet().getByName("x86"), amd64_family]
393
copy_archive = self.getTargetArchive(distroseries.distribution)
394
self.copyArchive(
395
copy_archive, distroseries, proc_families=proc_families)
396
self.checkBuilds(copy_archive, [package_info, package_info])
397
398
399
def diffArchives(self, target_archive, target_distroseries,
400
source_archive=None, source_distroseries=None):
401
"""Run a packageSetDiff of two archives."""
402
if source_distroseries is None:
403
source_distroseries = target_distroseries
404
if source_archive is None:
405
source_archive = source_distroseries.distribution.main_archive
406
source_location = PackageLocation(
407
source_archive, source_distroseries.distribution,
408
source_distroseries, PackagePublishingPocket.RELEASE)
409
target_location = PackageLocation(
410
target_archive, target_distroseries.distribution,
411
target_distroseries, PackagePublishingPocket.RELEASE)
412
cloner = getUtility(IPackageCloner)
413
return cloner.packageSetDiff(source_location, target_location)
414
415
def checkPackageDiff(self, expected_changed, expected_new, actual,
416
archive):
417
"""Check that the diff of two archives is as expected."""
418
actual_changed_keys, actual_new_keys = actual
419
expected_changed_tuples = [(e.name, e.version)
420
for e in expected_changed]
421
expected_new_tuples = [(e.name, e.version) for e in expected_new]
422
def get_tuples(source_keys):
423
tuples = []
424
for source_key in source_keys:
425
source = getUtility(IPublishingSet).getByIdAndArchive(
426
source_key, archive, source=True)
427
self.assertNotEqual(source, None, "Got a non-existant "
428
"source publishing record: %d" % source_key)
429
naked_source = removeSecurityProxy(source)
430
tuples.append(
431
(naked_source.source_package_name,
432
naked_source.source_package_version))
433
return tuples
434
actual_changed_tuples = get_tuples(actual_changed_keys)
435
actual_new_tuples = get_tuples(actual_new_keys)
436
self.assertEqual(expected_changed_tuples, actual_changed_tuples)
437
self.assertEqual(expected_new_tuples, actual_new_tuples)
438
439
def testPackageSetDiffWithNothingNew(self):
440
"""Test packageSetDiff."""
441
package_info = PackageInfo(
442
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED)
443
copy_archive, distroseries = self.makeCopyArchive([package_info])
444
diff = self.diffArchives(copy_archive, distroseries)
445
self.checkPackageDiff(
446
[], [], diff, distroseries.distribution.main_archive)
447
448
def testPackageSetDiffWithNewPackages(self):
449
package_info = PackageInfo(
450
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED)
451
copy_archive, distroseries = self.makeCopyArchive([package_info])
452
package_infos = [
453
PackageInfo(
454
"apt", "1.2", status=PackagePublishingStatus.PUBLISHED),
455
PackageInfo(
456
"gcc", "4.5", status=PackagePublishingStatus.PENDING),
457
]
458
self.createSourcePublications(package_infos, distroseries)
459
diff = self.diffArchives(copy_archive, distroseries)
460
self.checkPackageDiff(
461
[], package_infos, diff, distroseries.distribution.main_archive)
462
463
def testPackageSetDiffWithChangedPackages(self):
464
package_infos = [
465
PackageInfo(
466
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED),
467
PackageInfo(
468
"apt", "1.2", status=PackagePublishingStatus.PUBLISHED),
469
]
470
copy_archive, distroseries = self.makeCopyArchive(package_infos)
471
package_infos = [
472
PackageInfo(
473
"bzr", "2.2", status=PackagePublishingStatus.PUBLISHED),
474
PackageInfo(
475
"apt", "1.3", status=PackagePublishingStatus.PENDING),
476
]
477
self.createSourcePublications(package_infos, distroseries)
478
diff = self.diffArchives(copy_archive, distroseries)
479
self.checkPackageDiff(
480
package_infos, [], diff, distroseries.distribution.main_archive)
481
482
def testPackageSetDiffWithBoth(self):
483
package_infos = [
484
PackageInfo(
485
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED),
486
PackageInfo(
487
"apt", "1.2", status=PackagePublishingStatus.PUBLISHED),
488
]
489
copy_archive, distroseries = self.makeCopyArchive(package_infos)
490
package_infos = [
491
PackageInfo(
492
"bzr", "2.2", status=PackagePublishingStatus.PUBLISHED),
493
PackageInfo(
494
"gcc", "1.3", status=PackagePublishingStatus.PENDING),
495
]
496
self.createSourcePublications(package_infos, distroseries)
497
diff = self.diffArchives(copy_archive, distroseries)
498
self.checkPackageDiff(
499
[package_infos[0]], [package_infos[1]], diff,
500
distroseries.distribution.main_archive)
501
502
503
def mergeCopy(self, target_archive, target_distroseries,
504
source_archive=None, source_distroseries=None):
505
if source_distroseries is None:
506
source_distroseries = target_distroseries
507
if source_archive is None:
508
source_archive = source_distroseries.distribution.main_archive
509
source_location = PackageLocation(
510
source_archive, source_distroseries.distribution,
511
source_distroseries, PackagePublishingPocket.RELEASE)
512
target_location = PackageLocation(
513
target_archive, target_distroseries.distribution,
514
target_distroseries, PackagePublishingPocket.RELEASE)
515
cloner = getUtility(IPackageCloner)
516
return cloner.mergeCopy(source_location, target_location)
517
518
def testMergeCopyNoChanges(self):
519
package_info = PackageInfo(
520
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED)
521
copy_archive, distroseries = self.makeCopyArchive([package_info])
522
self.mergeCopy(copy_archive, distroseries)
523
self.checkCopiedSources(
524
copy_archive, distroseries, [package_info])
525
526
def testMergeCopyWithNewPackages(self):
527
package_info = PackageInfo(
528
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED)
529
copy_archive, distroseries = self.makeCopyArchive([package_info])
530
package_infos = [
531
PackageInfo(
532
"apt", "1.2", status=PackagePublishingStatus.PUBLISHED),
533
PackageInfo(
534
"gcc", "4.5", status=PackagePublishingStatus.PENDING),
535
]
536
self.createSourcePublications(package_infos, distroseries)
537
self.mergeCopy(copy_archive, distroseries)
538
self.checkCopiedSources(
539
copy_archive, distroseries, [package_info] + package_infos)
540
541
def testMergeCopyWithChangedPackages(self):
542
package_infos = [
543
PackageInfo(
544
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED),
545
PackageInfo(
546
"apt", "1.2", status=PackagePublishingStatus.PUBLISHED),
547
]
548
copy_archive, distroseries = self.makeCopyArchive(package_infos)
549
package_infos = [
550
PackageInfo(
551
"bzr", "2.2", status=PackagePublishingStatus.PUBLISHED),
552
PackageInfo(
553
"apt", "1.3", status=PackagePublishingStatus.PENDING),
554
]
555
self.createSourcePublications(package_infos, distroseries)
556
self.mergeCopy(copy_archive, distroseries)
557
# Critically there is only one record for each info, as the
558
# others have been obsoleted.
559
self.checkCopiedSources(
560
copy_archive, distroseries, package_infos)
561
562
def testMergeCopyWithBoth(self):
563
package_infos = [
564
PackageInfo(
565
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED),
566
PackageInfo(
567
"apt", "1.2", status=PackagePublishingStatus.PUBLISHED),
568
]
569
copy_archive, distroseries = self.makeCopyArchive(package_infos)
570
package_infos2 = [
571
PackageInfo(
572
"bzr", "2.2", status=PackagePublishingStatus.PUBLISHED),
573
PackageInfo(
574
"gcc", "1.3", status=PackagePublishingStatus.PENDING),
575
]
576
self.createSourcePublications(package_infos2, distroseries)
577
self.mergeCopy(copy_archive, distroseries)
578
# Again bzr is obsoleted, gcc is added and apt remains.
579
self.checkCopiedSources(
580
copy_archive, distroseries, [package_infos[1]] + package_infos2)
581
582
def setArchiveArchitectures(self, archive, proc_families):
583
"""Associate the archive with the processor families."""
584
aa_set = getUtility(IArchiveArchSet)
585
for proc_family in proc_families:
586
aa_set.new(archive, proc_family)
587
588
def testMergeCopyCreatesBuilds(self):
589
package_infos = [
590
PackageInfo(
591
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED),
592
PackageInfo(
593
"apt", "1.2", status=PackagePublishingStatus.PUBLISHED),
594
]
595
proc_families = [ProcessorFamilySet().getByName("x86")]
596
copy_archive, distroseries = self.makeCopyArchive(
597
package_infos, proc_families=proc_families)
598
self.setArchiveArchitectures(copy_archive, proc_families)
599
package_infos2 = [
600
PackageInfo(
601
"bzr", "2.2", status=PackagePublishingStatus.PUBLISHED),
602
PackageInfo(
603
"gcc", "1.3", status=PackagePublishingStatus.PENDING),
604
]
605
self.createSourcePublications(package_infos2, distroseries)
606
self.mergeCopy(copy_archive, distroseries)
607
# We get all builds, as superseding bzr doesn't cancel the
608
# build
609
self.checkBuilds(copy_archive, package_infos + package_infos2)
610
611
def testMergeCopyCreatesNoBuildsWhenNoArchitectures(self):
612
package_infos = [
613
PackageInfo(
614
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED),
615
]
616
# We specify no processor families at creation time
617
copy_archive, distroseries = self.makeCopyArchive(
618
package_infos, proc_families=[])
619
package_infos2 = [
620
PackageInfo(
621
"bzr", "2.2", status=PackagePublishingStatus.PUBLISHED),
622
]
623
self.createSourcePublications(package_infos2, distroseries)
624
self.mergeCopy(copy_archive, distroseries)
625
# And so we get no builds at merge time
626
self.checkBuilds(copy_archive, [])
627
628
def testMergeCopyCreatesBuildsForMultipleArchitectures(self):
629
package_infos = [
630
PackageInfo(
631
"bzr", "2.1", status=PackagePublishingStatus.PUBLISHED),
632
PackageInfo(
633
"apt", "1.2", status=PackagePublishingStatus.PUBLISHED),
634
]
635
distroseries = self.createSourceDistribution(package_infos)
636
# Create a DAS for a second family
637
amd64_family = ProcessorFamilySet().getByName("amd64")
638
self.factory.makeDistroArchSeries(
639
distroseries=distroseries, architecturetag="amd64",
640
processorfamily=amd64_family, supports_virtualized=True)
641
# The request builds for both families, so we expect two builds
642
# per source.
643
proc_families = [ProcessorFamilySet().getByName("x86"), amd64_family]
644
copy_archive = self.getTargetArchive(distroseries.distribution)
645
self.setArchiveArchitectures(copy_archive, proc_families)
646
self.copyArchive(
647
copy_archive, distroseries, proc_families=proc_families)
648
package_infos2 = [
649
PackageInfo(
650
"bzr", "2.2", status=PackagePublishingStatus.PUBLISHED),
651
PackageInfo(
652
"gcc", "1.3", status=PackagePublishingStatus.PENDING),
653
]
654
self.createSourcePublications(package_infos2, distroseries)
655
self.mergeCopy(copy_archive, distroseries)
656
# We get all builds twice, one for each architecture.
657
self.checkBuilds(
658
copy_archive,
659
package_infos + package_infos + package_infos2 + package_infos2)
Loggerhead is a web-based interface for Breezy
Version: 2.0.1