Coverage for britney2/policies/policy.py: 92%
1373 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-18 12:43 +0000
1import json
2import logging
3import optparse
4import os
5import re
6import sys
7import time
8from abc import ABC, abstractmethod
9from collections import defaultdict
10from collections.abc import Callable, Container, Iterator
11from dataclasses import dataclass, field
12from enum import Enum, IntEnum, StrEnum, auto, unique
13from itertools import chain
14from typing import TYPE_CHECKING, Any, Generic, Optional, TypeVar, cast
15from urllib.parse import quote
17import apt_pkg
18import yaml
19from more_itertools import first
21from britney2 import (
22 BinaryPackage,
23 BinaryPackageId,
24 DependencyType,
25 PackageId,
26 SourcePackage,
27 Suite,
28 SuiteClass,
29 Suites,
30 TargetSuite,
31)
32from britney2.excusedeps import DependencySpec
33from britney2.hints import (
34 Hint,
35 HintAnnotate,
36 HintCollection,
37 HintParser,
38 HintType,
39 PolicyHintParserProto,
40)
41from britney2.inputs.suiteloader import SuiteContentLoader
42from britney2.migrationitem import MigrationItem, MigrationItemFactory
43from britney2.policies import ApplySrcPolicy, PolicyVerdict
44from britney2.utils import (
45 binaries_from_source_version,
46 compute_reverse_tree,
47 filter_out_faux,
48 filter_out_faux_gen,
49 find_newer_binaries,
50 get_component,
51 get_dependency_solvers,
52 is_smooth_update_allowed,
53 parse_option,
54)
56if TYPE_CHECKING: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true
57 from ..britney import Britney
58 from ..excuse import Excuse
59 from ..installability.universe import BinaryPackageUniverse
62class PolicyLoadRequest:
63 __slots__ = ("_options_name", "_default_value", "_policy_constructor")
65 def __init__(
66 self,
67 policy_constructor: Callable[[optparse.Values, Suites], "BasePolicy"],
68 options_name: str | None,
69 default_value: bool,
70 ) -> None:
71 self._policy_constructor = policy_constructor
72 self._options_name = options_name
73 self._default_value = default_value
75 def is_enabled(self, options: optparse.Values) -> bool:
76 if self._options_name is None:
77 assert self._default_value
78 return True
79 actual_value = getattr(options, self._options_name, None)
80 if actual_value is None:
81 return self._default_value
82 return actual_value.lower() in ("yes", "y", "true", "t")
84 def load(self, options: optparse.Values, suite_info: Suites) -> "BasePolicy":
85 return self._policy_constructor(options, suite_info)
87 @classmethod
88 def always_load(
89 cls, policy_constructor: Callable[[optparse.Values, Suites], "BasePolicy"]
90 ) -> "PolicyLoadRequest":
91 return cls(policy_constructor, None, True)
93 @classmethod
94 def conditionally_load(
95 cls,
96 policy_constructor: Callable[[optparse.Values, Suites], "BasePolicy"],
97 option_name: str,
98 default_value: bool,
99 ) -> "PolicyLoadRequest":
100 return cls(policy_constructor, option_name, default_value)
103class PolicyEngine:
104 def __init__(self) -> None:
105 self._policies: list["BasePolicy"] = []
107 def add_policy(self, policy: "BasePolicy") -> None:
108 self._policies.append(policy)
110 def load_policies(
111 self,
112 options: optparse.Values,
113 suite_info: Suites,
114 policy_load_requests: list[PolicyLoadRequest],
115 ) -> None:
116 for policy_load_request in policy_load_requests:
117 if policy_load_request.is_enabled(options):
118 self.add_policy(policy_load_request.load(options, suite_info))
120 def register_policy_hints(self, hint_parser: HintParser) -> None:
121 for policy in self._policies:
122 policy.register_hints(hint_parser)
124 def initialise(self, britney: "Britney", hints: HintCollection) -> None:
125 for policy in self._policies:
126 policy.hints = hints
127 policy.initialise(britney)
129 def save_state(self, britney: "Britney") -> None:
130 for policy in self._policies:
131 policy.save_state(britney)
133 def apply_src_policies(
134 self,
135 source_t: SourcePackage | None,
136 source_u: SourcePackage,
137 excuse: "Excuse",
138 ) -> None:
139 excuse_verdict = excuse.policy_verdict
140 source_suite = excuse.item.suite
141 suite_class = source_suite.suite_class
142 for policy in self._policies:
143 pinfo: dict[str, Any] = {}
144 policy_verdict = PolicyVerdict.NOT_APPLICABLE
145 if suite_class in policy.applicable_suites:
146 if policy.src_policy.run_arch:
147 for arch in policy.options.architectures:
148 v = policy.apply_srcarch_policy_impl(
149 pinfo, arch, source_t, source_u, excuse
150 )
151 policy_verdict = PolicyVerdict.worst_of(policy_verdict, v)
152 if policy.src_policy.run_src:
153 v = policy.apply_src_policy_impl(pinfo, source_t, source_u, excuse)
154 policy_verdict = PolicyVerdict.worst_of(policy_verdict, v)
155 # The base policy provides this field, so the subclass should leave it blank
156 assert "verdict" not in pinfo
157 if policy_verdict is not PolicyVerdict.NOT_APPLICABLE:
158 excuse.policy_info[policy.policy_id] = pinfo
159 pinfo["verdict"] = policy_verdict.name
160 excuse_verdict = PolicyVerdict.worst_of(policy_verdict, excuse_verdict)
161 excuse.policy_verdict = excuse_verdict
163 def apply_srcarch_policies(
164 self,
165 arch: str,
166 source_t: SourcePackage | None,
167 source_u: SourcePackage,
168 excuse: "Excuse",
169 ) -> None:
170 excuse_verdict = excuse.policy_verdict
171 source_suite = excuse.item.suite
172 suite_class = source_suite.suite_class
173 for policy in self._policies:
174 pinfo: dict[str, Any] = {}
175 if suite_class in policy.applicable_suites:
176 policy_verdict = policy.apply_srcarch_policy_impl(
177 pinfo, arch, source_t, source_u, excuse
178 )
179 excuse_verdict = PolicyVerdict.worst_of(policy_verdict, excuse_verdict)
180 # The base policy provides this field, so the subclass should leave it blank
181 assert "verdict" not in pinfo
182 if policy_verdict is not PolicyVerdict.NOT_APPLICABLE:
183 excuse.policy_info[policy.policy_id] = pinfo
184 pinfo["verdict"] = policy_verdict.name
185 excuse.policy_verdict = excuse_verdict
188class BasePolicy(ABC):
189 britney: "Britney"
190 policy_id: str
191 hints: HintCollection | None
192 applicable_suites: set[SuiteClass]
193 src_policy: ApplySrcPolicy
194 options: optparse.Values
195 suite_info: Suites
197 def __init__(
198 self,
199 options: optparse.Values,
200 suite_info: Suites,
201 ) -> None:
202 """The BasePolicy constructor
204 :param options: The options member of Britney with all the
205 config values.
206 """
208 @property
209 @abstractmethod
210 def state_dir(self) -> str: ... 210 ↛ exitline 210 didn't return from function 'state_dir' because
212 def register_hints(self, hint_parser: HintParser) -> None: # pragma: no cover
213 """Register new hints that this policy accepts
215 :param hint_parser: (see HintParser.register_hint_type)
216 """
218 def initialise(self, britney: "Britney") -> None: # pragma: no cover
219 """Called once to make the policy initialise any data structures
221 This is useful for e.g. parsing files or other "heavy do-once" work.
223 :param britney: This is the instance of the "Britney" class.
224 """
225 self.britney = britney
227 def save_state(self, britney: "Britney") -> None: # pragma: no cover
228 """Called once at the end of the run to make the policy save any persistent data
230 Note this will *not* be called for "dry-runs" as such runs should not change
231 the state.
233 :param britney: This is the instance of the "Britney" class.
234 """
236 def apply_src_policy_impl(
237 self,
238 policy_info: dict[str, Any],
239 source_data_tdist: SourcePackage | None,
240 source_data_srcdist: SourcePackage,
241 excuse: "Excuse",
242 ) -> PolicyVerdict: # pragma: no cover
243 """Apply a policy on a given source migration
245 Britney will call this method on a given source package, when
246 Britney is considering to migrate it from the given source
247 suite to the target suite. The policy will then evaluate the
248 the migration and then return a verdict.
250 :param policy_info: A dictionary of all policy results. The
251 policy can add a value stored in a key related to its name.
252 (e.g. policy_info['age'] = {...}). This will go directly into
253 the "excuses.yaml" output.
255 :param source_data_tdist: Information about the source package
256 in the target distribution (e.g. "testing"). This is the
257 data structure in source_suite.sources[source_name]
259 :param source_data_srcdist: Information about the source
260 package in the source distribution (e.g. "unstable" or "tpu").
261 This is the data structure in target_suite.sources[source_name]
263 :return: A Policy Verdict (e.g. PolicyVerdict.PASS)
264 """
265 return PolicyVerdict.NOT_APPLICABLE
267 def apply_srcarch_policy_impl(
268 self,
269 policy_info: dict[str, Any],
270 arch: str,
271 source_data_tdist: SourcePackage | None,
272 source_data_srcdist: SourcePackage,
273 excuse: "Excuse",
274 ) -> PolicyVerdict:
275 """Apply a policy on a given binary migration
277 Britney will call this method on binaries from a given source package
278 on a given architecture, when Britney is considering to migrate them
279 from the given source suite to the target suite. The policy will then
280 evaluate the migration and then return a verdict.
282 :param policy_info: A dictionary of all policy results. The
283 policy can add a value stored in a key related to its name.
284 (e.g. policy_info['age'] = {...}). This will go directly into
285 the "excuses.yaml" output.
287 :param arch: The architecture the item is applied to. This is mostly
288 relevant for policies where src_policy is not ApplySrcPolicy.RUN_SRC
289 (as that is the only case where arch can differ from item.architecture)
291 :param source_data_tdist: Information about the source package
292 in the target distribution (e.g. "testing"). This is the
293 data structure in source_suite.sources[source_name]
295 :param source_data_srcdist: Information about the source
296 package in the source distribution (e.g. "unstable" or "tpu").
297 This is the data structure in target_suite.sources[source_name]
299 :return: A Policy Verdict (e.g. PolicyVerdict.PASS)
300 """
301 # if the policy doesn't implement this function, assume it's OK
302 return PolicyVerdict.NOT_APPLICABLE
305class AbstractBasePolicy(BasePolicy):
306 """
307 A shared abstract class for building BasePolicy objects.
309 tests/test_policy.py:initialize_policy() needs to be able to build BasePolicy
310 objects with just a two-item constructor, while all other uses of BasePolicy-
311 derived objects need the 5-item constructor. So AbstractBasePolicy was split
312 out to document this.
313 """
315 def __init__(
316 self,
317 policy_id: str,
318 options: optparse.Values,
319 suite_info: Suites,
320 applicable_suites: set[SuiteClass],
321 src_policy: ApplySrcPolicy = ApplySrcPolicy.RUN_SRC,
322 ) -> None:
323 """Concrete initializer.
325 :param policy_id: Identifies the policy. It will
326 determine the key used for the excuses.yaml etc.
328 :param options: The options member of Britney with all the
329 config values.
331 :param applicable_suites: Where this policy applies.
332 """
333 self.policy_id = policy_id
334 self.options = options
335 self.suite_info = suite_info
336 self.applicable_suites = applicable_suites
337 self.src_policy = src_policy
338 self.hints: HintCollection | None = None
339 logger_name = ".".join((self.__class__.__module__, self.__class__.__name__))
340 self.logger = logging.getLogger(logger_name)
342 @property
343 def state_dir(self) -> str:
344 return cast(str, self.options.state_dir)
347_T = TypeVar("_T")
350class SimplePolicyHint(Hint, Generic[_T]):
351 def __init__(
352 self,
353 user: str,
354 hint_type: HintType,
355 policy_parameter: _T,
356 packages: list[MigrationItem],
357 ) -> None:
358 super().__init__(user, hint_type, packages)
359 self._policy_parameter = policy_parameter
361 def __eq__(self, other: Any) -> bool:
362 if self.type != other.type or self._policy_parameter != other._policy_parameter:
363 return False
364 return super().__eq__(other)
366 def str(self) -> str:
367 return "{} {} {}".format(
368 self._type,
369 str(self._policy_parameter),
370 " ".join(x.name for x in self._packages),
371 )
374class AgeDayHint(SimplePolicyHint[int]):
375 @property
376 def days(self) -> int:
377 return self._policy_parameter
380class IgnoreRCBugHint(SimplePolicyHint[frozenset[str]]):
381 @property
382 def ignored_rcbugs(self) -> frozenset[str]:
383 return self._policy_parameter
386def simple_policy_hint_parser_function(
387 class_name: Callable[[str, HintType, _T, list[MigrationItem]], Hint],
388 converter: Callable[[str], _T],
389) -> PolicyHintParserProto:
390 def f(
391 mi_factory: MigrationItemFactory,
392 hints: HintCollection,
393 who: str,
394 hint_type: HintType,
395 *args: str,
396 ) -> None:
397 policy_parameter = args[0]
398 for item in mi_factory.parse_items(args[1:]):
399 hints.add_hint(
400 class_name(who, hint_type, converter(policy_parameter), [item])
401 )
403 return f
406class AgePolicy(AbstractBasePolicy):
407 """Configurable Aging policy for source migrations
409 The AgePolicy will let packages stay in the source suite for a pre-defined
410 amount of days before letting migrate (based on their urgency, if any).
412 The AgePolicy's decision is influenced by the following:
414 State files:
415 * ${STATE_DIR}/age-policy-urgencies: File containing urgencies for source
416 packages. Note that urgencies are "sticky" and the most "urgent" urgency
417 will be used (i.e. the one with lowest age-requirements).
418 - This file needs to be updated externally, if the policy should take
419 urgencies into consideration. If empty (or not updated), the policy
420 will simply use the default urgency (see the "Config" section below)
421 - In Debian, these values are taken from the .changes file, but that is
422 not a requirement for Britney.
423 * ${STATE_DIR}/age-policy-dates: File containing the age of all source
424 packages.
425 - The policy will automatically update this file.
426 Config:
427 * DEFAULT_URGENCY: Name of the urgency used for packages without an urgency
428 (or for unknown urgencies). Will also be used to set the "minimum"
429 aging requirements for packages not in the target suite.
430 * MINDAYS_<URGENCY>: The age-requirements in days for packages with the
431 given urgency.
432 - Commonly used urgencies are: low, medium, high, emergency, critical
433 Hints:
434 * urgent <source>/<version>: Disregard the age requirements for a given
435 source/version.
436 * age-days X <source>/<version>: Set the age requirements for a given
437 source/version to X days. Note that X can exceed the highest
438 age-requirement normally given.
440 """
442 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
443 super().__init__("age", options, suite_info, {SuiteClass.PRIMARY_SOURCE_SUITE})
444 self._min_days = self._generate_mindays_table()
445 self._min_days_default = 0
446 # britney's "day" begins at 7pm (we want aging to occur in the 22:00Z run and we run Britney 2-4 times a day)
447 # NB: _date_now is used in tests
448 time_now = time.time()
449 if hasattr(self.options, "fake_runtime"):
450 time_now = int(self.options.fake_runtime)
451 self.logger.info("overriding runtime with fake_runtime %d", time_now)
453 self._date_now = int(((time_now / (60 * 60)) - 19) / 24)
454 self._dates: dict[str, tuple[str, int]] = {}
455 self._urgencies: dict[str, str] = {}
456 self._default_urgency: str = self.options.default_urgency
457 self._penalty_immune_urgencies: frozenset[str] = frozenset()
458 if hasattr(self.options, "no_penalties"):
459 self._penalty_immune_urgencies = frozenset(
460 x.strip() for x in self.options.no_penalties.split()
461 )
462 self._bounty_min_age: int | None = None # initialised later
464 def _generate_mindays_table(self) -> dict[str, int]:
465 mindays: dict[str, int] = {}
466 for k in dir(self.options):
467 if not k.startswith("mindays_"):
468 continue
469 v = getattr(self.options, k)
470 try:
471 as_days = int(v)
472 except ValueError:
473 raise ValueError(
474 "Unable to parse "
475 + k
476 + " as a number of days. Must be 0 or a positive integer"
477 )
478 if as_days < 0: 478 ↛ 479line 478 didn't jump to line 479 because the condition on line 478 was never true
479 raise ValueError(
480 "The value of " + k + " must be zero or a positive integer"
481 )
482 mindays[k.split("_")[1]] = as_days
483 return mindays
485 def register_hints(self, hint_parser: HintParser) -> None:
486 hint_parser.register_hint_type(
487 HintType(
488 "age-days",
489 simple_policy_hint_parser_function(AgeDayHint, int),
490 min_args=2,
491 )
492 )
493 hint_parser.register_hint_type(HintType("urgent"))
495 def initialise(self, britney: "Britney") -> None:
496 super().initialise(britney)
497 self._read_dates_file()
498 self._read_urgencies_file()
499 if self._default_urgency not in self._min_days: # pragma: no cover
500 raise ValueError(
501 f"Missing age-requirement for default urgency (MINDAYS_{self._default_urgency})"
502 )
503 self._min_days_default = self._min_days[self._default_urgency]
504 try:
505 self._bounty_min_age = int(self.options.bounty_min_age)
506 except ValueError: 506 ↛ 507line 506 didn't jump to line 507 because the exception caught by line 506 didn't happen
507 if self.options.bounty_min_age in self._min_days:
508 self._bounty_min_age = self._min_days[self.options.bounty_min_age]
509 else: # pragma: no cover
510 raise ValueError(
511 "Please fix BOUNTY_MIN_AGE in the britney configuration"
512 )
513 except AttributeError:
514 # The option wasn't defined in the configuration
515 self._bounty_min_age = 0
517 def save_state(self, britney: "Britney") -> None:
518 super().save_state(britney)
519 self._write_dates_file()
521 def apply_src_policy_impl(
522 self,
523 age_info: dict[str, Any],
524 source_data_tdist: SourcePackage | None,
525 source_data_srcdist: SourcePackage,
526 excuse: "Excuse",
527 ) -> PolicyVerdict:
528 # retrieve the urgency for the upload, ignoring it if this is a NEW package
529 # (not present in the target suite)
530 source_name = excuse.item.package
531 urgency = self._urgencies.get(source_name, self._default_urgency)
533 if urgency not in self._min_days: 533 ↛ 534line 533 didn't jump to line 534 because the condition on line 533 was never true
534 age_info["unknown-urgency"] = urgency
535 urgency = self._default_urgency
537 if not source_data_tdist:
538 if self._min_days[urgency] < self._min_days_default:
539 age_info["urgency-reduced"] = {
540 "from": urgency,
541 "to": self._default_urgency,
542 }
543 urgency = self._default_urgency
545 if source_name not in self._dates:
546 self._dates[source_name] = (source_data_srcdist.version, self._date_now)
547 elif self._dates[source_name][0] != source_data_srcdist.version:
548 self._dates[source_name] = (source_data_srcdist.version, self._date_now)
550 days_old = self._date_now - self._dates[source_name][1]
551 min_days = self._min_days[urgency]
552 for bounty, bounty_value in excuse.bounty.items():
553 if bounty_value: 553 ↛ 552line 553 didn't jump to line 552 because the condition on line 553 was always true
554 self.logger.info(
555 "Applying bounty for %s granted by %s: %d days",
556 source_name,
557 bounty,
558 bounty_value,
559 )
560 excuse.addinfo(
561 f"Required age reduced by {bounty_value} days because of {bounty}"
562 )
563 assert bounty_value > 0, "negative bounties shouldn't happen"
564 min_days -= bounty_value
565 if urgency not in self._penalty_immune_urgencies:
566 for penalty, penalty_value in excuse.penalty.items():
567 if penalty_value: 567 ↛ 566line 567 didn't jump to line 566 because the condition on line 567 was always true
568 self.logger.info(
569 "Applying penalty for %s given by %s: %d days",
570 source_name,
571 penalty,
572 penalty_value,
573 )
574 excuse.addinfo(
575 f"Required age increased by {penalty_value} days because of {penalty}"
576 )
577 assert (
578 penalty_value > 0
579 ), "negative penalties should be handled earlier"
580 min_days += penalty_value
582 assert self._bounty_min_age is not None
583 # the age in BOUNTY_MIN_AGE can be higher than the one associated with
584 # the real urgency, so don't forget to take it into account
585 bounty_min_age = min(self._bounty_min_age, self._min_days[urgency])
586 if min_days < bounty_min_age:
587 min_days = bounty_min_age
588 excuse.addinfo(f"Required age is not allowed to drop below {min_days} days")
590 age_info["current-age"] = days_old
592 assert self.hints is not None
593 for hint in self.hints.search(
594 "age-days", package=source_name, version=source_data_srcdist.version
595 ):
596 age_days_hint = cast("AgeDayHint", hint)
598 new_req = age_days_hint.days
599 age_info["age-requirement-reduced"] = {
600 "new-requirement": new_req,
601 "changed-by": age_days_hint.user,
602 }
603 if "original-age-requirement" not in age_info: 603 ↛ 605line 603 didn't jump to line 605 because the condition on line 603 was always true
604 age_info["original-age-requirement"] = min_days
605 min_days = new_req
607 age_info["age-requirement"] = min_days
608 res = PolicyVerdict.PASS
610 if days_old < min_days:
611 if (
612 urgent_hint := self.hints.search_first(
613 "urgent", package=source_name, version=source_data_srcdist.version
614 )
615 ) is not None:
616 age_info["age-requirement-reduced"] = {
617 "new-requirement": 0,
618 "changed-by": urgent_hint.user,
619 }
620 res = PolicyVerdict.PASS_HINTED
621 else:
622 res = PolicyVerdict.REJECTED_TEMPORARILY
624 # update excuse
625 age_hint = age_info.get("age-requirement-reduced")
626 age_min_req = age_info["age-requirement"]
627 if age_hint is not None:
628 new_req = age_hint["new-requirement"]
629 who = age_hint["changed-by"]
630 if new_req:
631 excuse.addinfo(
632 f"Overriding age needed from {age_min_req} days to {new_req} by {who}"
633 )
634 age_min_req = new_req
635 else:
636 excuse.addinfo(f"Too young, but urgency pushed by {who}")
637 age_min_req = 0
638 excuse.setdaysold(age_info["current-age"], age_min_req)
640 if age_min_req == 0:
641 excuse.addinfo(f"{days_old} days old")
642 elif days_old < age_min_req:
643 excuse.add_verdict_info(
644 res, f"Too young, only {days_old} of {age_min_req} days old"
645 )
646 else:
647 excuse.addinfo(f"{days_old} days old (needed {age_min_req} days)")
649 return res
651 def _read_dates_file(self) -> None:
652 """Parse the dates file"""
653 dates = self._dates
654 fallback_filename = os.path.join(self.suite_info.target_suite.path, "Dates")
655 using_new_name = False
656 try:
657 filename = os.path.join(self.state_dir, "age-policy-dates")
658 if not os.path.exists(filename) and os.path.exists(fallback_filename): 658 ↛ 659line 658 didn't jump to line 659 because the condition on line 658 was never true
659 filename = fallback_filename
660 else:
661 using_new_name = True
662 except AttributeError:
663 if os.path.exists(fallback_filename):
664 filename = fallback_filename
665 else:
666 raise RuntimeError("Please set STATE_DIR in the britney configuration")
668 try:
669 with open(filename, encoding="utf-8") as fd:
670 for line in fd:
671 if line.startswith("#"):
672 # Ignore comment lines (mostly used for tests)
673 continue
674 # <source> <version> <date>)
675 ln = line.split()
676 if len(ln) != 3: # pragma: no cover
677 continue
678 try:
679 dates[ln[0]] = (ln[1], int(ln[2]))
680 except ValueError: # pragma: no cover
681 pass
682 except FileNotFoundError:
683 if not using_new_name: 683 ↛ 685line 683 didn't jump to line 685 because the condition on line 683 was never true
684 # If we using the legacy name, then just give up
685 raise
686 self.logger.info("%s does not appear to exist. Creating it", filename)
687 with open(filename, mode="x", encoding="utf-8"):
688 pass
690 def _read_urgencies_file(self) -> None:
691 urgencies = self._urgencies
692 min_days_default = self._min_days_default
693 fallback_filename = os.path.join(self.suite_info.target_suite.path, "Urgency")
694 try:
695 filename = os.path.join(self.state_dir, "age-policy-urgencies")
696 if not os.path.exists(filename) and os.path.exists(fallback_filename): 696 ↛ 697line 696 didn't jump to line 697 because the condition on line 696 was never true
697 filename = fallback_filename
698 except AttributeError:
699 filename = fallback_filename
701 sources_s = self.suite_info.primary_source_suite.sources
702 sources_t = self.suite_info.target_suite.sources
704 with open(filename, errors="surrogateescape", encoding="ascii") as fd:
705 for line in fd:
706 if line.startswith("#"):
707 # Ignore comment lines (mostly used for tests)
708 continue
709 # <source> <version> <urgency>
710 ln = line.split()
711 if len(ln) != 3: 711 ↛ 712line 711 didn't jump to line 712 because the condition on line 711 was never true
712 continue
714 # read the minimum days associated with the urgencies
715 urgency_old = urgencies.get(ln[0], None)
716 mindays_old = self._min_days.get(urgency_old, 1000) # type: ignore[arg-type]
717 mindays_new = self._min_days.get(ln[2], min_days_default)
719 # if the new urgency is lower (so the min days are higher), do nothing
720 if mindays_old <= mindays_new:
721 continue
723 # if the package exists in the target suite and it is more recent, do nothing
724 tsrcv = sources_t.get(ln[0], None)
725 if tsrcv and apt_pkg.version_compare(tsrcv.version, ln[1]) >= 0:
726 continue
728 # if the package doesn't exist in the primary source suite or it is older, do nothing
729 usrcv = sources_s.get(ln[0], None)
730 if not usrcv or apt_pkg.version_compare(usrcv.version, ln[1]) < 0: 730 ↛ 731line 730 didn't jump to line 731 because the condition on line 730 was never true
731 continue
733 # update the urgency for the package
734 urgencies[ln[0]] = ln[2]
736 def _write_dates_file(self) -> None:
737 dates = self._dates
738 try:
739 directory = self.state_dir
740 basename = "age-policy-dates"
741 old_file = os.path.join(self.suite_info.target_suite.path, "Dates")
742 except AttributeError:
743 directory = self.suite_info.target_suite.path
744 basename = "Dates"
745 old_file = None
746 filename = os.path.join(directory, basename)
747 filename_tmp = os.path.join(directory, f"{basename}_new")
748 with open(filename_tmp, "w", encoding="utf-8") as fd:
749 fd.writelines(
750 f"{pkg} {version} {date}\n"
751 for pkg, (version, date) in sorted(dates.items())
752 )
753 os.rename(filename_tmp, filename)
754 if old_file is not None and os.path.exists(old_file): 754 ↛ 755line 754 didn't jump to line 755 because the condition on line 754 was never true
755 self.logger.info("Removing old age-policy-dates file %s", old_file)
756 os.unlink(old_file)
759class RCBugPolicy(AbstractBasePolicy):
760 """RC bug regression policy for source migrations
762 The RCBugPolicy will read provided list of RC bugs and block any
763 source upload that would introduce a *new* RC bug in the target
764 suite.
766 The RCBugPolicy's decision is influenced by the following:
768 State files:
769 * ${STATE_DIR}/rc-bugs-${SUITE_NAME}: File containing RC bugs for packages in
770 the given suite (one for both primary source suite and the target sutie is
771 needed).
772 - These files need to be updated externally.
773 """
775 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
776 super().__init__(
777 "rc-bugs", options, suite_info, {SuiteClass.PRIMARY_SOURCE_SUITE}
778 )
779 self._bugs_source: dict[str, set[str]] | None = None
780 self._bugs_target: dict[str, set[str]] | None = None
782 def register_hints(self, hint_parser: HintParser) -> None:
783 f = simple_policy_hint_parser_function(
784 IgnoreRCBugHint, lambda x: frozenset(x.split(","))
785 )
786 hint_parser.register_hint_type(HintType("ignore-rc-bugs", f, min_args=2))
788 def initialise(self, britney: "Britney") -> None:
789 super().initialise(britney)
790 source_suite = self.suite_info.primary_source_suite
791 target_suite = self.suite_info.target_suite
792 fallback_unstable = os.path.join(source_suite.path, "BugsV")
793 fallback_testing = os.path.join(target_suite.path, "BugsV")
794 try:
795 filename_unstable = os.path.join(
796 self.state_dir, f"rc-bugs-{source_suite.name}"
797 )
798 filename_testing = os.path.join(
799 self.state_dir, f"rc-bugs-{target_suite.name}"
800 )
801 if ( 801 ↛ 807line 801 didn't jump to line 807
802 not os.path.exists(filename_unstable)
803 and not os.path.exists(filename_testing)
804 and os.path.exists(fallback_unstable)
805 and os.path.exists(fallback_testing)
806 ):
807 filename_unstable = fallback_unstable
808 filename_testing = fallback_testing
809 except AttributeError:
810 filename_unstable = fallback_unstable
811 filename_testing = fallback_testing
812 self._bugs_source = self._read_bugs(filename_unstable)
813 self._bugs_target = self._read_bugs(filename_testing)
815 def apply_src_policy_impl(
816 self,
817 rcbugs_info: dict[str, Any],
818 source_data_tdist: SourcePackage | None,
819 source_data_srcdist: SourcePackage,
820 excuse: "Excuse",
821 ) -> PolicyVerdict:
822 assert self._bugs_source is not None # for type checking
823 assert self._bugs_target is not None # for type checking
824 bugs_t = set()
825 bugs_s = set()
826 source_name = excuse.item.package
827 binaries_s = {x.package_name for x in source_data_srcdist.binaries}
828 try:
829 binaries_t = {x.package_name for x in source_data_tdist.binaries} # type: ignore[union-attr]
830 except AttributeError:
831 binaries_t = set()
833 src_key = f"src:{source_name}"
834 if (
835 source_data_tdist is not None
836 and (bugs := self._bugs_target.get(src_key)) is not None
837 ):
838 bugs_t.update(bugs)
839 if (bugs := self._bugs_source.get(src_key)) is not None:
840 bugs_s.update(bugs)
842 for pkg in binaries_s:
843 if (bugs := self._bugs_source.get(pkg)) is not None:
844 bugs_s |= bugs
845 for pkg in binaries_t:
846 if (bugs := self._bugs_target.get(pkg)) is not None:
847 bugs_t |= bugs
849 # The bts seems to support filing source bugs against a binary of the
850 # same name if that binary isn't built by any source. An example is bug
851 # 820347 against Package: juce (in the live-2016-04-11 test). Add those
852 # bugs too.
853 if (
854 source_name not in (binaries_s | binaries_t)
855 and not any(
856 source_name in binaries
857 for binaries in self.suite_info.primary_source_suite.binaries.values()
858 )
859 and not any(
860 source_name in binaries
861 for binaries in self.suite_info.target_suite.binaries.values()
862 )
863 ):
864 if (bugs := self._bugs_source.get(source_name)) is not None:
865 bugs_s |= bugs
866 if (bugs := self._bugs_target.get(source_name)) is not None: 866 ↛ 867line 866 didn't jump to line 867 because the condition on line 866 was never true
867 bugs_t |= bugs
869 # If a package is not in the target suite, it has no RC bugs per
870 # definition. Unfortunately, it seems that the live-data is
871 # not always accurate (e.g. live-2011-12-13 suggests that
872 # obdgpslogger had the same bug in testing and unstable,
873 # but obdgpslogger was not in testing at that time).
874 # - For the curious, obdgpslogger was removed on that day
875 # and the BTS probably had not caught up with that fact.
876 # (https://tracker.debian.org/news/415935)
877 assert (
878 not bugs_t or source_data_tdist
879 ), f"{source_name} had bugs in the target suite but is not present"
881 verdict = PolicyVerdict.PASS
883 assert self.hints is not None
884 for hint in self.hints.search(
885 "ignore-rc-bugs",
886 package=source_name,
887 version=source_data_srcdist.version,
888 ):
889 ignore_hint = cast(IgnoreRCBugHint, hint)
890 ignored_bugs = ignore_hint.ignored_rcbugs
892 # Only handle one hint for now
893 if (ignored := rcbugs_info.get("ignored-bugs")) is not None:
894 self.logger.info(
895 "Ignoring ignore-rc-bugs hint from %s on %s due to another hint from %s",
896 ignore_hint.user,
897 source_name,
898 ignored["issued-by"],
899 )
900 continue
901 if not ignored_bugs.isdisjoint(bugs_s): 901 ↛ 910line 901 didn't jump to line 910 because the condition on line 901 was always true
902 bugs_s -= ignored_bugs
903 bugs_t -= ignored_bugs
904 rcbugs_info["ignored-bugs"] = {
905 "bugs": sorted(ignored_bugs),
906 "issued-by": ignore_hint.user,
907 }
908 verdict = PolicyVerdict.PASS_HINTED
909 else:
910 self.logger.info(
911 "Ignoring ignore-rc-bugs hint from %s on %s as none of %s affect the package",
912 ignore_hint.user,
913 source_name,
914 ignored_bugs,
915 )
917 rcbugs_info["shared-bugs"] = sorted(bugs_s & bugs_t)
918 rcbugs_info["unique-source-bugs"] = sorted(bugs_s - bugs_t)
919 rcbugs_info["unique-target-bugs"] = sorted(bugs_t - bugs_s)
921 # update excuse
922 new_bugs = rcbugs_info["unique-source-bugs"]
923 old_bugs = rcbugs_info["unique-target-bugs"]
924 excuse.setbugs(old_bugs, new_bugs)
926 if new_bugs:
927 verdict = PolicyVerdict.REJECTED_PERMANENTLY
928 excuse.add_verdict_info(
929 verdict,
930 "Updating %s would introduce bugs in %s: %s"
931 % (
932 source_name,
933 self.suite_info.target_suite.name,
934 ", ".join(
935 f'<a href="https://bugs.debian.org/{quote(a)}">#{a}</a>'
936 for a in new_bugs
937 ),
938 ),
939 )
941 if old_bugs:
942 excuse.addinfo(
943 "Updating %s will fix bugs in %s: %s"
944 % (
945 source_name,
946 self.suite_info.target_suite.name,
947 ", ".join(
948 f'<a href="https://bugs.debian.org/{quote(a)}">#{a}</a>'
949 for a in old_bugs
950 ),
951 )
952 )
954 return verdict
956 def _read_bugs(self, filename: str) -> dict[str, set[str]]:
957 """Read the release critical bug summary from the specified file
959 The file contains rows with the format:
961 <package-name> <bug number>[,<bug number>...]
963 The method returns a dictionary where the key is the binary package
964 name and the value is the list of open RC bugs for it.
965 """
966 bugs: dict[str, set[str]] = {}
967 self.logger.info("Loading RC bugs data from %s", filename)
968 with open(filename, encoding="ascii") as f:
969 for line in f:
970 ln = line.split()
971 if len(ln) != 2: # pragma: no cover
972 self.logger.warning("Malformed line found in line %s", line)
973 continue
974 pkg = ln[0]
975 if pkg not in bugs:
976 bugs[pkg] = set()
977 bugs[pkg].update(ln[1].split(","))
978 return bugs
981class PiupartsState(Enum):
982 FAIL = auto()
983 PASS = auto()
984 WAITING = auto()
985 UNKNOWN = auto()
987 @staticmethod
988 def from_str(val: str) -> "PiupartsState":
989 match val:
990 case "F":
991 return PiupartsState.FAIL
992 case "P":
993 return PiupartsState.PASS
994 case "W": 994 ↛ 996line 994 didn't jump to line 996 because the pattern on line 994 always matched
995 return PiupartsState.WAITING
996 case "X":
997 return PiupartsState.UNKNOWN
998 case _:
999 raise ValueError(f"Invalid piuparts state {val}")
1002class PiupartsResult(StrEnum):
1003 PASS = "pass"
1004 REGRESSION = "regression"
1005 FAILED = "failed"
1006 WAITING_FOR_TESTS = "waiting-for-test-results"
1007 CANNOT_BE_TESTED = "cannot-be-tested"
1010class PiupartsPolicy(AbstractBasePolicy):
1011 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
1012 super().__init__(
1013 "piuparts", options, suite_info, {SuiteClass.PRIMARY_SOURCE_SUITE}
1014 )
1015 self._piuparts_source: dict[str, tuple[PiupartsState, str]] | None = None
1016 self._piuparts_target: dict[str, PiupartsState] | None = None
1018 def register_hints(self, hint_parser: HintParser) -> None:
1019 hint_parser.register_hint_type(HintType("ignore-piuparts"))
1021 def initialise(self, britney: "Britney") -> None:
1022 super().initialise(britney)
1023 source_suite = self.suite_info.primary_source_suite
1024 target_suite = self.suite_info.target_suite
1025 try:
1026 filename_unstable = os.path.join(
1027 self.state_dir, f"piuparts-summary-{source_suite.name}.json"
1028 )
1029 filename_testing = os.path.join(
1030 self.state_dir, f"piuparts-summary-{target_suite.name}.json"
1031 )
1032 except AttributeError as e: # pragma: no cover
1033 raise RuntimeError(
1034 "Please set STATE_DIR in the britney configuration"
1035 ) from e
1036 self._piuparts_source = self._read_piuparts_summary(filename_unstable)
1037 self._piuparts_target = self._read_piuparts_summary_without_url(
1038 filename_testing
1039 )
1041 def apply_src_policy_impl(
1042 self,
1043 piuparts_info: dict[str, Any],
1044 source_data_tdist: SourcePackage | None,
1045 source_data_srcdist: SourcePackage,
1046 excuse: "Excuse",
1047 ) -> PolicyVerdict:
1048 assert self._piuparts_source is not None # for type checking
1049 assert self._piuparts_target is not None # for type checking
1050 source_name = excuse.item.package
1052 testing_state = self._piuparts_target.get(source_name)
1053 if testing_state is None:
1054 testing_state = PiupartsState.UNKNOWN
1055 url: str | None
1056 if (data := self._piuparts_source.get(source_name)) is not None:
1057 unstable_state, url = data
1058 else:
1059 unstable_state = PiupartsState.UNKNOWN
1060 url = None
1061 url_html = "(no link yet)"
1062 if url is not None:
1063 url_html = '<a href="{0}">{0}</a>'.format(url)
1065 match unstable_state:
1066 case PiupartsState.PASS:
1067 # Not a regression
1068 msg = f"Piuparts tested OK - {url_html}"
1069 result = PolicyVerdict.PASS
1070 piuparts_info["test-results"] = PiupartsResult.PASS
1071 case PiupartsState.FAIL if testing_state is not PiupartsState.FAIL:
1072 piuparts_info["test-results"] = PiupartsResult.REGRESSION
1073 msg = f"Piuparts regression - {url_html}"
1074 result = PolicyVerdict.REJECTED_PERMANENTLY
1075 case PiupartsState.FAIL:
1076 piuparts_info["test-results"] = PiupartsResult.FAILED
1077 msg = f"Piuparts failure (not a regression) - {url_html}"
1078 result = PolicyVerdict.PASS
1079 case PiupartsState.WAITING:
1080 msg = f"Piuparts check waiting for test results - {url_html}"
1081 result = PolicyVerdict.REJECTED_TEMPORARILY
1082 piuparts_info["test-results"] = PiupartsResult.WAITING_FOR_TESTS
1083 case _:
1084 msg = f"Piuparts can't test {source_name} (not a blocker) - {url_html}"
1085 piuparts_info["test-results"] = PiupartsResult.CANNOT_BE_TESTED
1086 result = PolicyVerdict.PASS
1088 if url is not None:
1089 piuparts_info["piuparts-test-url"] = url
1090 if result.is_rejected:
1091 excuse.add_verdict_info(result, msg)
1092 else:
1093 excuse.addinfo(msg)
1095 if result.is_rejected:
1096 assert self.hints is not None
1097 if (
1098 ignore_hint := self.hints.search_first(
1099 "ignore-piuparts",
1100 package=source_name,
1101 version=source_data_srcdist.version,
1102 )
1103 ) is not None:
1104 piuparts_info["ignored-piuparts"] = {"issued-by": ignore_hint.user}
1105 result = PolicyVerdict.PASS_HINTED
1106 excuse.addinfo(
1107 f"Piuparts issue ignored as requested by {ignore_hint.user}"
1108 )
1110 return result
1112 def _read_piuparts_summary_gen(
1113 self, filename: str
1114 ) -> Iterator[tuple[str, PiupartsState, str]]:
1115 self.logger.info("Loading piuparts report from %s", filename)
1116 with open(filename) as fd: 1116 ↛ exitline 1116 didn't return from function '_read_piuparts_summary_gen' because the return on line 1118 wasn't executed
1117 if os.fstat(fd.fileno()).st_size < 1: 1117 ↛ 1118line 1117 didn't jump to line 1118 because the condition on line 1117 was never true
1118 return
1119 data = json.load(fd)
1120 try:
1121 if (
1122 data["_id"] != "Piuparts Package Test Results Summary"
1123 or data["_version"] != "1.0"
1124 ): # pragma: no cover
1125 raise ValueError(
1126 f"Piuparts results in {filename} does not have the correct ID or version"
1127 )
1128 except KeyError as e: # pragma: no cover
1129 raise ValueError(
1130 f"Piuparts results in {filename} is missing id or version field"
1131 ) from e
1132 for source, suite_data in data["packages"].items():
1133 if len(suite_data) != 1: # pragma: no cover
1134 raise ValueError(
1135 f"Piuparts results in {filename}, the source {source} does not have "
1136 "exactly one result set"
1137 )
1138 item = first(suite_data.values())
1139 state, _, url = item
1140 yield (source, PiupartsState.from_str(state), url)
1142 def _read_piuparts_summary(
1143 self, filename: str
1144 ) -> dict[str, tuple[PiupartsState, str]]:
1145 return {
1146 source: (state, url)
1147 for (source, state, url) in self._read_piuparts_summary_gen(filename)
1148 }
1150 def _read_piuparts_summary_without_url(
1151 self, filename: str
1152 ) -> dict[str, PiupartsState]:
1153 return {
1154 source: state
1155 for (source, state, _) in self._read_piuparts_summary_gen(filename)
1156 }
1159class DependsPolicy(AbstractBasePolicy):
1160 pkg_universe: "BinaryPackageUniverse"
1161 broken_packages: frozenset["BinaryPackageId"]
1162 all_binaries: dict["BinaryPackageId", "BinaryPackage"]
1163 allow_uninst: dict[str, set[str | None]]
1165 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
1166 super().__init__(
1167 "depends",
1168 options,
1169 suite_info,
1170 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
1171 ApplySrcPolicy.RUN_ON_EVERY_ARCH_ONLY,
1172 )
1173 self.nobreakall_arches = None
1174 self.new_arches = None
1175 self.break_arches = None
1177 def initialise(self, britney: "Britney") -> None:
1178 super().initialise(britney)
1179 self.pkg_universe = britney.pkg_universe
1180 self.broken_packages = self.pkg_universe.broken_packages
1181 self.all_binaries = britney.all_binaries
1182 self.nobreakall_arches = self.options.nobreakall_arches
1183 self.new_arches = self.options.new_arches
1184 self.break_arches = self.options.break_arches
1185 self.allow_uninst = britney.allow_uninst
1187 def apply_srcarch_policy_impl(
1188 self,
1189 deps_info: dict[str, Any],
1190 arch: str,
1191 source_data_tdist: SourcePackage | None,
1192 source_data_srcdist: SourcePackage,
1193 excuse: "Excuse",
1194 ) -> PolicyVerdict:
1195 verdict = PolicyVerdict.PASS
1197 assert self.break_arches is not None
1198 assert self.new_arches is not None
1199 if arch in self.break_arches or arch in self.new_arches:
1200 # we don't check these in the policy (TODO - for now?)
1201 return verdict
1203 item = excuse.item
1204 source_suite = item.suite
1205 target_suite = self.suite_info.target_suite
1207 packages_s_a = source_suite.binaries[arch]
1208 packages_t_a = target_suite.binaries[arch]
1210 my_bins = sorted(filter_out_faux_gen(excuse.packages[arch]))
1212 arch_all_installable = set()
1213 arch_arch_installable = set()
1214 consider_it_regression = True
1216 for pkg_id in my_bins:
1217 pkg_name = pkg_id.package_name
1218 binary_u = packages_s_a[pkg_name]
1219 pkg_arch = binary_u.architecture
1221 # in some cases, we want to track the uninstallability of a
1222 # package (because the autopkgtest policy uses this), but we still
1223 # want to allow the package to be uninstallable
1224 skip_dep_check = False
1226 if binary_u.source_version != source_data_srcdist.version:
1227 # don't check cruft in unstable
1228 continue
1230 if item.architecture != "source" and pkg_arch == "all":
1231 # we don't care about the existing arch: all binaries when
1232 # checking a binNMU item, because the arch: all binaries won't
1233 # migrate anyway
1234 skip_dep_check = True
1236 if pkg_arch == "all" and arch not in self.nobreakall_arches:
1237 skip_dep_check = True
1239 if pkg_name in self.allow_uninst[arch]: 1239 ↛ 1242line 1239 didn't jump to line 1242 because the condition on line 1239 was never true
1240 # this binary is allowed to become uninstallable, so we don't
1241 # need to check anything
1242 skip_dep_check = True
1244 if (oldbin := packages_t_a.get(pkg_name)) is not None:
1245 if not target_suite.is_installable(oldbin.pkg_id):
1246 # as the current binary in testing is already
1247 # uninstallable, the newer version is allowed to be
1248 # uninstallable as well, so we don't need to check
1249 # anything
1250 skip_dep_check = True
1251 consider_it_regression = False
1253 if pkg_id in self.broken_packages:
1254 if pkg_arch == "all":
1255 arch_all_installable.add(False)
1256 else:
1257 arch_arch_installable.add(False)
1258 # dependencies can't be satisfied by all the known binaries -
1259 # this certainly won't work...
1260 excuse.add_unsatisfiable_on_arch(arch)
1261 if skip_dep_check:
1262 # ...but if the binary is allowed to become uninstallable,
1263 # we don't care
1264 # we still want the binary to be listed as uninstallable,
1265 continue
1266 verdict = PolicyVerdict.REJECTED_PERMANENTLY
1267 if pkg_name.endswith("-faux-build-depends"): 1267 ↛ 1268line 1267 didn't jump to line 1268 because the condition on line 1267 was never true
1268 name = pkg_name.removesuffix("-faux-build-depends")
1269 excuse.add_verdict_info(
1270 verdict,
1271 f"src:{name} has unsatisfiable build dependency",
1272 )
1273 else:
1274 excuse.add_verdict_info(
1275 verdict, f"{pkg_name}/{arch} has unsatisfiable dependency"
1276 )
1277 excuse.addreason("depends")
1278 else:
1279 if pkg_arch == "all":
1280 arch_all_installable.add(True)
1281 else:
1282 arch_arch_installable.add(True)
1284 if skip_dep_check:
1285 continue
1287 for dep in self.pkg_universe.dependencies_of(pkg_id):
1288 # dep is a list of packages, each of which satisfy the
1289 # dependency
1291 if not dep:
1292 continue
1293 is_ok = False
1294 needed_for_dep = set()
1296 for alternative in dep:
1297 if target_suite.is_pkg_in_the_suite(alternative):
1298 # dep can be satisfied in testing - ok
1299 is_ok = True
1300 break
1301 elif alternative in my_bins:
1302 # can be satisfied by binary from same item: will be
1303 # ok if item migrates
1304 is_ok = True
1305 break
1306 else:
1307 needed_for_dep.add(alternative)
1309 if not is_ok:
1310 spec = DependencySpec(DependencyType.DEPENDS, arch)
1311 excuse.add_package_depends(spec, needed_for_dep)
1313 # The autopkgtest policy needs delicate trade offs for
1314 # non-installability. The current choice (considering source
1315 # migration and only binaries built by the version of the
1316 # source):
1317 #
1318 # * Run autopkgtest if all arch:$arch binaries are installable
1319 # (but some or all arch:all binaries are not)
1320 #
1321 # * Don't schedule nor wait for not installable arch:all only package
1322 # on ! NOBREAKALL_ARCHES
1323 #
1324 # * Run autopkgtest if installability isn't a regression (there are (or
1325 # rather, should) not be a lot of packages in this state, and most
1326 # likely they'll just fail quickly)
1327 #
1328 # * Don't schedule, but wait otherwise
1329 if arch_arch_installable == {True} and False in arch_all_installable:
1330 deps_info.setdefault("autopkgtest_run_anyways", []).append(arch)
1331 elif (
1332 arch not in self.nobreakall_arches
1333 and not arch_arch_installable
1334 and False in arch_all_installable
1335 ):
1336 deps_info.setdefault("arch_all_not_installable", []).append(arch)
1337 elif not consider_it_regression:
1338 deps_info.setdefault("autopkgtest_run_anyways", []).append(arch)
1340 return verdict
1343@unique
1344class BuildDepResult(IntEnum):
1345 # relation is satisfied in target
1346 OK = 1
1347 # relation can be satisfied by other packages in source
1348 DEPENDS = 2
1349 # relation cannot be satisfied
1350 FAILED = 3
1353class BuildDependsPolicy(AbstractBasePolicy):
1355 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
1356 super().__init__(
1357 "build-depends",
1358 options,
1359 suite_info,
1360 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
1361 )
1362 self._all_buildarch: list[str] = []
1364 parse_option(options, "all_buildarch")
1366 def initialise(self, britney: "Britney") -> None:
1367 super().initialise(britney)
1368 if self.options.all_buildarch:
1369 self._all_buildarch = SuiteContentLoader.config_str_as_list(
1370 self.options.all_buildarch, []
1371 )
1373 def apply_src_policy_impl(
1374 self,
1375 build_deps_info: dict[str, Any],
1376 source_data_tdist: SourcePackage | None,
1377 source_data_srcdist: SourcePackage,
1378 excuse: "Excuse",
1379 ) -> PolicyVerdict:
1380 verdict = PolicyVerdict.PASS
1382 # analyze the dependency fields (if present)
1383 if deps := source_data_srcdist.build_deps_arch:
1384 v = self._check_build_deps(
1385 deps,
1386 DependencyType.BUILD_DEPENDS,
1387 build_deps_info,
1388 source_data_srcdist,
1389 excuse,
1390 )
1391 verdict = PolicyVerdict.worst_of(verdict, v)
1393 if ideps := source_data_srcdist.build_deps_indep:
1394 v = self._check_build_deps(
1395 ideps,
1396 DependencyType.BUILD_DEPENDS_INDEP,
1397 build_deps_info,
1398 source_data_srcdist,
1399 excuse,
1400 )
1401 verdict = PolicyVerdict.worst_of(verdict, v)
1403 return verdict
1405 def _get_check_archs(
1406 self, archs: Container[str], dep_type: DependencyType
1407 ) -> list[str]:
1408 oos = self.options.outofsync_arches
1410 if dep_type is DependencyType.BUILD_DEPENDS:
1411 return [
1412 arch
1413 for arch in self.options.architectures
1414 if arch in archs and arch not in oos
1415 ]
1417 # first try the all buildarch
1418 checkarchs = list(self._all_buildarch)
1419 # then try the architectures where this source has arch specific
1420 # binaries (in the order of the architecture config file)
1421 checkarchs.extend(
1422 arch
1423 for arch in self.options.architectures
1424 if arch in archs and arch not in checkarchs
1425 )
1426 # then try all other architectures
1427 checkarchs.extend(
1428 arch for arch in self.options.architectures if arch not in checkarchs
1429 )
1431 # and drop OUTOFSYNC_ARCHES
1432 return [arch for arch in checkarchs if arch not in oos]
1434 def _add_info_for_arch(
1435 self,
1436 arch: str,
1437 excuses_info: dict[str, list[str]],
1438 blockers: dict[str, set[BinaryPackageId]],
1439 results: dict[str, BuildDepResult],
1440 dep_type: DependencyType,
1441 excuse: "Excuse",
1442 verdict: PolicyVerdict,
1443 ) -> PolicyVerdict:
1444 if (packages := blockers.get(arch)) is not None:
1445 # for the solving packages, update the excuse to add the dependencies
1446 for p in packages:
1447 if arch not in self.options.break_arches: 1447 ↛ 1446line 1447 didn't jump to line 1446 because the condition on line 1447 was always true
1448 spec = DependencySpec(dep_type, arch)
1449 excuse.add_package_depends(spec, {p})
1451 if arch in results and results[arch] is BuildDepResult.FAILED:
1452 verdict = PolicyVerdict.worst_of(
1453 verdict, PolicyVerdict.REJECTED_PERMANENTLY
1454 )
1456 if arch in excuses_info:
1457 for excuse_text in excuses_info[arch]:
1458 if verdict.is_rejected: 1458 ↛ 1461line 1458 didn't jump to line 1461 because the condition on line 1458 was always true
1459 excuse.add_verdict_info(verdict, excuse_text)
1460 else:
1461 excuse.addinfo(excuse_text)
1463 return verdict
1465 def _check_build_deps(
1466 self,
1467 deps: str,
1468 dep_type: DependencyType,
1469 build_deps_info: dict[str, Any],
1470 source_data_srcdist: SourcePackage,
1471 excuse: "Excuse",
1472 ) -> PolicyVerdict:
1473 verdict = PolicyVerdict.PASS
1474 any_arch_ok = dep_type is DependencyType.BUILD_DEPENDS_INDEP
1476 britney = self.britney
1478 # local copies for better performance
1479 parse_src_depends = apt_pkg.parse_src_depends
1480 solvers = get_dependency_solvers
1482 source_name = excuse.item.package
1483 source_suite = excuse.item.suite
1484 target_suite = self.suite_info.target_suite
1485 binaries_s = source_suite.binaries
1486 provides_s = source_suite.provides_table
1487 binaries_t = target_suite.binaries
1488 provides_t = target_suite.provides_table
1489 unsat_bd: dict[str, list[str]] = {}
1490 relevant_archs: set[str] = {
1491 binary.architecture
1492 for binary in filter_out_faux_gen(source_data_srcdist.binaries)
1493 if britney.all_binaries[binary].architecture != "all"
1494 }
1496 excuses_info: dict[str, list[str]] = defaultdict(list)
1497 blockers: dict[str, set[BinaryPackageId]] = defaultdict(set)
1498 arch_results = {}
1499 result_archs = defaultdict(list)
1500 bestresult = BuildDepResult.FAILED
1501 check_archs = self._get_check_archs(relevant_archs, dep_type)
1502 if not check_archs:
1503 # when the arch list is empty, we check the b-d on any arch, instead of all archs
1504 # this happens for Build-Depens on a source package that only produces arch: all binaries
1505 any_arch_ok = True
1506 check_archs = self._get_check_archs(
1507 self.options.architectures, DependencyType.BUILD_DEPENDS_INDEP
1508 )
1510 for arch in check_archs:
1511 # retrieve the binary package from the specified suite and arch
1512 binaries_s_a = binaries_s[arch]
1513 provides_s_a = provides_s[arch]
1514 binaries_t_a = binaries_t[arch]
1515 provides_t_a = provides_t[arch]
1516 arch_results[arch] = BuildDepResult.OK
1517 # for every dependency block (formed as conjunction of disjunction)
1518 for block_txt in deps.split(","):
1519 block_list = parse_src_depends(block_txt, False, arch)
1520 # Unlike regular dependencies, some clauses of the Build-Depends(-Arch|-Indep) can be
1521 # filtered out by (e.g.) architecture restrictions. We need to cope with this while
1522 # keeping block_txt and block aligned.
1523 if not block_list:
1524 # Relation is not relevant for this architecture.
1525 continue
1526 block = block_list[0]
1527 # if the block is satisfied in the target suite, then skip the block
1528 if (
1529 next(
1530 solvers(block, binaries_t_a, provides_t_a, build_depends=True),
1531 None,
1532 )
1533 is not None
1534 ):
1535 # Satisfied in the target suite; all ok.
1536 continue
1538 # check if the block can be satisfied in the source suite, and list the solving packages
1539 packages = list(
1540 solvers(block, binaries_s_a, provides_s_a, build_depends=True)
1541 )
1543 # if the dependency can be satisfied by the same source package, skip the block:
1544 # obviously both binary packages will enter the target suite together
1545 if any(source_name == p.source for p in packages): 1545 ↛ 1546line 1545 didn't jump to line 1546 because the condition on line 1545 was never true
1546 continue
1548 # if no package can satisfy the dependency, add this information to the excuse
1549 if not packages:
1550 excuses_info[arch].append(
1551 f"{source_name} unsatisfiable {dep_type} on {arch}: {block_txt.strip()}"
1552 )
1553 if arch not in unsat_bd: 1553 ↛ 1555line 1553 didn't jump to line 1555 because the condition on line 1553 was always true
1554 unsat_bd[arch] = []
1555 unsat_bd[arch].append(block_txt.strip())
1556 arch_results[arch] = BuildDepResult.FAILED
1557 continue
1559 blockers[arch].update(p.pkg_id for p in packages)
1560 if arch_results[arch] < BuildDepResult.DEPENDS:
1561 arch_results[arch] = BuildDepResult.DEPENDS
1563 if any_arch_ok:
1564 if arch_results[arch] < bestresult:
1565 bestresult = arch_results[arch]
1566 result_archs[arch_results[arch]].append(arch)
1567 if bestresult is BuildDepResult.OK:
1568 # we found an architecture where the b-deps-indep are
1569 # satisfied in the target suite, so we can stop
1570 break
1572 if any_arch_ok:
1573 arch = result_archs[bestresult][0]
1574 excuse.add_detailed_info(f"Checking {dep_type.get_description()} on {arch}")
1575 key = "check-%s-on-arch" % dep_type.get_reason()
1576 build_deps_info[key] = arch
1577 verdict = self._add_info_for_arch(
1578 arch,
1579 excuses_info,
1580 blockers,
1581 arch_results,
1582 dep_type,
1583 excuse,
1584 verdict,
1585 )
1587 else:
1588 for arch in check_archs:
1589 verdict = self._add_info_for_arch(
1590 arch,
1591 excuses_info,
1592 blockers,
1593 arch_results,
1594 dep_type,
1595 excuse,
1596 verdict,
1597 )
1599 if unsat_bd:
1600 build_deps_info["unsatisfiable-arch-build-depends"] = unsat_bd
1602 return verdict
1605class BuiltUsingPolicy(AbstractBasePolicy):
1606 """Built-Using policy
1608 Binaries that incorporate (part of) another source package must list these
1609 sources under 'Built-Using'.
1611 This policy checks if the corresponding sources are available in the
1612 target suite. If they are not, but they are candidates for migration, a
1613 dependency is added.
1615 If the binary incorporates a newer version of a source, that is not (yet)
1616 a candidate, we don't want to accept that binary. A rebuild later in the
1617 primary suite wouldn't fix the issue, because that would incorporate the
1618 newer version again.
1620 If the binary incorporates an older version of the source, a newer version
1621 will be accepted as a replacement. We assume that this can be fixed by
1622 rebuilding the binary at some point during the development cycle.
1624 Requiring exact version of the source would not be useful in practice. A
1625 newer upload of that source wouldn't be blocked by this policy, so the
1626 built-using would be outdated anyway.
1628 """
1630 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
1631 super().__init__(
1632 "built-using",
1633 options,
1634 suite_info,
1635 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
1636 ApplySrcPolicy.RUN_ON_EVERY_ARCH_ONLY,
1637 )
1639 def initialise(self, britney: "Britney") -> None:
1640 super().initialise(britney)
1642 def apply_srcarch_policy_impl(
1643 self,
1644 build_deps_info: dict[str, Any],
1645 arch: str,
1646 source_data_tdist: SourcePackage | None,
1647 source_data_srcdist: SourcePackage,
1648 excuse: "Excuse",
1649 ) -> PolicyVerdict:
1650 verdict = PolicyVerdict.PASS
1652 source_suite = excuse.item.suite
1653 target_suite = self.suite_info.target_suite
1654 binaries_s = source_suite.binaries
1656 def check_bu_in_suite(
1657 bu_source: str, bu_version: str, source_suite: Suite
1658 ) -> bool:
1659 if (s_source := source_suite.sources.get(bu_source)) is None:
1660 return False
1661 s_ver = s_source.version
1662 if apt_pkg.version_compare(s_ver, bu_version) >= 0:
1663 dep = PackageId(bu_source, s_ver, "source")
1664 if arch in self.options.break_arches:
1665 excuse.add_detailed_info(
1666 f"Ignoring Built-Using for {pkg_name}/{arch} on {dep.uvname}"
1667 )
1668 else:
1669 spec = DependencySpec(DependencyType.BUILT_USING, arch)
1670 excuse.add_package_depends(spec, {dep})
1671 excuse.add_detailed_info(
1672 f"{pkg_name}/{arch} has Built-Using on {dep.uvname}"
1673 )
1674 return True
1676 return False
1678 for pkg_id in sorted(
1679 x
1680 for x in filter_out_faux_gen(source_data_srcdist.binaries)
1681 if x.architecture == arch
1682 ):
1683 pkg_name = pkg_id.package_name
1685 # retrieve the testing (if present) and unstable corresponding binary packages
1686 binary_s = binaries_s[arch][pkg_name]
1687 if binary_s.builtusing is None:
1688 continue
1690 for bu in binary_s.builtusing:
1691 bu_source = bu[0]
1692 bu_version = bu[1]
1693 if (t_source := target_suite.sources.get(bu_source)) is not None:
1694 t_ver = t_source.version
1695 if apt_pkg.version_compare(t_ver, bu_version) >= 0:
1696 continue
1698 if check_bu_in_suite(bu_source, bu_version, source_suite):
1699 continue
1701 if source_suite.suite_class.is_additional_source and check_bu_in_suite(
1702 bu_source, bu_version, self.suite_info.primary_source_suite
1703 ):
1704 continue
1706 if arch in self.options.break_arches:
1707 excuse.add_detailed_info(
1708 f"Ignoring unsatisfiable Built-Using for {pkg_name}/{arch} on {bu_source} {bu_version}"
1709 )
1710 else:
1711 verdict = PolicyVerdict.worst_of(
1712 verdict, PolicyVerdict.REJECTED_PERMANENTLY
1713 )
1714 excuse.add_verdict_info(
1715 verdict,
1716 f"{pkg_name}/{arch} has unsatisfiable Built-Using on {bu_source} {bu_version}",
1717 )
1719 return verdict
1722class BlockPolicy(AbstractBasePolicy):
1723 BLOCK_HINT_REGEX = re.compile("^(un)?(block-?.*)$")
1725 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
1726 super().__init__(
1727 "block",
1728 options,
1729 suite_info,
1730 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
1731 )
1732 self._blockall: dict[str | None, Hint] = {}
1734 def initialise(self, britney: "Britney") -> None:
1735 super().initialise(britney)
1736 assert self.hints is not None
1737 for hint in self.hints.search(type="block-all"):
1738 self._blockall[hint.package] = hint
1740 self._key_packages: frozenset[str] = frozenset()
1741 if "key" in self._blockall:
1742 self._key_packages = self._read_key_packages()
1744 def _read_key_packages(self) -> frozenset[str]:
1745 """Read the list of key packages
1747 The file contains data in the yaml format :
1749 - reason: <something>
1750 source: <package>
1752 The method returns a list of all key packages.
1753 """
1754 filename = os.path.join(self.state_dir, "key_packages.yaml")
1755 self.logger.info("Loading key packages from %s", filename)
1756 if os.path.exists(filename): 1756 ↛ 1761line 1756 didn't jump to line 1761 because the condition on line 1756 was always true
1757 with open(filename) as f:
1758 data = yaml.safe_load(f)
1759 key_packages = frozenset(item["source"] for item in data)
1760 else:
1761 self.logger.error(
1762 "Britney was asked to block key packages, "
1763 + "but no key_packages.yaml file was found."
1764 )
1765 sys.exit(1)
1767 return key_packages
1769 def register_hints(self, hint_parser: HintParser) -> None:
1770 # block related hints are currently defined in hint.py
1771 pass
1773 def _check_blocked(
1774 self, arch: str, version: str, excuse: "Excuse"
1775 ) -> PolicyVerdict:
1776 verdict = PolicyVerdict.PASS
1777 blocked = {}
1778 unblocked = {}
1779 block_info = {}
1780 source_suite = excuse.item.suite
1781 suite_name = source_suite.name
1782 src = excuse.item.package
1783 is_primary = source_suite.suite_class is SuiteClass.PRIMARY_SOURCE_SUITE
1785 tooltip = (
1786 f"please contact {self.options.distribution}-release if update is needed"
1787 )
1789 assert self.hints is not None
1790 mismatches = False
1791 r = self.BLOCK_HINT_REGEX
1792 for hint in self.hints.search(package=src):
1793 m = r.match(hint.type)
1794 if m:
1795 if m.group(1) == "un":
1796 assert hint.suite is not None
1797 if (
1798 hint.version != version
1799 or hint.suite.name != suite_name
1800 or (hint.architecture != arch and hint.architecture != "source")
1801 ):
1802 self.logger.info(
1803 "hint mismatch: %s %s %s", version, arch, suite_name
1804 )
1805 mismatches = True
1806 else:
1807 unblocked[m.group(2)] = hint.user
1808 excuse.add_hint(hint)
1809 else:
1810 # block(-*) hint: only accepts a source, so this will
1811 # always match
1812 blocked[m.group(2)] = hint.user
1813 excuse.add_hint(hint)
1815 if "block" not in blocked and is_primary:
1816 # if there is a specific block hint for this package, we don't
1817 # check for the general hints
1819 if self.options.distribution == "debian": 1819 ↛ 1823line 1819 didn't jump to line 1823 because the condition on line 1819 was always true
1820 url = "https://release.debian.org/testing/freeze_policy.html"
1821 tooltip = f'Follow the <a href="{url}">freeze policy</a> when applying for an unblock'
1823 if (block := self._blockall.get("source")) is not None:
1824 blocked["block"] = block.user
1825 excuse.add_hint(block)
1826 elif (
1827 block := self._blockall.get("new-source")
1828 ) is not None and src not in self.suite_info.target_suite.sources:
1829 blocked["block"] = block.user
1830 excuse.add_hint(block)
1831 # no tooltip: new sources will probably not be accepted anyway
1832 block_info["block"] = (
1833 f"blocked by {block.user}: is not in {self.suite_info.target_suite.name}"
1834 )
1835 elif (
1836 block := self._blockall.get("key")
1837 ) is not None and src in self._key_packages:
1838 blocked["block"] = block.user
1839 excuse.add_hint(block)
1840 block_info["block"] = (
1841 f"blocked by {block.user}: is a key package ({tooltip})"
1842 )
1843 elif (block := self._blockall.get("no-autopkgtest")) is not None:
1844 if excuse.autopkgtest_results == {"PASS"}:
1845 if not blocked: 1845 ↛ 1863line 1845 didn't jump to line 1863 because the condition on line 1845 was always true
1846 excuse.addinfo("not blocked: has successful autopkgtest")
1847 else:
1848 blocked["block"] = block.user
1849 excuse.add_hint(block)
1850 if not excuse.autopkgtest_results:
1851 block_info["block"] = (
1852 f"blocked by {block.user}: does not have autopkgtest ({tooltip})"
1853 )
1854 else:
1855 block_info["block"] = (
1856 f"blocked by {block.user}: autopkgtest not fully successful ({tooltip})"
1857 )
1859 elif not is_primary:
1860 blocked["block"] = suite_name
1861 excuse.needs_approval = True
1863 for block_cmd in blocked:
1864 unblock_cmd = f"un{block_cmd}"
1865 if block_cmd in unblocked:
1866 if is_primary or block_cmd == "block-udeb":
1867 excuse.addinfo(
1868 f"Ignoring {block_cmd} request by {blocked[block_cmd]}, "
1869 f"due to {unblock_cmd} request by {unblocked[block_cmd]}"
1870 )
1871 else:
1872 excuse.addinfo(f"Approved by {unblocked[block_cmd]}")
1873 else:
1874 verdict = PolicyVerdict.REJECTED_NEEDS_APPROVAL
1875 if is_primary or block_cmd == "block-udeb":
1876 # redirect people to d-i RM for udeb things:
1877 if block_cmd == "block-udeb":
1878 tooltip = "please contact the d-i release manager if an update is needed"
1879 info = block_info.get(block_cmd)
1880 if info is None:
1881 info = f"Not touching package due to {block_cmd} request by {blocked[block_cmd]} ({tooltip})"
1882 excuse.add_verdict_info(verdict, info)
1883 else:
1884 excuse.add_verdict_info(verdict, "NEEDS APPROVAL BY RM")
1885 excuse.addreason("block")
1886 if mismatches:
1887 excuse.add_detailed_info(
1888 f"Some hints for {src} do not match this item"
1889 )
1890 return verdict
1892 def apply_src_policy_impl(
1893 self,
1894 block_info: dict[str, Any],
1895 source_data_tdist: SourcePackage | None,
1896 source_data_srcdist: SourcePackage,
1897 excuse: "Excuse",
1898 ) -> PolicyVerdict:
1899 return self._check_blocked("source", source_data_srcdist.version, excuse)
1901 def apply_srcarch_policy_impl(
1902 self,
1903 block_info: dict[str, Any],
1904 arch: str,
1905 source_data_tdist: SourcePackage | None,
1906 source_data_srcdist: SourcePackage,
1907 excuse: "Excuse",
1908 ) -> PolicyVerdict:
1909 return self._check_blocked(arch, source_data_srcdist.version, excuse)
1912class BuiltOnBuilddPolicy(AbstractBasePolicy):
1914 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
1915 super().__init__(
1916 "builtonbuildd",
1917 options,
1918 suite_info,
1919 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
1920 ApplySrcPolicy.RUN_ON_EVERY_ARCH_ONLY,
1921 )
1922 self._signer_info: dict[str, Any] = {}
1924 def register_hints(self, hint_parser: HintParser) -> None:
1925 hint_parser.register_hint_type(
1926 HintType(
1927 "allow-archall-maintainer-upload",
1928 versioned=HintAnnotate.FORBIDDEN,
1929 )
1930 )
1932 def initialise(self, britney: "Britney") -> None:
1933 super().initialise(britney)
1934 try:
1935 filename_signerinfo = os.path.join(self.state_dir, "signers.json")
1936 except AttributeError as e: # pragma: no cover
1937 raise RuntimeError(
1938 "Please set STATE_DIR in the britney configuration"
1939 ) from e
1940 self._signer_info = self._read_signerinfo(filename_signerinfo)
1942 def apply_srcarch_policy_impl(
1943 self,
1944 buildd_info: dict[str, Any],
1945 arch: str,
1946 source_data_tdist: SourcePackage | None,
1947 source_data_srcdist: SourcePackage,
1948 excuse: "Excuse",
1949 ) -> PolicyVerdict:
1950 verdict = PolicyVerdict.PASS
1951 signers = self._signer_info
1953 if "signed-by" not in buildd_info:
1954 buildd_info["signed-by"] = {}
1956 item = excuse.item
1957 source_suite = item.suite
1959 # we use the source component, because a binary in contrib can
1960 # belong to a source in main
1961 component = get_component(source_data_srcdist.section)
1963 packages_s_a = source_suite.binaries[arch]
1964 assert self.hints is not None
1966 for pkg_id in sorted(
1967 x
1968 for x in filter_out_faux_gen(source_data_srcdist.binaries)
1969 if x.architecture == arch
1970 ):
1971 pkg_name = pkg_id.package_name
1972 binary_u = packages_s_a[pkg_name]
1973 pkg_arch = binary_u.architecture
1975 if binary_u.source_version != source_data_srcdist.version: 1975 ↛ 1976line 1975 didn't jump to line 1976 because the condition on line 1975 was never true
1976 continue
1978 if item.architecture != "source" and pkg_arch == "all":
1979 # we don't care about the existing arch: all binaries when
1980 # checking a binNMU item, because the arch: all binaries won't
1981 # migrate anyway
1982 continue
1984 signer = None
1985 uid = None
1986 uidinfo = ""
1987 buildd_ok = False
1988 failure_verdict = PolicyVerdict.REJECTED_PERMANENTLY
1989 try:
1990 signer = signers[pkg_name][pkg_id.version][pkg_arch]
1991 if signer["buildd"]:
1992 buildd_ok = True
1993 uid = signer["uid"]
1994 uidinfo = f"arch {pkg_arch} binaries uploaded by {uid}"
1995 except KeyError:
1996 self.logger.info(
1997 "signer info for %s %s (%s) on %s not found",
1998 pkg_name,
1999 binary_u.version,
2000 pkg_arch,
2001 arch,
2002 )
2003 uidinfo = f"upload info for arch {pkg_arch} binaries not found"
2004 failure_verdict = PolicyVerdict.REJECTED_CANNOT_DETERMINE_IF_PERMANENT
2005 if not buildd_ok:
2006 if component != "main":
2007 if pkg_arch not in buildd_info["signed-by"]: 2007 ↛ 2011line 2007 didn't jump to line 2011 because the condition on line 2007 was always true
2008 excuse.add_detailed_info(
2009 f"{uidinfo}, but package in {component}"
2010 )
2011 buildd_ok = True
2012 elif pkg_arch == "all":
2013 if (
2014 allow_hint := self.hints.search_first(
2015 "allow-archall-maintainer-upload", package=item.package
2016 )
2017 ) is not None:
2018 buildd_ok = True
2019 verdict = PolicyVerdict.worst_of(
2020 verdict, PolicyVerdict.PASS_HINTED
2021 )
2022 if pkg_arch not in buildd_info["signed-by"]:
2023 excuse.addinfo(
2024 f"{uidinfo}, but whitelisted by {allow_hint.user}"
2025 )
2026 if not buildd_ok:
2027 verdict = failure_verdict
2028 if pkg_arch not in buildd_info["signed-by"]:
2029 if pkg_arch == "all":
2030 uidinfo += (
2031 ", a new source-only upload is needed to allow migration"
2032 )
2033 excuse.add_verdict_info(verdict, f"Not built on buildd: {uidinfo}")
2035 if ( 2035 ↛ 2038line 2035 didn't jump to line 2038 because the condition on line 2035 was never true
2036 data := buildd_info["signed-by"].get(pkg_arch)
2037 ) is not None and data != uid:
2038 self.logger.info(
2039 "signer mismatch for %s (%s %s) on %s: %s, while %s already listed",
2040 pkg_name,
2041 binary_u.source,
2042 binary_u.source_version,
2043 pkg_arch,
2044 uid,
2045 data,
2046 )
2048 buildd_info["signed-by"][pkg_arch] = uid
2050 return verdict
2052 def _read_signerinfo(self, filename: str) -> dict[str, Any]:
2053 signerinfo: dict[str, Any] = {}
2054 self.logger.info("Loading signer info from %s", filename)
2055 with open(filename) as fd: 2055 ↛ exitline 2055 didn't return from function '_read_signerinfo' because the return on line 2057 wasn't executed
2056 if os.fstat(fd.fileno()).st_size < 1: 2056 ↛ 2057line 2056 didn't jump to line 2057 because the condition on line 2056 was never true
2057 return signerinfo
2058 signerinfo = json.load(fd)
2060 return signerinfo
2063class ImplicitDependencyPolicy(AbstractBasePolicy):
2064 """Implicit Dependency policy
2066 Upgrading a package pkg-a can break the installability of a package pkg-b.
2067 A newer version (or the removal) of pkg-b might fix the issue. In that
2068 case, pkg-a has an 'implicit dependency' on pkg-b, because pkg-a can only
2069 migrate if pkg-b also migrates.
2071 This policy tries to discover a few common cases, and adds the relevant
2072 info to the excuses. If another item is needed to fix the
2073 uninstallability, a dependency is added. If no newer item can fix it, this
2074 excuse will be blocked.
2076 Note that the migration step will check the installability of every
2077 package, so this policy doesn't need to handle every corner case. It
2078 must, however, make sure that no excuse is unnecessarily blocked.
2080 Some cases that should be detected by this policy:
2082 * pkg-a is upgraded from 1.0-1 to 2.0-1, while
2083 pkg-b has "Depends: pkg-a (<< 2.0)"
2084 This typically happens if pkg-b has a strict dependency on pkg-a because
2085 it uses some non-stable internal interface (examples are glibc,
2086 binutils, python3-defaults, ...)
2088 * pkg-a is upgraded from 1.0-1 to 2.0-1, and
2089 pkg-a 1.0-1 has "Provides: provides-1",
2090 pkg-a 2.0-1 has "Provides: provides-2",
2091 pkg-b has "Depends: provides-1"
2092 This typically happens when pkg-a has an interface that changes between
2093 versions, and a virtual package is used to identify the version of this
2094 interface (e.g. perl-api-x.y)
2096 """
2098 _pkg_universe: "BinaryPackageUniverse"
2099 _all_binaries: dict["BinaryPackageId", "BinaryPackage"]
2100 _allow_uninst: dict[str, set[str | None]]
2101 _nobreakall_arches: list[str]
2103 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
2104 super().__init__(
2105 "implicit-deps",
2106 options,
2107 suite_info,
2108 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
2109 ApplySrcPolicy.RUN_ON_EVERY_ARCH_ONLY,
2110 )
2112 def initialise(self, britney: "Britney") -> None:
2113 super().initialise(britney)
2114 self._pkg_universe = britney.pkg_universe
2115 self._all_binaries = britney.all_binaries
2116 self._smooth_updates = britney.options.smooth_updates
2117 self._nobreakall_arches = self.options.nobreakall_arches
2118 self._new_arches = self.options.new_arches
2119 self._break_arches = self.options.break_arches
2120 self._allow_uninst = britney.allow_uninst
2121 self._outofsync_arches = self.options.outofsync_arches
2123 def can_be_removed(self, pkg: BinaryPackage) -> bool:
2124 src = pkg.source
2125 target_suite = self.suite_info.target_suite
2127 # TODO these conditions shouldn't be hardcoded here
2128 # ideally, we would be able to look up excuses to see if the removal
2129 # is in there, but in the current flow, this policy is called before
2130 # all possible excuses exist, so there is no list for us to check
2132 if src not in self.suite_info.primary_source_suite.sources:
2133 # source for pkg not in unstable: candidate for removal
2134 return True
2136 source_t = target_suite.sources[src]
2137 assert self.hints is not None
2138 if self.hints.has_hint("remove", package=src, version=source_t.version):
2139 # removal hint for the source in testing: candidate for removal
2140 return True
2142 if target_suite.is_cruft(pkg):
2143 # if pkg is cruft in testing, removal will be tried
2144 return True
2146 # the case were the newer version of the source no longer includes the
2147 # binary (or includes a cruft version of the binary) will be handled
2148 # separately (in that case there might be an implicit dependency on
2149 # the newer source)
2151 return False
2153 def should_skip_rdep(
2154 self, pkg: BinaryPackage, source_name: str, myarch: str
2155 ) -> bool:
2156 target_suite = self.suite_info.target_suite
2158 if not target_suite.is_pkg_in_the_suite(pkg.pkg_id):
2159 # it is not in the target suite, migration cannot break anything
2160 return True
2162 if pkg.source == source_name:
2163 # if it is built from the same source, it will be upgraded
2164 # with the source
2165 return True
2167 if self.can_be_removed(pkg):
2168 # could potentially be removed, so if that happens, it won't be
2169 # broken
2170 return True
2172 if pkg.architecture == "all" and myarch not in self._nobreakall_arches:
2173 # arch all on non nobreakarch is allowed to become uninstallable
2174 return True
2176 if pkg.pkg_id.package_name in self._allow_uninst[myarch]:
2177 # there is a hint to allow this binary to become uninstallable
2178 return True
2180 if not target_suite.is_installable(pkg.pkg_id):
2181 # it is already uninstallable in the target suite, migration
2182 # cannot break anything
2183 return True
2185 return False
2187 def breaks_installability(
2188 self,
2189 pkg_id_t: BinaryPackageId,
2190 pkg_id_s: BinaryPackageId | None,
2191 pkg_to_check: BinaryPackageId,
2192 ) -> bool:
2193 """
2194 Check if upgrading pkg_id_t to pkg_id_s breaks the installability of
2195 pkg_to_check.
2197 To check if removing pkg_id_t breaks pkg_to_check, set pkg_id_s to
2198 None.
2199 """
2201 pkg_universe = self._pkg_universe
2202 negative_deps = pkg_universe.negative_dependencies_of(pkg_to_check)
2204 for dep in pkg_universe.dependencies_of(pkg_to_check):
2205 if pkg_id_t not in dep:
2206 # this depends doesn't have pkg_id_t as alternative, so
2207 # upgrading pkg_id_t cannot break this dependency clause
2208 continue
2210 # We check all the alternatives for this dependency, to find one
2211 # that can satisfy it when pkg_id_t is upgraded to pkg_id_s
2212 found_alternative = False
2213 for d in dep:
2214 if d in negative_deps:
2215 # If this alternative dependency conflicts with
2216 # pkg_to_check, it cannot be used to satisfy the
2217 # dependency.
2218 # This commonly happens when breaks are added to pkg_id_s.
2219 continue
2221 if d.package_name != pkg_id_t.package_name:
2222 # a binary different from pkg_id_t can satisfy the dep, so
2223 # upgrading pkg_id_t won't break this dependency
2224 found_alternative = True
2225 break
2227 if d != pkg_id_s:
2228 # We want to know the impact of the upgrade of
2229 # pkg_id_t to pkg_id_s. If pkg_id_s migrates to the
2230 # target suite, any other version of this binary will
2231 # not be there, so it cannot satisfy this dependency.
2232 # This includes pkg_id_t, but also other versions.
2233 continue
2235 # pkg_id_s can satisfy the dep
2236 found_alternative = True
2237 break
2239 if not found_alternative:
2240 return True
2241 return False
2243 def check_upgrade(
2244 self,
2245 pkg_id_t: BinaryPackageId,
2246 pkg_id_s: BinaryPackageId | None,
2247 source_name: str,
2248 myarch: str,
2249 broken_binaries: set[str],
2250 excuse: "Excuse",
2251 ) -> PolicyVerdict:
2252 verdict = PolicyVerdict.PASS
2254 pkg_universe = self._pkg_universe
2255 all_binaries = self._all_binaries
2257 # check all rdeps of the package in testing
2258 rdeps_t = pkg_universe.reverse_dependencies_of(pkg_id_t)
2260 for rdep_pkg in sorted(rdeps_t):
2261 rdep_p = all_binaries[rdep_pkg]
2263 # check some cases where the rdep won't become uninstallable, or
2264 # where we don't care if it does
2265 if self.should_skip_rdep(rdep_p, source_name, myarch):
2266 continue
2268 if not self.breaks_installability(pkg_id_t, pkg_id_s, rdep_pkg):
2269 # if upgrading pkg_id_t to pkg_id_s doesn't break rdep_pkg,
2270 # there is no implicit dependency
2271 continue
2273 # The upgrade breaks the installability of the rdep. We need to
2274 # find out if there is a newer version of the rdep that solves the
2275 # uninstallability. If that is the case, there is an implicit
2276 # dependency. If not, the upgrade will fail.
2278 # check source versions
2279 good_newer_versions = set()
2280 for npkg, suite in find_newer_binaries(
2281 self.suite_info, rdep_p, add_source_for_dropped_bin=True
2282 ):
2283 if npkg.architecture == "source":
2284 # When a newer version of the source package doesn't have
2285 # the binary, we get the source as 'newer version'. In
2286 # this case, the binary will not be uninstallable if the
2287 # newer source migrates, because it is no longer there.
2288 good_newer_versions.add(npkg)
2289 continue
2290 assert isinstance(npkg, BinaryPackageId)
2291 if not self.breaks_installability(pkg_id_t, pkg_id_s, npkg):
2292 good_newer_versions.add(npkg)
2294 if good_newer_versions:
2295 spec = DependencySpec(DependencyType.IMPLICIT_DEPENDENCY, myarch)
2296 excuse.add_package_depends(spec, good_newer_versions)
2297 else:
2298 # no good newer versions: no possible solution
2299 broken_binaries.add(rdep_pkg.name)
2300 if pkg_id_s:
2301 action = f"migrating {pkg_id_s.name} to {self.suite_info.target_suite.name}"
2302 else:
2303 action = f"removing {pkg_id_t.name} from {self.suite_info.target_suite.name}"
2304 if rdep_pkg.package_name.endswith("-faux-build-depends"):
2305 name = rdep_pkg.package_name.removesuffix("-faux-build-depends")
2306 info = f'{action} makes Build-Depends of src:<a href="#{name}">{name}</a> uninstallable'
2307 else:
2308 info = '{0} makes <a href="#{1}">{1}</a> uninstallable'.format(
2309 action, rdep_pkg.name
2310 )
2311 verdict = PolicyVerdict.REJECTED_PERMANENTLY
2312 excuse.add_verdict_info(verdict, info)
2314 return verdict
2316 def apply_srcarch_policy_impl(
2317 self,
2318 implicit_dep_info: dict[str, Any],
2319 arch: str,
2320 source_data_tdist: SourcePackage | None,
2321 source_data_srcdist: SourcePackage,
2322 excuse: "Excuse",
2323 ) -> PolicyVerdict:
2324 verdict = PolicyVerdict.PASS
2326 if not source_data_tdist:
2327 # this item is not currently in testing: no implicit dependency
2328 return verdict
2330 if excuse.hasreason("missingbuild"):
2331 # if the build is missing, the policy would treat this as if the
2332 # binaries would be removed, which would give incorrect (and
2333 # confusing) info
2334 info = f"missing build, not checking implicit dependencies on {arch}"
2335 excuse.add_detailed_info(info)
2336 return verdict
2338 source_suite = excuse.item.suite
2339 source_name = excuse.item.package
2340 target_suite = self.suite_info.target_suite
2341 all_binaries = self._all_binaries
2343 # we check all binaries for this excuse that are currently in testing
2344 relevant_binaries = sorted(
2345 x
2346 for x in source_data_tdist.binaries
2347 if (arch == "source" or x.architecture == arch)
2348 and x.package_name in target_suite.binaries[x.architecture]
2349 and x.architecture not in self._new_arches
2350 and x.architecture not in self._break_arches
2351 and x.architecture not in self._outofsync_arches
2352 )
2354 broken_binaries: set[str] = set()
2356 assert self.hints is not None
2357 for pkg_id_t in relevant_binaries:
2358 mypkg = pkg_id_t.package_name
2359 myarch = pkg_id_t.architecture
2360 binaries_t_a = target_suite.binaries[myarch]
2361 binaries_s_a = source_suite.binaries[myarch]
2363 if target_suite.is_cruft(all_binaries[pkg_id_t]):
2364 # this binary is cruft in testing: it will stay around as long
2365 # as necessary to satisfy dependencies, so we don't need to
2366 # care
2367 continue
2369 if (mybin := binaries_s_a.get(mypkg)) is not None:
2370 pkg_id_s: Optional["BinaryPackageId"] = mybin.pkg_id
2371 if mybin.source != source_name:
2372 # hijack: this is too complicated to check, so we ignore
2373 # it (the migration code will check the installability
2374 # later anyway)
2375 pass
2376 elif mybin.source_version != source_data_srcdist.version:
2377 # cruft in source suite: pretend the binary doesn't exist
2378 pkg_id_s = None
2379 elif pkg_id_t == pkg_id_s:
2380 # same binary (probably arch: all from a binNMU):
2381 # 'upgrading' doesn't change anything, for this binary, so
2382 # it won't break anything
2383 continue
2384 else:
2385 pkg_id_s = None
2387 if not pkg_id_s and is_smooth_update_allowed(
2388 binaries_t_a[mypkg], self._smooth_updates, self.hints
2389 ):
2390 # the binary isn't in the new version (or is cruft there), and
2391 # smooth updates are allowed: the binary can stay around if
2392 # that is necessary to satisfy dependencies, so we don't need
2393 # to check it
2394 continue
2396 if (
2397 not pkg_id_s
2398 and source_data_tdist.version == source_data_srcdist.version
2399 and source_suite.suite_class is SuiteClass.ADDITIONAL_SOURCE_SUITE
2400 and binaries_t_a[mypkg].architecture == "all"
2401 ):
2402 # we're very probably migrating a binNMU built in tpu where the arch:all
2403 # binaries were not copied to it as that's not needed. This policy could
2404 # needlessly block.
2405 continue
2407 v = self.check_upgrade(
2408 pkg_id_t, pkg_id_s, source_name, myarch, broken_binaries, excuse
2409 )
2410 verdict = PolicyVerdict.worst_of(verdict, v)
2412 # each arch is processed separately, so if we already have info from
2413 # other archs, we need to merge the info from this arch
2414 broken_old = set(implicit_dep_info.get("broken-binaries", []))
2415 implicit_dep_info["broken-binaries"] = sorted(broken_old | broken_binaries)
2417 return verdict
2420class ReverseRemovalPolicy(AbstractBasePolicy):
2421 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
2422 super().__init__(
2423 "reverseremoval",
2424 options,
2425 suite_info,
2426 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
2427 )
2429 def register_hints(self, hint_parser: HintParser) -> None:
2430 hint_parser.register_hint_type(HintType("ignore-reverse-remove"))
2432 def initialise(self, britney: "Britney") -> None:
2433 super().initialise(britney)
2435 pkg_universe = britney.pkg_universe
2436 source_suites = britney.suite_info.source_suites
2437 target_suite = britney.suite_info.target_suite
2439 # Build set of the sources of reverse (Build-) Depends
2440 assert self.hints is not None
2442 rev_bin: dict[BinaryPackageId, set[str]] = defaultdict(set)
2443 for hint in self.hints.search("remove"):
2444 for item in hint.packages:
2445 # I think we don't need to look at the target suite
2446 for src_suite in source_suites:
2447 try:
2448 # Explicitly not running filter_out_faux here
2449 my_bins = set(src_suite.sources[item.uvname].binaries)
2450 except KeyError:
2451 continue
2452 compute_reverse_tree(pkg_universe, my_bins)
2453 for this_bin in my_bins:
2454 rev_bin.setdefault(this_bin, set()).add(item.uvname)
2456 rev_src: dict[str, set[str]] = defaultdict(set)
2457 for bin_pkg, reasons in rev_bin.items():
2458 # If the pkg is in the target suite, there's nothing this
2459 # policy wants to do.
2460 if target_suite.is_pkg_in_the_suite(bin_pkg):
2461 continue
2462 that_bin = britney.all_binaries[bin_pkg]
2463 bin_src = that_bin.source + "/" + that_bin.source_version
2464 rev_src.setdefault(bin_src, set()).update(reasons)
2465 self._block_src_for_rm_hint = rev_src
2467 def apply_src_policy_impl(
2468 self,
2469 rev_remove_info: dict[str, Any],
2470 source_data_tdist: SourcePackage | None,
2471 source_data_srcdist: SourcePackage,
2472 excuse: "Excuse",
2473 ) -> PolicyVerdict:
2474 verdict = PolicyVerdict.PASS
2476 item = excuse.item
2477 if item.name in self._block_src_for_rm_hint:
2478 reason = ", ".join(sorted(self._block_src_for_rm_hint[item.name]))
2479 assert self.hints is not None
2480 excuse.addreason("reverseremoval")
2481 if (
2482 ignore_hint := self.hints.search_first(
2483 "ignore-reverse-remove", package=item.uvname, version=item.version
2484 )
2485 ) is not None:
2486 excuse.addreason("ignore-reverse-remove")
2487 excuse.addinfo(
2488 f"Should block migration because of remove hint for {reason}, but forced by {ignore_hint.user}"
2489 )
2490 verdict = PolicyVerdict.PASS_HINTED
2491 else:
2492 verdict = PolicyVerdict.REJECTED_PERMANENTLY
2493 excuse.add_verdict_info(
2494 verdict, f"Remove hint for (transitive) dependency: {reason}"
2495 )
2497 return verdict
2500class ReproducibleState(Enum):
2501 BAD = auto()
2502 FAIL = auto()
2503 GOOD = auto()
2504 UNKNOWN = auto()
2506 @staticmethod
2507 def from_str(val: str | None) -> "ReproducibleState":
2508 match val:
2509 case "BAD":
2510 return ReproducibleState.BAD
2511 case "FAIL": 2511 ↛ 2512line 2511 didn't jump to line 2512 because the pattern on line 2511 never matched
2512 return ReproducibleState.FAIL
2513 case "GOOD":
2514 return ReproducibleState.GOOD
2515 case "UNKNOWN" | None: 2515 ↛ 2517line 2515 didn't jump to line 2517 because the pattern on line 2515 always matched
2516 return ReproducibleState.UNKNOWN
2517 case _:
2518 raise ValueError(f"Invalid reproducability state f{str}")
2521@dataclass(slots=True, frozen=True)
2522class ReproducibleData:
2523 state: ReproducibleState
2524 build_id: str | None = field(default=None, kw_only=True)
2525 diffoscope_log_id: str | None = field(default=None, kw_only=True)
2526 artifact_id: str | None = field(default=None, kw_only=True)
2529class ReproduciblePolicy(AbstractBasePolicy):
2530 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
2531 super().__init__(
2532 "reproducible",
2533 options,
2534 suite_info,
2535 {SuiteClass.PRIMARY_SOURCE_SUITE},
2536 ApplySrcPolicy.RUN_ON_EVERY_ARCH_ONLY,
2537 )
2538 self._reproducible: dict[str, dict[tuple[str, str], ReproducibleData]] = {}
2539 self._components: tuple[str, ...] = ()
2541 # Default values for this policy's options
2542 parse_option(options, "repro_success_bounty", default=0, to_int=True)
2543 parse_option(options, "repro_regression_penalty", default=0, to_int=True)
2544 parse_option(options, "repro_log_url")
2545 parse_option(options, "repro_excuse_url")
2546 parse_option(options, "repro_retry_url")
2547 parse_option(options, "repro_components")
2549 def register_hints(self, hint_parser: HintParser) -> None:
2550 hint_parser.register_hint_type(
2551 HintType(
2552 "ignore-reproducible-src",
2553 versioned=HintAnnotate.OPTIONAL,
2554 architectured=HintAnnotate.OPTIONAL,
2555 )
2556 )
2557 hint_parser.register_hint_type(
2558 HintType(
2559 "ignore-reproducible",
2560 versioned=HintAnnotate.OPTIONAL,
2561 architectured=HintAnnotate.OPTIONAL,
2562 )
2563 )
2565 def initialise(self, britney: "Britney") -> None:
2566 super().initialise(britney)
2567 summary = self._reproducible
2569 valid_release_names = {
2570 suite.codename
2571 for suite in chain(
2572 (britney.suite_info.target_suite,),
2573 britney.suite_info.source_suites,
2574 )
2575 } | {
2576 suite.name
2577 for suite in chain(
2578 (britney.suite_info.target_suite,),
2579 britney.suite_info.source_suites,
2580 )
2581 }
2583 assert hasattr(
2584 self, "state_dir"
2585 ), "Please set STATE_DIR in the britney configuration"
2586 assert (
2587 self.options.repro_components
2588 ), "Please set REPRO_COMPONENTS in the britney configuration"
2589 self._components = tuple(self.options.repro_components.split())
2591 for file in os.listdir(self.state_dir):
2592 if not file.startswith("reproducible-") or not file.endswith(".json"):
2593 continue
2594 filename = os.path.join(self.state_dir, file)
2596 self.logger.info("Loading reproducibility report from %s", filename)
2597 with open(filename) as fd:
2598 if os.fstat(fd.fileno()).st_size < 1:
2599 continue
2600 data = json.load(fd)
2602 for result in data["records"]:
2603 if ( 2603 ↛ 2607line 2603 didn't jump to line 2607 because the condition on line 2603 was never true
2604 release := result.get("release")
2605 ) is not None and release not in valid_release_names:
2606 # tests do not have a release set
2607 continue
2609 state = ReproducibleState.from_str(result.get("status"))
2610 repo = {
2611 key: value
2612 for key, value in result.items()
2613 if key in ("build_id", "diffoscope_log_id", "artifact_id")
2614 }
2616 summary.setdefault(result["architecture"], {})[
2617 (result["name"], result["version"])
2618 ] = ReproducibleData(state, **repo)
2620 def _lookup_data(
2621 self, package_name: str, version: str, arch: str
2622 ) -> tuple[ReproducibleData, str] | None:
2623 key = (package_name, version)
2624 if (repo := self._reproducible[arch].get(key)) is not None:
2625 return repo, arch
2627 repo = self._reproducible["all"].get(key)
2628 return (repo, "all") if repo is not None else None
2630 def _format_link(self, bpid: BinaryPackageId, arch: str) -> str:
2631 data = self._lookup_data(bpid.package_name, bpid.version, arch)
2632 assert data is not None
2633 repo, arch = data
2634 if repo.diffoscope_log_id and (diff_id := repo.artifact_id): 2634 ↛ 2635line 2634 didn't jump to line 2635 because the condition on line 2634 was never true
2635 endpoint = f"artifacts/{diff_id}/diffoscope"
2636 else:
2637 endpoint = "log"
2638 url = self.options.repro_log_url.format(
2639 arch=arch, build_id=repo.build_id, endpoint=endpoint
2640 )
2641 return f'<a href="{url}">{bpid.package_name}</a>'
2643 def _create_link_to_log(self, arch: str, failed_bpids: set[BinaryPackageId]) -> str:
2644 if not self.options.repro_log_url: 2644 ↛ 2645line 2644 didn't jump to line 2645 because the condition on line 2644 was never true
2645 return ": " + ", ".join(bpid.package_name for bpid in sorted(failed_bpids))
2647 return ": " + ", ".join(
2648 self._format_link(bpid, arch) for bpid in sorted(failed_bpids)
2649 )
2651 def apply_srcarch_policy_impl(
2652 self,
2653 policy_info: dict[str, Any],
2654 arch: str,
2655 source_data_tdist: SourcePackage | None,
2656 source_data_srcdist: SourcePackage,
2657 excuse: "Excuse",
2658 ) -> PolicyVerdict:
2659 verdict = PolicyVerdict.PASS
2660 eligible_for_bounty = False
2661 all_hints = []
2663 assert self.hints is not None # Needed for type checking / mypy
2665 # we don't want to apply this policy (yet) on binNMUs
2666 if excuse.item.architecture != "source": 2666 ↛ 2667line 2666 didn't jump to line 2667 because the condition on line 2666 was never true
2667 return verdict
2669 # we're not supposed to judge on this arch
2670 if arch not in self.options.repro_arches: 2670 ↛ 2671line 2670 didn't jump to line 2671 because the condition on line 2670 was never true
2671 return verdict
2673 # bail out if this arch has no packages for this source (not build
2674 # here)
2675 if arch not in excuse.packages: 2675 ↛ 2676line 2675 didn't jump to line 2676 because the condition on line 2675 was never true
2676 return verdict
2678 component = get_component(source_data_srcdist.section)
2680 if self._components and component not in self._components: 2680 ↛ 2681line 2680 didn't jump to line 2681 because the condition on line 2680 was never true
2681 self.logger.debug(
2682 "%s skipping reproducible policy, component %s not requested",
2683 excuse.name,
2684 component,
2685 )
2686 return verdict
2688 source_name = excuse.item.package
2690 if self.options.repro_excuse_url:
2691 url = self.options.repro_excuse_url.format(
2692 package=quote(source_name), arch=arch
2693 )
2694 url_html = ' - <a href="%s">info</a>' % url
2695 # When run on multiple archs, the last one "wins"
2696 policy_info["status-url"] = url
2697 else:
2698 url = None
2699 url_html = ""
2701 if arch not in self._reproducible: 2701 ↛ 2702line 2701 didn't jump to line 2702 because the condition on line 2701 was never true
2702 verdict = PolicyVerdict.REJECTED_TEMPORARILY
2703 msg = f"No reproducibility data available at all for {arch}"
2704 excuse.add_verdict_info(verdict, msg)
2705 return verdict
2706 if "all" not in self._reproducible: 2706 ↛ 2707line 2706 didn't jump to line 2707 because the condition on line 2706 was never true
2707 verdict = PolicyVerdict.REJECTED_TEMPORARILY
2708 msg = "No reproducibility data available at all for arch:all"
2709 excuse.add_verdict_info(verdict, msg)
2710 return verdict
2712 # skip/delay policy until both arch:arch and arch:all builds are done
2713 if (arch or "all") in excuse.missing_builds: 2713 ↛ 2714line 2713 didn't jump to line 2714 because the condition on line 2713 was never true
2714 self.logger.debug(
2715 "%s not built for %s or all, skipping reproducible policy",
2716 excuse.name,
2717 arch,
2718 )
2719 verdict = PolicyVerdict.REJECTED_TEMPORARILY
2720 excuse.add_verdict_info(
2721 verdict,
2722 f"Reproducibility check deferred on {arch}: missing builds{url_html}",
2723 )
2724 return verdict
2726 source_suite_state = "not-unknown"
2727 failed_bpids: set[BinaryPackageId] = set()
2728 # The states should either be GOOD/BAD for all binaries, UNKNOWN for all
2729 # binaries, or missing for all binaries, but let's not assume that.
2730 # They can be from different components after all.
2731 bins_src, src_suite_name = binaries_from_source_version(
2732 source_data_srcdist, self.suite_info
2733 )
2734 for bpid in bins_src:
2735 if bpid.architecture not in ("all", arch): 2735 ↛ 2736line 2735 didn't jump to line 2736 because the condition on line 2735 was never true
2736 continue
2737 in_component = True
2738 for suite in self.suite_info.source_suites:
2739 if suite.name == src_suite_name and (
2740 (
2741 component := get_component(
2742 suite.all_binaries_in_suite[bpid].section
2743 )
2744 )
2745 not in self._components
2746 ):
2747 self.logger.debug(
2748 "repro check for %s skipped due to component %s",
2749 bpid,
2750 component,
2751 )
2752 in_component = False
2753 break
2754 if not in_component:
2755 # TODO: should we update the excuses text?
2756 continue
2758 if (
2759 data := self._lookup_data(bpid.package_name, bpid.version, arch)
2760 ) is not None:
2761 pkg_info, _ = data
2762 self.logger.debug("repro data for %s: %s", bpid, pkg_info.state)
2763 if pkg_info.state is ReproducibleState.BAD:
2764 failed_bpids.add(bpid)
2765 # not changing source_suite_state here on purpose
2766 elif ( 2766 ↛ 2770line 2766 didn't jump to line 2770
2767 pkg_info.state is ReproducibleState.FAIL
2768 or pkg_info.state is ReproducibleState.UNKNOWN
2769 ):
2770 source_suite_state = "unknown"
2771 else:
2772 self.logger.debug("No repro data found for %s", bpid)
2773 # but maybe it's hinted (e.g. at the time of writing
2774 # reproduce.debian.net has a bug where udebs go missing)
2775 if (
2776 bpid_hints := self.hints.search_first(
2777 "ignore-reproducible",
2778 package=bpid.package_name,
2779 version=bpid.version,
2780 architecture=bpid.architecture,
2781 )
2782 ) is not None:
2783 all_hints.append(bpid_hints)
2784 self.logger.debug("repro: hint found for %s: %s", source_name, bpid)
2785 else:
2786 source_suite_state = "unknown"
2787 break
2789 if source_suite_state == "not-unknown":
2790 source_suite_state = "known"
2792 excuse_info = []
2793 if source_suite_state == "unknown":
2794 verdict = PolicyVerdict.REJECTED_TEMPORARILY
2795 excuse_info.append(
2796 f"Reproducibility check waiting for results on {arch}{url_html}"
2797 )
2798 policy_info.setdefault("state", {}).setdefault(arch, "unavailable")
2799 elif failed_bpids:
2800 ignored_bpids: set[BinaryPackageId] = set()
2801 if source_data_tdist is None: 2801 ↛ 2802line 2801 didn't jump to line 2802 because the condition on line 2801 was never true
2802 target_suite_state = "new"
2803 else:
2804 target_suite_state = "reproducible"
2805 for bpid in failed_bpids:
2806 pkg_name = bpid.package_name
2807 for bpid_t in filter_out_faux_gen(source_data_tdist.binaries):
2808 if bpid_t.architecture not in ("all", arch): 2808 ↛ 2809line 2808 didn't jump to line 2809 because the condition on line 2808 was never true
2809 continue
2810 if pkg_name != bpid_t.package_name: 2810 ↛ 2811line 2810 didn't jump to line 2811 because the condition on line 2810 was never true
2811 continue
2812 if ( 2812 ↛ 2827line 2812 didn't jump to line 2827 because the condition on line 2812 was always true
2813 data := self._lookup_data(pkg_name, bpid_t.version, arch)
2814 ) is not None:
2815 pkg_info, _ = data
2816 self.logger.debug(
2817 "testing repro data for %s: %s", bpid_t, pkg_info.state
2818 )
2819 if pkg_info.state is ReproducibleState.BAD:
2820 ignored_bpids.add(bpid)
2821 elif ( 2821 ↛ 2825line 2821 didn't jump to line 2825
2822 pkg_info.state is ReproducibleState.FAIL
2823 or pkg_info.state is ReproducibleState.UNKNOWN
2824 ):
2825 target_suite_state = "unknown"
2826 else:
2827 self.logger.debug(
2828 "No testing repro data found for %s", bpid_t
2829 )
2830 # This shouldn't happen as for the past migration
2831 # to have been allowed, there should be data.
2832 target_suite_state = "unknown"
2833 break
2835 # Reminder: code here is part of the non-reproducibile source-suite branch
2836 if target_suite_state == "new": 2836 ↛ 2837line 2836 didn't jump to line 2837 because the condition on line 2836 was never true
2837 verdict = PolicyVerdict.REJECTED_PERMANENTLY
2838 excuse_info.append(
2839 f"New but not reproduced on {arch}{url_html}"
2840 f"{self._create_link_to_log(arch, failed_bpids)}"
2841 )
2842 policy_info.setdefault("state", {}).setdefault(
2843 arch, "new but not reproducible"
2844 )
2845 elif target_suite_state == "unknown": 2845 ↛ 2847line 2845 didn't jump to line 2847 because the condition on line 2845 was never true
2846 # Shouldn't happen after initial bootstrap once blocking
2847 verdict = PolicyVerdict.REJECTED_TEMPORARILY
2848 excuse_info.append(
2849 f"Reproducibility check failed and now waiting for reference "
2850 f"results on {arch}{url_html}"
2851 f"{self._create_link_to_log(arch, failed_bpids)}"
2852 )
2853 policy_info.setdefault("state", {}).setdefault(
2854 arch, "waiting for reference"
2855 )
2856 elif failed_bpids <= ignored_bpids:
2857 # For the forseeable future we want to prevent regressions, one day
2858 # we might want to even block these.
2859 # verdict = PolicyVerdict.REJECTED_PERMANENTLY
2860 excuse_info.append(
2861 f"Not reproduced on {arch} (not a regression)"
2862 f"{self._create_link_to_log(arch, failed_bpids)}"
2863 )
2864 policy_info.setdefault("state", {}).setdefault(arch, "not reproducible")
2865 else:
2866 verdict = PolicyVerdict.REJECTED_PERMANENTLY
2867 excuse_info.append(
2868 f"Reproducibility regression on {arch}"
2869 f"{self._create_link_to_log(arch, failed_bpids - ignored_bpids)}"
2870 )
2871 policy_info.setdefault("state", {}).setdefault(arch, "regression")
2873 # non-reproducible source-suite cases are handled above, so here we
2874 # handle the last of the source-suite cases
2875 else:
2876 excuse_info.append(f"Reproduced on {arch}{url_html}")
2877 policy_info.setdefault("state", {}).setdefault(arch, "reproducible")
2878 eligible_for_bounty = True
2880 if verdict.is_rejected:
2881 for hint_arch in ("source", arch):
2882 if (
2883 ignore_hint := self.hints.search_first(
2884 "ignore-reproducible-src",
2885 package=source_name,
2886 version=source_data_srcdist.version,
2887 architecture=hint_arch,
2888 )
2889 ) is not None:
2890 # one hint is enough, take the first one encountered
2891 verdict = PolicyVerdict.PASS_HINTED
2892 policy_info.setdefault("hints", {}).setdefault(arch, []).append(
2893 f"{ignore_hint.user}: {str(ignore_hint)}"
2894 )
2895 if hint_arch == arch: 2895 ↛ 2898line 2895 didn't jump to line 2898 because the condition on line 2895 was always true
2896 on_arch = f" on {arch}"
2897 else:
2898 on_arch = ""
2899 excuse_info.append(
2900 f"Reproducibility issues ignored for src:{ignore_hint.package}"
2901 f"{on_arch} as requested by {ignore_hint.user}"
2902 )
2903 break
2905 if verdict.is_rejected:
2906 if source_suite_state == "known":
2907 check_bpids = failed_bpids - ignored_bpids
2908 else:
2909 # Let's not wait for results if all binaries have a hint
2910 check_bpids = filter_out_faux(source_data_srcdist.binaries)
2911 missed_bpids = set()
2913 for bpid in check_bpids:
2914 if (
2915 bpid_hint := self.hints.search_first(
2916 "ignore-reproducible",
2917 package=bpid.package_name,
2918 version=bpid.version,
2919 architecture=bpid.architecture,
2920 )
2921 ) is not None:
2922 # one hint per binary is enough
2923 all_hints.append(bpid_hint)
2924 self.logger.debug(
2925 "repro: hint found for %s: %s", source_name, bpid
2926 )
2927 else:
2928 missed_bpids.add(bpid)
2930 if not missed_bpids:
2931 verdict = PolicyVerdict.PASS_HINTED
2932 for hint in all_hints:
2933 policy_info.setdefault("hints", {}).setdefault(arch, []).append(
2934 hint.user + ": " + str(hint)
2935 )
2936 # TODO: we're going to print this for arch:all binaries on each arch
2937 excuse_info.append(
2938 f"Reproducibility issues ignored for {hint.package} on {arch} as "
2939 f"requested by {hint.user}"
2940 )
2941 elif all_hints: 2941 ↛ 2942line 2941 didn't jump to line 2942 because the condition on line 2941 was never true
2942 self.logger.info(
2943 "repro: binary hints for %s ignored as they don't cover these binaries %s",
2944 source_name,
2945 missed_bpids,
2946 )
2948 # A binary without results got hinted
2949 if not verdict.is_rejected and all_hints:
2950 for hint in all_hints:
2951 excuse_info.append(
2952 f"Reproducibility unknown for {hint.package} but ignored on {arch} as "
2953 f"requested by {hint.user}"
2954 )
2956 if self.options.repro_success_bounty and eligible_for_bounty: 2956 ↛ 2957line 2956 didn't jump to line 2957 because the condition on line 2956 was never true
2957 excuse.add_bounty("reproducibility", self.options.repro_success_bounty)
2959 if verdict.is_rejected and self.options.repro_regression_penalty: 2959 ↛ 2961line 2959 didn't jump to line 2961 because the condition on line 2959 was never true
2960 # With a non-zero penalty, we shouldn't block on this policy
2961 verdict = PolicyVerdict.PASS
2962 if self.options.repro_regression_penalty > 0:
2963 excuse.add_penalty(
2964 "reproducibility", self.options.repro_regression_penalty
2965 )
2967 for msg in excuse_info:
2968 if verdict.is_rejected:
2969 excuse.add_verdict_info(verdict, msg)
2970 else:
2971 excuse.addinfo(msg)
2973 return verdict