1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
|
# Copyright 2011 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Tests of DistributionSourcePackageRelease."""
from testtools.matchers import LessThan
from zope.component import getUtility
from canonical.testing.layers import DatabaseFunctionalLayer
from lp.soyuz.enums import PackagePublishingStatus
from lp.soyuz.interfaces.distroarchseries import IDistroArchSeriesSet
from lp.soyuz.model.distributionsourcepackagerelease import (
DistributionSourcePackageRelease,
)
from lp.soyuz.tests.test_publishing import SoyuzTestPublisher
from lp.testing import (
StormStatementRecorder,
TestCaseWithFactory,
)
from lp.testing.matchers import HasQueryCount
class TestDistributionSourcePackageRelease(TestCaseWithFactory):
"""Tests for DistributionSourcePackageRelease."""
layer = DatabaseFunctionalLayer
def setUp(self):
super(TestDistributionSourcePackageRelease, self).setUp()
self.sourcepackagerelease = self.factory.makeSourcePackageRelease()
self.distroarchseries = self.factory.makeDistroArchSeries(
distroseries=self.sourcepackagerelease.sourcepackage.distroseries)
distribution = self.distroarchseries.distroseries.distribution
self.dsp_release = DistributionSourcePackageRelease(
distribution, self.sourcepackagerelease)
def makeBinaryPackageRelease(self, name=None):
if name is None:
name = self.factory.makeBinaryPackageName()
bp_build = self.factory.makeBinaryPackageBuild(
source_package_release=self.sourcepackagerelease,
distroarchseries=self.distroarchseries)
bp_release = self.factory.makeBinaryPackageRelease(
build=bp_build, binarypackagename=name)
sourcepackagename = self.sourcepackagerelease.sourcepackagename
self.factory.makeSourcePackagePublishingHistory(
sourcepackagename=sourcepackagename,
sourcepackagerelease=self.sourcepackagerelease,
distroseries=self.sourcepackagerelease.sourcepackage.distroseries,
status=PackagePublishingStatus.PUBLISHED)
self.factory.makeBinaryPackagePublishingHistory(
binarypackagerelease=bp_release,
distroarchseries=self.distroarchseries)
def test_sample_binary_packages__no_releases(self):
# If no binary releases exist,
# DistributionSourcePackageRelease.sample_binary_packages is empty.
self.assertEqual(0, self.dsp_release.sample_binary_packages.count())
def test_sample_binary_packages__one_release(self):
# If a binary release exists,
# DistributionSourcePackageRelease.sample_binary_packages
# returns it.
self.makeBinaryPackageRelease(
self.factory.makeBinaryPackageName(name='binary-package'))
self.assertEqual(
['binary-package'],
[release.name
for release in self.dsp_release.sample_binary_packages])
def test_sample_binary_packages__two_releases_one_binary_package(self):
# If two binary releases with the same name exist,
# DistributionSourcePackageRelease.sample_binary_packages
# returns only one.
name = self.factory.makeBinaryPackageName(name='binary-package')
self.makeBinaryPackageRelease(name)
self.makeBinaryPackageRelease(name)
self.assertEqual(
['binary-package'],
[release.name
for release in self.dsp_release.sample_binary_packages])
def test_sample_binary_packages__two_release_two_binary_packages(self):
# If a two binary releases with different names exist,
# DistributionSourcePackageRelease.sample_binary_packages
# returns both.
self.makeBinaryPackageRelease(
self.factory.makeBinaryPackageName(name='binary-package'))
self.makeBinaryPackageRelease(
self.factory.makeBinaryPackageName(name='binary-package-2'))
self.assertEqual(
['binary-package', 'binary-package-2'],
[release.name
for release in self.dsp_release.sample_binary_packages])
def updateDistroSeriesPackageCache(self):
# Create DistroSeriesPackageCache records for new binary
# packages.
#
# SoyuzTestPublisher.updateDistroSeriesPackageCache() creates
# a DistroSeriesPackageCache record for the new binary package.
# The method closes the current DB connection, making references
# to DB objects in other DB objects unusable. Starting with
# the distroarchseries, we can create new, valid, instances of
# objects required later in the test again.
# of the objects we need later.
sourcepackagename = self.sourcepackagerelease.sourcepackagename
publisher = SoyuzTestPublisher()
publisher.updateDistroSeriesPackageCache(
self.distroarchseries.distroseries)
self.distroarchseries = getUtility(IDistroArchSeriesSet).get(
self.distroarchseries.id)
distribution = self.distroarchseries.distroseries.distribution
releases = distribution.getCurrentSourceReleases([sourcepackagename])
[(distribution_sourcepackage, dsp_release)] = releases.items()
self.dsp_release = dsp_release
self.sourcepackagerelease = dsp_release.sourcepackagerelease
def test_sample_binary_packages__constant_number_sql_queries(self):
# Retrieving
# DistributionSourcePackageRelease.sample_binary_packages and
# accessing the property "summary" of its items requires a
# constant number of SQL queries, regardless of the number
# of existing binary package releases.
self.makeBinaryPackageRelease()
self.updateDistroSeriesPackageCache()
with StormStatementRecorder() as recorder:
for ds_package in self.dsp_release.sample_binary_packages:
ds_package.summary
self.assertThat(recorder, HasQueryCount(LessThan(5)))
self.assertEqual(1, self.dsp_release.sample_binary_packages.count())
for iteration in range(5):
self.makeBinaryPackageRelease()
self.updateDistroSeriesPackageCache()
with StormStatementRecorder() as recorder:
for ds_package in self.dsp_release.sample_binary_packages:
ds_package.summary
self.assertThat(recorder, HasQueryCount(LessThan(5)))
self.assertEqual(6, self.dsp_release.sample_binary_packages.count())
# Even if the cache is not updated for binary packages,
# DistributionSourcePackageRelease objects do not try to
# retrieve DistroSeriesPackageCache records if they know
# that such records do not exist.
for iteration in range(5):
self.makeBinaryPackageRelease()
with StormStatementRecorder() as recorder:
for ds_package in self.dsp_release.sample_binary_packages:
ds_package.summary
self.assertThat(recorder, HasQueryCount(LessThan(5)))
self.assertEqual(11, self.dsp_release.sample_binary_packages.count())
|