Coverage for daklib/cruft.py: 87%
86 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-03 16:46 +0000
1"""
2helper functions for cruft-report
4@contact: Debian FTPMaster <ftpmaster@debian.org>
5@copyright 2011 Torsten Werner <twerner@debian.org>
6"""
8# This program is free software; you can redistribute it and/or modify
9# it under the terms of the GNU General Public License as published by
10# the Free Software Foundation; either version 2 of the License, or
11# (at your option) any later version.
13# This program is distributed in the hope that it will be useful,
14# but WITHOUT ANY WARRANTY; without even the implied warranty of
15# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16# GNU General Public License for more details.
18# You should have received a copy of the GNU General Public License
19# along with this program; if not, write to the Free Software
20# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
22################################################################################
24from typing import TYPE_CHECKING, cast, override
26from sqlalchemy import func, select, sql
27from sqlalchemy.engine import CursorResult
28from sqlalchemy.orm import aliased, object_session, with_parent
30from daklib.dbconn import Architecture, DBBinary, DBSource, Suite, get_suite
32if TYPE_CHECKING:
33 from sqlalchemy.orm import Session
34 from sqlalchemy.sql import FromClause, Select
37def newer_version(
38 lowersuite_name: str, highersuite_name: str, session: "Session", include_equal=False
39) -> list[tuple[str, str, str]]:
40 """
41 Finds newer versions in lowersuite_name than in highersuite_name. Returns a
42 list of tuples (source, higherversion, lowerversion) where higherversion is
43 the newest version from highersuite_name and lowerversion is the newest
44 version from lowersuite_name.
45 """
47 lowersuite = get_suite(lowersuite_name, session)
48 assert lowersuite is not None
49 highersuite = get_suite(highersuite_name, session)
50 assert highersuite is not None
52 def get_suite_sources(suite: Suite) -> "FromClause":
53 q1 = (
54 select(DBSource.source, func.max(DBSource.version).label("version"))
55 .where(with_parent(suite, Suite.sources))
56 .group_by(DBSource.source)
57 .subquery()
58 )
59 return aliased(q1)
61 def get_suite_binaries(suite: Suite) -> "FromClause":
62 q1 = (
63 select(
64 DBBinary.package,
65 DBSource.source,
66 func.max(DBSource.version).label("version"),
67 Architecture.arch_string,
68 func.max(DBBinary.version).label("binversion"),
69 )
70 .where(DBBinary.suites.contains(suite))
71 .join(DBBinary.source)
72 .join(DBBinary.architecture)
73 .group_by(
74 DBBinary.package,
75 DBSource.source,
76 Architecture.arch_string,
77 )
78 .subquery()
79 )
80 return aliased(q1)
82 highq = get_suite_sources(highersuite)
83 lowq = get_suite_sources(lowersuite)
85 query = select(
86 highq.c.source,
87 highq.c.version.label("higherversion"),
88 lowq.c.version.label("lowerversion"),
89 ).join(lowq, highq.c.source == lowq.c.source)
91 if include_equal:
92 query = query.where(highq.c.version <= lowq.c.version)
93 else:
94 query = query.where(highq.c.version < lowq.c.version)
96 list = []
97 # get all sources that have a higher version in lowersuite than in
98 # highersuite
99 for source, higherversion, lowerversion in session.execute(query):
100 q1 = (
101 select(
102 DBBinary.package,
103 DBSource.source,
104 DBSource.version,
105 Architecture.arch_string,
106 )
107 .where(DBBinary.suites.contains(highersuite))
108 .join(DBBinary.source)
109 .join(DBBinary.architecture)
110 .where(DBSource.source == source)
111 .subquery()
112 )
113 q2 = select(q1.c.arch_string).group_by(q1.c.arch_string)
114 # all architectures for which source has binaries in highersuite
115 archs_high = {x[0] for x in session.execute(q2)}
117 highq = get_suite_binaries(highersuite)
118 lowq = get_suite_binaries(lowersuite)
120 arch_query = (
121 select(highq.c.arch_string)
122 .join(lowq, highq.c.source == lowq.c.source)
123 .where(highq.c.arch_string == lowq.c.arch_string)
124 .where(highq.c.package == lowq.c.package)
125 .where(highq.c.source == source)
126 )
128 if include_equal:
129 arch_query = arch_query.where(
130 highq.c.binversion <= lowq.c.binversion
131 ).where(highq.c.version <= lowq.c.version)
132 else:
133 arch_query = arch_query.where(highq.c.binversion < lowq.c.binversion).where(
134 highq.c.version < lowq.c.version
135 )
137 arch_query = arch_query.group_by(highq.c.arch_string)
139 # all architectures for which source has a newer binary in lowersuite
140 archs_newer = {x[0] for x in session.execute(arch_query)}
142 # if has at least one binary in lowersuite which is newer than the one
143 # in highersuite on each architecture for which source has binaries in
144 # highersuite, we know that the builds for all relevant architecture
145 # are done, so we can remove the old source with it's binaries
146 if archs_newer >= archs_high:
147 list.append((source, higherversion, lowerversion))
149 list.sort()
150 return list
153def get_package_names(suite: Suite) -> "Select[tuple[str]]":
154 """
155 Returns a select() that selects all distinct package names from suite
156 ordered by package name.
157 """
159 return (
160 select(DBBinary.package)
161 .where(with_parent(suite, Suite.binaries))
162 .group_by(DBBinary.package)
163 .order_by(DBBinary.package)
164 )
167class NamedSource:
168 """
169 A source package identified by its name with all of its versions in a
170 suite.
171 """
173 def __init__(self, suite: Suite, source: str):
174 self.source = source
175 query = suite.sources.filter_by(source=source).order_by(DBSource.version)
176 self.versions = [src.version for src in query]
178 @override
179 def __str__(self):
180 return "%s(%s)" % (self.source, ", ".join(self.versions))
183class DejavuBinary:
184 """
185 A binary package identified by its name which gets built by multiple source
186 packages in a suite. The architecture is ignored which leads to the
187 following corner case, e.g.:
189 If a source package 'foo-mips' that builds a binary package 'foo' on mips
190 and another source package 'foo-mipsel' builds a binary package with the
191 same name 'foo' on mipsel then the binary package 'foo' will be reported as
192 built from multiple source packages.
193 """
195 def __init__(self, suite: Suite, package: str):
196 self.package = package
197 session = object_session(suite)
198 assert session is not None
199 # We need a subquery to make sure that both binary and source packages
200 # are in the right suite.
201 bin_query = suite.binaries.filter_by(package=package).subquery()
202 src_query = (
203 select(DBSource.source)
204 .where(with_parent(suite, Suite.sources))
205 .join(bin_query)
206 .order_by(DBSource.source)
207 .group_by(DBSource.source)
208 )
209 self.sources = []
210 rows = session.execute(src_query).all()
211 if len(rows) > 1:
212 for (source,) in rows:
213 self.sources.append(str(NamedSource(suite, source)))
215 def has_multiple_sources(self) -> bool:
216 "Has the package been built by multiple sources?"
217 return len(self.sources) > 1
219 @override
220 def __str__(self):
221 return "%s built by: %s" % (self.package, ", ".join(self.sources))
224def report_multiple_source(suite: Suite) -> None:
225 """
226 Reports binary packages built from multiple source package with different
227 names.
228 """
230 session = object_session(suite)
231 assert session is not None
233 print("Built from multiple source packages")
234 print("-----------------------------------")
235 print()
236 for (package,) in session.execute(get_package_names(suite)):
237 binary = DejavuBinary(suite, package)
238 if binary.has_multiple_sources():
239 print(binary)
240 print()
243def query_without_source(
244 suite_id: int, session: "Session"
245) -> CursorResult[tuple[str, str]]:
246 """searches for arch: all packages from suite that do no longer
247 reference a source package in the same suite
249 subquery unique_binaries: selects all packages with only 1 version
250 in suite since 'dak rm' does not allow to specify version numbers"""
252 query = """
253 with unique_binaries as
254 (select package, max(version) as version, max(source) as source
255 from bin_associations_binaries
256 where architecture = 2 and suite = :suite_id
257 group by package having count(*) = 1)
258 select ub.package, ub.version
259 from unique_binaries ub
260 left join src_associations_src sas
261 on ub.source = sas.src and sas.suite = :suite_id
262 where sas.id is null
263 order by ub.package"""
264 return cast(CursorResult, session.execute(sql.text(query), {"suite_id": suite_id}))
267def queryNBS(
268 suite_id: int, session: "Session"
269) -> CursorResult[tuple[list[str], list[str], str, str]]:
270 """This one is really complex. It searches arch != all packages that
271 are no longer built from current source packages in suite.
273 temp table unique_binaries: will be populated with packages that
274 have only one version in suite because 'dak rm' does not allow
275 specifying version numbers
277 temp table newest_binaries: will be populated with packages that are
278 built from current sources
280 subquery uptodate_arch: returns all architectures built from current
281 sources
283 subquery unique_binaries_uptodate_arch: returns all packages in
284 architectures from uptodate_arch
286 subquery unique_binaries_uptodate_arch_agg: same as
287 unique_binaries_uptodate_arch but with column architecture
288 aggregated to array
290 subquery uptodate_packages: similar to uptodate_arch but returns all
291 packages built from current sources
293 subquery outdated_packages: returns all packages with architectures
294 no longer built from current source
295 """
297 query = """
298with
299 unique_binaries as
300 (select
301 bab.package,
302 bab.architecture,
303 max(bab.source) as source
304 from bin_associations_binaries bab
305 where bab.suite = :suite_id and bab.architecture > 2
306 group by package, architecture having count(*) = 1),
307 newest_binaries as
308 (select ub.package, ub.architecture, nsa.source, nsa.version
309 from unique_binaries ub
310 join newest_src_association nsa
311 on ub.source = nsa.src and nsa.suite = :suite_id),
312 uptodate_arch as
313 (select architecture, source, version
314 from newest_binaries
315 group by architecture, source, version),
316 unique_binaries_uptodate_arch as
317 (select ub.package, (select a.arch_string from architecture a where a.id = ub.architecture) as arch_string, ua.source, ua.version
318 from unique_binaries ub
319 join source s
320 on ub.source = s.id
321 join uptodate_arch ua
322 on ub.architecture = ua.architecture and s.source = ua.source),
323 unique_binaries_uptodate_arch_agg as
324 (select ubua.package,
325 array_agg(ubua.arch_string order by ubua.arch_string) as arch_list,
326 ubua.source, ubua.version
327 from unique_binaries_uptodate_arch ubua
328 group by ubua.source, ubua.version, ubua.package),
329 uptodate_packages as
330 (select package, source, version
331 from newest_binaries
332 group by package, source, version),
333 outdated_packages as
334 (select array_agg(package order by package) as pkg_list,
335 arch_list, source, version
336 from unique_binaries_uptodate_arch_agg
337 where package not in
338 (select package from uptodate_packages)
339 group by arch_list, source, version)
340 select * from outdated_packages order by source"""
341 return cast(CursorResult, session.execute(sql.text(query), {"suite_id": suite_id}))
344def queryNBS_metadata(
345 suite_id: int, session: "Session"
346) -> CursorResult[tuple[str, str]]:
347 """searches for NBS packages based on metadata extraction of the
348 newest source for a given suite"""
350 query = """
351 select string_agg(bin.package, ' ' order by bin.package), (
352 select arch_string
353 from architecture
354 where id = bin.architecture) as architecture, src.source, newsrc.version
355 from bin_associations_binaries bin
356 join src_associations_src src
357 on src.src = bin.source
358 and src.suite = bin.suite
359 join newest_src_association newsrc
360 on newsrc.source = src.source
361 and newsrc.version != src.version
362 and newsrc.suite = bin.suite
363 where bin.suite = :suite_id
364 and bin.package not in (
365 select trim(' \n' from unnest(string_to_array(meta.value, ',')))
366 from source_metadata meta
367 where meta.src_id = (
368 select newsrc.src
369 from newest_src_association newsrc
370 where newsrc.source = (
371 select s.source
372 from source s
373 where s.id = bin.source)
374 and newsrc.suite = bin.suite)
375 and key_id = (
376 select key_id
377 from metadata_keys
378 where key = 'Binary'))
379 group by src.source, newsrc.version, architecture
380 order by src.source, newsrc.version, bin.architecture"""
381 return cast(CursorResult, session.execute(sql.text(query), {"suite_id": suite_id}))