Coverage for daklib/queue.py: 23%
118 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-03 16:46 +0000
1# vim:set et sw=4:
3"""
4Queue utility functions for dak
6@contact: Debian FTP Master <ftpmaster@debian.org>
7@copyright: 2001 - 2006 James Troup <james@nocrew.org>
8@copyright: 2009, 2010 Joerg Jaspert <joerg@debian.org>
9@license: GNU General Public License version 2 or later
10"""
12# This program is free software; you can redistribute it and/or modify
13# it under the terms of the GNU General Public License as published by
14# the Free Software Foundation; either version 2 of the License, or
15# (at your option) any later version.
17# This program is distributed in the hope that it will be useful,
18# but WITHOUT ANY WARRANTY; without even the implied warranty of
19# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
20# GNU General Public License for more details.
22# You should have received a copy of the GNU General Public License
23# along with this program; if not, write to the Free Software
24# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
26###############################################################################
28from collections.abc import Iterable
29from typing import TYPE_CHECKING, Literal
31from sqlalchemy import select
33from . import utils
34from .config import Config
35from .dbconn import (
36 Architecture,
37 DBBinary,
38 DBSource,
39 NewComment,
40 PolicyQueueUpload,
41 Priority,
42 Section,
43 Suite,
44 get_mapped_component,
45)
46from .regexes import re_default_answer
48if TYPE_CHECKING:
49 from sqlalchemy.engine import Row
50 from sqlalchemy.orm import Session
52 from .policy import MissingOverride
54################################################################################
57def check_valid(overrides: list["MissingOverride"], session: "Session") -> bool:
58 """Check if section and priority for new overrides exist in database.
60 Additionally does sanity checks:
61 - debian-installer packages have to be udeb (or source)
62 - non debian-installer packages cannot be udeb
64 :param overrides: list of overrides to check. The overrides need
65 to be given in form of a dict with the following keys:
67 - package: package name
68 - priority
69 - section
70 - component
71 - type: type of requested override ('dsc', 'deb' or 'udeb')
73 All values are strings.
74 :return: :const:`True` if all overrides are valid, :const:`False` if there is any
75 invalid override.
76 """
77 all_valid = True
78 for o in overrides:
79 o["valid"] = True
80 priority_q = select(Priority).where(Priority.priority == o["priority"]).limit(1)
81 if session.scalars(priority_q).first() is None: 81 ↛ 82line 81 didn't jump to line 82 because the condition on line 81 was never true
82 o["valid"] = False
83 section_q = select(Section).where(Section.section == o["section"]).limit(1)
84 if session.scalars(section_q).first() is None: 84 ↛ 85line 84 didn't jump to line 85 because the condition on line 84 was never true
85 o["valid"] = False
86 if get_mapped_component(o["component"], session) is None: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true
87 o["valid"] = False
88 if o["type"] not in ("dsc", "deb", "udeb"): 88 ↛ 89line 88 didn't jump to line 89 because the condition on line 88 was never true
89 raise Exception("Unknown override type {0}".format(o["type"]))
90 if o["type"] == "udeb" and o["section"].split("/", 1)[-1] != "debian-installer": 90 ↛ 91line 90 didn't jump to line 91 because the condition on line 90 was never true
91 o["valid"] = False
92 if o["section"].split("/", 1)[-1] == "debian-installer" and o["type"] not in ( 92 ↛ 96line 92 didn't jump to line 96 because the condition on line 92 was never true
93 "dsc",
94 "udeb",
95 ):
96 o["valid"] = False
97 all_valid = all_valid and o["valid"]
98 return all_valid
101###############################################################################
104def prod_maintainer(
105 notes: Iterable[NewComment],
106 upload: PolicyQueueUpload,
107 session: "Session",
108 trainee=False,
109) -> Literal[0] | None:
110 cnf = Config()
111 changes = upload.changes
112 whitelists = [upload.target_suite.mail_whitelist]
114 # Here we prepare an editor and get them ready to prod...
115 prod_message = "\n\n=====\n\n".join([note.comment for note in notes])
116 answer = "E"
117 while answer == "E":
118 prod_message = utils.call_editor(prod_message)
119 print("Prod message:")
120 print(
121 utils.prefix_multi_line_string(prod_message, " ", include_blank_lines=True)
122 )
123 prompt = "[P]rod, Bug, Edit, Abandon, Quit ?"
124 answer = "XXX"
125 while prompt.find(answer) == -1:
126 answer = utils.input_or_exit(prompt)
127 m = re_default_answer.search(prompt)
128 if answer == "":
129 assert m is not None
130 answer = m.group(1)
131 answer = answer[:1].upper()
132 if answer == "A":
133 return None
134 elif answer == "Q":
135 return 0
136 # Otherwise, do the proding...
137 user_email_address = utils.whoami() + " <%s>" % (cnf["Dinstall::MyAdminAddress"])
139 is_bug = answer == "B"
141 changed_by = changes.changedby or changes.maintainer
142 maintainer = changes.maintainer
143 maintainer_to = utils.mail_addresses_for_upload(
144 maintainer, changed_by, changes.fingerprint, changes.authorized_by_fingerprint
145 )
147 Subst = {
148 "__SOURCE__": upload.changes.source,
149 "__VERSION__": upload.changes.version,
150 "__ARCHITECTURE__": upload.changes.architecture,
151 "__CHANGES_FILENAME__": upload.changes.changesname,
152 "__MAINTAINER_TO__": ", ".join(maintainer_to),
153 }
155 Subst["__FROM_ADDRESS__"] = user_email_address
156 Subst["__PROD_MESSAGE__"] = prod_message
157 Subst["__CC__"] = "Cc: " + cnf["Dinstall::MyEmailAddress"]
159 if is_bug:
160 Subst["__DEBBUGS_CC__"] = ", ".join(
161 maintainer_to + [cnf["Dinstall::MyEmailAddress"]]
162 )
163 prod_mail_message = utils.TemplateSubst(
164 Subst, cnf["Dir::Templates"] + "/process-new.bug"
165 )
166 else:
167 prod_mail_message = utils.TemplateSubst(
168 Subst, cnf["Dir::Templates"] + "/process-new.prod"
169 )
171 # Send the prod mail
172 utils.send_mail(prod_mail_message, whitelists=whitelists)
174 if is_bug:
175 print("Filed bug against source package")
176 else:
177 print("Sent prodding message")
179 answer = utils.input_or_exit("Store prod message as note? (Y/n)?").lower()
180 if answer != "n":
181 comment = NewComment()
182 comment.policy_queue = upload.policy_queue
183 comment.package = upload.changes.source
184 comment.version = upload.changes.version
185 comment.comment = prod_mail_message
186 comment.author = utils.whoami()
187 comment.trainee = trainee
188 session.add(comment)
189 session.commit()
191 return None
194################################################################################
197def edit_note(
198 upload: PolicyQueueUpload, session: "Session", trainee=False
199) -> Literal[0] | None:
200 newnote = ""
201 answer = "E"
202 while answer == "E":
203 newnote = utils.call_editor(newnote).rstrip()
204 print("New Note:")
205 print(utils.prefix_multi_line_string(newnote, " "))
206 empty_note = not newnote.strip()
207 if empty_note:
208 prompt = "Done, Edit, [A]bandon, Quit ?"
209 else:
210 prompt = "[D]one, Edit, Abandon, Quit ?"
211 answer = "XXX"
212 while prompt.find(answer) == -1:
213 answer = utils.input_or_exit(prompt)
214 m = re_default_answer.search(prompt)
215 if answer == "":
216 assert m is not None
217 answer = m.group(1)
218 answer = answer[:1].upper()
219 if answer == "A":
220 return None
221 elif answer == "Q":
222 return 0
224 comment = NewComment()
225 comment.policy_queue = upload.policy_queue
226 comment.package = upload.changes.source
227 comment.version = upload.changes.version
228 comment.comment = newnote
229 comment.author = utils.whoami()
230 comment.trainee = trainee
231 session.add(comment)
232 session.commit()
234 return None
237###############################################################################
240def get_suite_version_by_source(
241 source: str, session: "Session"
242) -> "list[Row[tuple[str, str]]]":
243 "returns a list of tuples (suite_name, version) for source package"
244 q = (
245 select(Suite.suite_name, DBSource.version)
246 .join(Suite.sources)
247 .where(DBSource.source == source)
248 )
249 return list(session.execute(q))
252def get_suite_version_by_package(
253 package: str, arch_string: str, session: "Session"
254) -> "list[Row[tuple[str, str]]]":
255 """
256 returns a list of tuples (suite_name, version) for binary package and
257 arch_string
258 """
259 q = (
260 select(Suite.suite_name, DBBinary.version)
261 .join(Suite.binaries)
262 .where(DBBinary.package == package)
263 .join(DBBinary.architecture)
264 .where(Architecture.arch_string.in_([arch_string, "all"]))
265 )
266 return list(session.execute(q))