Coverage for britney2/policies/policy.py: 92%
1378 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-07-30 07:06 +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
20from britney2 import (
21 BinaryPackage,
22 BinaryPackageId,
23 DependencyType,
24 PackageId,
25 SourcePackage,
26 Suite,
27 SuiteClass,
28 Suites,
29 TargetSuite,
30)
31from britney2.excusedeps import DependencySpec
32from britney2.hints import (
33 Hint,
34 HintAnnotate,
35 HintCollection,
36 HintParser,
37 HintType,
38 PolicyHintParserProto,
39)
40from britney2.inputs.suiteloader import SuiteContentLoader
41from britney2.migrationitem import MigrationItem, MigrationItemFactory
42from britney2.policies import ApplySrcPolicy, PolicyVerdict
43from britney2.utils import (
44 GetDependencySolversProto,
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 args = args[1:]
399 for item in mi_factory.parse_items(*args):
400 hints.add_hint(
401 class_name(who, hint_type, converter(policy_parameter), [item])
402 )
404 return f
407class AgePolicy(AbstractBasePolicy):
408 """Configurable Aging policy for source migrations
410 The AgePolicy will let packages stay in the source suite for a pre-defined
411 amount of days before letting migrate (based on their urgency, if any).
413 The AgePolicy's decision is influenced by the following:
415 State files:
416 * ${STATE_DIR}/age-policy-urgencies: File containing urgencies for source
417 packages. Note that urgencies are "sticky" and the most "urgent" urgency
418 will be used (i.e. the one with lowest age-requirements).
419 - This file needs to be updated externally, if the policy should take
420 urgencies into consideration. If empty (or not updated), the policy
421 will simply use the default urgency (see the "Config" section below)
422 - In Debian, these values are taken from the .changes file, but that is
423 not a requirement for Britney.
424 * ${STATE_DIR}/age-policy-dates: File containing the age of all source
425 packages.
426 - The policy will automatically update this file.
427 Config:
428 * DEFAULT_URGENCY: Name of the urgency used for packages without an urgency
429 (or for unknown urgencies). Will also be used to set the "minimum"
430 aging requirements for packages not in the target suite.
431 * MINDAYS_<URGENCY>: The age-requirements in days for packages with the
432 given urgency.
433 - Commonly used urgencies are: low, medium, high, emergency, critical
434 Hints:
435 * urgent <source>/<version>: Disregard the age requirements for a given
436 source/version.
437 * age-days X <source>/<version>: Set the age requirements for a given
438 source/version to X days. Note that X can exceed the highest
439 age-requirement normally given.
441 """
443 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
444 super().__init__("age", options, suite_info, {SuiteClass.PRIMARY_SOURCE_SUITE})
445 self._min_days = self._generate_mindays_table()
446 self._min_days_default = 0
447 # 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)
448 # NB: _date_now is used in tests
449 time_now = time.time()
450 if hasattr(self.options, "fake_runtime"):
451 time_now = int(self.options.fake_runtime)
452 self.logger.info("overriding runtime with fake_runtime %d", time_now)
454 self._date_now = int(((time_now / (60 * 60)) - 19) / 24)
455 self._dates: dict[str, tuple[str, int]] = {}
456 self._urgencies: dict[str, str] = {}
457 self._default_urgency: str = self.options.default_urgency
458 self._penalty_immune_urgencies: frozenset[str] = frozenset()
459 if hasattr(self.options, "no_penalties"):
460 self._penalty_immune_urgencies = frozenset(
461 x.strip() for x in self.options.no_penalties.split()
462 )
463 self._bounty_min_age: int | None = None # initialised later
465 def _generate_mindays_table(self) -> dict[str, int]:
466 mindays: dict[str, int] = {}
467 for k in dir(self.options):
468 if not k.startswith("mindays_"):
469 continue
470 v = getattr(self.options, k)
471 try:
472 as_days = int(v)
473 except ValueError:
474 raise ValueError(
475 "Unable to parse "
476 + k
477 + " as a number of days. Must be 0 or a positive integer"
478 )
479 if as_days < 0: 479 ↛ 480line 479 didn't jump to line 480 because the condition on line 479 was never true
480 raise ValueError(
481 "The value of " + k + " must be zero or a positive integer"
482 )
483 mindays[k.split("_")[1]] = as_days
484 return mindays
486 def register_hints(self, hint_parser: HintParser) -> None:
487 hint_parser.register_hint_type(
488 HintType(
489 "age-days",
490 simple_policy_hint_parser_function(AgeDayHint, int),
491 min_args=2,
492 )
493 )
494 hint_parser.register_hint_type(HintType("urgent"))
496 def initialise(self, britney: "Britney") -> None:
497 super().initialise(britney)
498 self._read_dates_file()
499 self._read_urgencies_file()
500 if self._default_urgency not in self._min_days: # pragma: no cover
501 raise ValueError(
502 f"Missing age-requirement for default urgency (MINDAYS_{self._default_urgency})"
503 )
504 self._min_days_default = self._min_days[self._default_urgency]
505 try:
506 self._bounty_min_age = int(self.options.bounty_min_age)
507 except ValueError: 507 ↛ 508line 507 didn't jump to line 508 because the exception caught by line 507 didn't happen
508 if self.options.bounty_min_age in self._min_days:
509 self._bounty_min_age = self._min_days[self.options.bounty_min_age]
510 else: # pragma: no cover
511 raise ValueError(
512 "Please fix BOUNTY_MIN_AGE in the britney configuration"
513 )
514 except AttributeError:
515 # The option wasn't defined in the configuration
516 self._bounty_min_age = 0
518 def save_state(self, britney: "Britney") -> None:
519 super().save_state(britney)
520 self._write_dates_file()
522 def apply_src_policy_impl(
523 self,
524 age_info: dict[str, Any],
525 source_data_tdist: SourcePackage | None,
526 source_data_srcdist: SourcePackage,
527 excuse: "Excuse",
528 ) -> PolicyVerdict:
529 # retrieve the urgency for the upload, ignoring it if this is a NEW package
530 # (not present in the target suite)
531 source_name = excuse.item.package
532 urgency = self._urgencies.get(source_name, self._default_urgency)
534 if urgency not in self._min_days: 534 ↛ 535line 534 didn't jump to line 535 because the condition on line 534 was never true
535 age_info["unknown-urgency"] = urgency
536 urgency = self._default_urgency
538 if not source_data_tdist:
539 if self._min_days[urgency] < self._min_days_default:
540 age_info["urgency-reduced"] = {
541 "from": urgency,
542 "to": self._default_urgency,
543 }
544 urgency = self._default_urgency
546 if source_name not in self._dates:
547 self._dates[source_name] = (source_data_srcdist.version, self._date_now)
548 elif self._dates[source_name][0] != source_data_srcdist.version:
549 self._dates[source_name] = (source_data_srcdist.version, self._date_now)
551 days_old = self._date_now - self._dates[source_name][1]
552 min_days = self._min_days[urgency]
553 for bounty, bounty_value in excuse.bounty.items():
554 if bounty_value: 554 ↛ 553line 554 didn't jump to line 553 because the condition on line 554 was always true
555 self.logger.info(
556 "Applying bounty for %s granted by %s: %d days",
557 source_name,
558 bounty,
559 bounty_value,
560 )
561 excuse.addinfo(
562 f"Required age reduced by {bounty_value} days because of {bounty}"
563 )
564 assert bounty_value > 0, "negative bounties shouldn't happen"
565 min_days -= bounty_value
566 if urgency not in self._penalty_immune_urgencies:
567 for penalty, penalty_value in excuse.penalty.items():
568 if penalty_value: 568 ↛ 567line 568 didn't jump to line 567 because the condition on line 568 was always true
569 self.logger.info(
570 "Applying penalty for %s given by %s: %d days",
571 source_name,
572 penalty,
573 penalty_value,
574 )
575 excuse.addinfo(
576 f"Required age increased by {penalty_value} days because of {penalty}"
577 )
578 assert (
579 penalty_value > 0
580 ), "negative penalties should be handled earlier"
581 min_days += penalty_value
583 assert self._bounty_min_age is not None
584 # the age in BOUNTY_MIN_AGE can be higher than the one associated with
585 # the real urgency, so don't forget to take it into account
586 bounty_min_age = min(self._bounty_min_age, self._min_days[urgency])
587 if min_days < bounty_min_age:
588 min_days = bounty_min_age
589 excuse.addinfo(f"Required age is not allowed to drop below {min_days} days")
591 age_info["current-age"] = days_old
593 assert self.hints is not None
594 for hint in self.hints.search(
595 "age-days", package=source_name, version=source_data_srcdist.version
596 ):
597 age_days_hint = cast("AgeDayHint", hint)
599 new_req = age_days_hint.days
600 age_info["age-requirement-reduced"] = {
601 "new-requirement": new_req,
602 "changed-by": age_days_hint.user,
603 }
604 if "original-age-requirement" not in age_info: 604 ↛ 606line 604 didn't jump to line 606 because the condition on line 604 was always true
605 age_info["original-age-requirement"] = min_days
606 min_days = new_req
608 age_info["age-requirement"] = min_days
609 res = PolicyVerdict.PASS
611 if days_old < min_days:
612 if (
613 urgent_hint := self.hints.search_first(
614 "urgent", package=source_name, version=source_data_srcdist.version
615 )
616 ) is not None:
617 age_info["age-requirement-reduced"] = {
618 "new-requirement": 0,
619 "changed-by": urgent_hint.user,
620 }
621 res = PolicyVerdict.PASS_HINTED
622 else:
623 res = PolicyVerdict.REJECTED_TEMPORARILY
625 # update excuse
626 age_hint = age_info.get("age-requirement-reduced", None)
627 age_min_req = age_info["age-requirement"]
628 if age_hint is not None:
629 new_req = age_hint["new-requirement"]
630 who = age_hint["changed-by"]
631 if new_req:
632 excuse.addinfo(
633 f"Overriding age needed from {age_min_req} days to {new_req} by {who}"
634 )
635 age_min_req = new_req
636 else:
637 excuse.addinfo(f"Too young, but urgency pushed by {who}")
638 age_min_req = 0
639 excuse.setdaysold(age_info["current-age"], age_min_req)
641 if age_min_req == 0:
642 excuse.addinfo(f"{days_old} days old")
643 elif days_old < age_min_req:
644 excuse.add_verdict_info(
645 res, f"Too young, only {days_old} of {age_min_req} days old"
646 )
647 else:
648 excuse.addinfo(f"{days_old} days old (needed {age_min_req} days)")
650 return res
652 def _read_dates_file(self) -> None:
653 """Parse the dates file"""
654 dates = self._dates
655 fallback_filename = os.path.join(self.suite_info.target_suite.path, "Dates")
656 using_new_name = False
657 try:
658 filename = os.path.join(self.state_dir, "age-policy-dates")
659 if not os.path.exists(filename) and os.path.exists(fallback_filename): 659 ↛ 660line 659 didn't jump to line 660 because the condition on line 659 was never true
660 filename = fallback_filename
661 else:
662 using_new_name = True
663 except AttributeError:
664 if os.path.exists(fallback_filename):
665 filename = fallback_filename
666 else:
667 raise RuntimeError("Please set STATE_DIR in the britney configuration")
669 try:
670 with open(filename, encoding="utf-8") as fd:
671 for line in fd:
672 if line.startswith("#"):
673 # Ignore comment lines (mostly used for tests)
674 continue
675 # <source> <version> <date>)
676 ln = line.split()
677 if len(ln) != 3: # pragma: no cover
678 continue
679 try:
680 dates[ln[0]] = (ln[1], int(ln[2]))
681 except ValueError: # pragma: no cover
682 pass
683 except FileNotFoundError:
684 if not using_new_name: 684 ↛ 686line 684 didn't jump to line 686 because the condition on line 684 was never true
685 # If we using the legacy name, then just give up
686 raise
687 self.logger.info("%s does not appear to exist. Creating it", filename)
688 with open(filename, mode="x", encoding="utf-8"):
689 pass
691 def _read_urgencies_file(self) -> None:
692 urgencies = self._urgencies
693 min_days_default = self._min_days_default
694 fallback_filename = os.path.join(self.suite_info.target_suite.path, "Urgency")
695 try:
696 filename = os.path.join(self.state_dir, "age-policy-urgencies")
697 if not os.path.exists(filename) and os.path.exists(fallback_filename): 697 ↛ 698line 697 didn't jump to line 698 because the condition on line 697 was never true
698 filename = fallback_filename
699 except AttributeError:
700 filename = fallback_filename
702 sources_s = self.suite_info.primary_source_suite.sources
703 sources_t = self.suite_info.target_suite.sources
705 with open(filename, errors="surrogateescape", encoding="ascii") as fd:
706 for line in fd:
707 if line.startswith("#"):
708 # Ignore comment lines (mostly used for tests)
709 continue
710 # <source> <version> <urgency>
711 ln = line.split()
712 if len(ln) != 3: 712 ↛ 713line 712 didn't jump to line 713 because the condition on line 712 was never true
713 continue
715 # read the minimum days associated with the urgencies
716 urgency_old = urgencies.get(ln[0], None)
717 mindays_old = self._min_days.get(urgency_old, 1000) # type: ignore[arg-type]
718 mindays_new = self._min_days.get(ln[2], min_days_default)
720 # if the new urgency is lower (so the min days are higher), do nothing
721 if mindays_old <= mindays_new:
722 continue
724 # if the package exists in the target suite and it is more recent, do nothing
725 tsrcv = sources_t.get(ln[0], None)
726 if tsrcv and apt_pkg.version_compare(tsrcv.version, ln[1]) >= 0:
727 continue
729 # if the package doesn't exist in the primary source suite or it is older, do nothing
730 usrcv = sources_s.get(ln[0], None)
731 if not usrcv or apt_pkg.version_compare(usrcv.version, ln[1]) < 0: 731 ↛ 732line 731 didn't jump to line 732 because the condition on line 731 was never true
732 continue
734 # update the urgency for the package
735 urgencies[ln[0]] = ln[2]
737 def _write_dates_file(self) -> None:
738 dates = self._dates
739 try:
740 directory = self.state_dir
741 basename = "age-policy-dates"
742 old_file = os.path.join(self.suite_info.target_suite.path, "Dates")
743 except AttributeError:
744 directory = self.suite_info.target_suite.path
745 basename = "Dates"
746 old_file = None
747 filename = os.path.join(directory, basename)
748 filename_tmp = os.path.join(directory, f"{basename}_new")
749 with open(filename_tmp, "w", encoding="utf-8") as fd:
750 fd.writelines(
751 f"{pkg} {version} {date}\n"
752 for pkg, (version, date) in sorted(dates.items())
753 )
754 os.rename(filename_tmp, filename)
755 if old_file is not None and os.path.exists(old_file): 755 ↛ 756line 755 didn't jump to line 756 because the condition on line 755 was never true
756 self.logger.info("Removing old age-policy-dates file %s", old_file)
757 os.unlink(old_file)
760class RCBugPolicy(AbstractBasePolicy):
761 """RC bug regression policy for source migrations
763 The RCBugPolicy will read provided list of RC bugs and block any
764 source upload that would introduce a *new* RC bug in the target
765 suite.
767 The RCBugPolicy's decision is influenced by the following:
769 State files:
770 * ${STATE_DIR}/rc-bugs-${SUITE_NAME}: File containing RC bugs for packages in
771 the given suite (one for both primary source suite and the target sutie is
772 needed).
773 - These files need to be updated externally.
774 """
776 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
777 super().__init__(
778 "rc-bugs", options, suite_info, {SuiteClass.PRIMARY_SOURCE_SUITE}
779 )
780 self._bugs_source: dict[str, set[str]] | None = None
781 self._bugs_target: dict[str, set[str]] | None = None
783 def register_hints(self, hint_parser: HintParser) -> None:
784 f = simple_policy_hint_parser_function(
785 IgnoreRCBugHint, lambda x: frozenset(x.split(","))
786 )
787 hint_parser.register_hint_type(HintType("ignore-rc-bugs", f, min_args=2))
789 def initialise(self, britney: "Britney") -> None:
790 super().initialise(britney)
791 source_suite = self.suite_info.primary_source_suite
792 target_suite = self.suite_info.target_suite
793 fallback_unstable = os.path.join(source_suite.path, "BugsV")
794 fallback_testing = os.path.join(target_suite.path, "BugsV")
795 try:
796 filename_unstable = os.path.join(
797 self.state_dir, f"rc-bugs-{source_suite.name}"
798 )
799 filename_testing = os.path.join(
800 self.state_dir, f"rc-bugs-{target_suite.name}"
801 )
802 if ( 802 ↛ 808line 802 didn't jump to line 808
803 not os.path.exists(filename_unstable)
804 and not os.path.exists(filename_testing)
805 and os.path.exists(fallback_unstable)
806 and os.path.exists(fallback_testing)
807 ):
808 filename_unstable = fallback_unstable
809 filename_testing = fallback_testing
810 except AttributeError:
811 filename_unstable = fallback_unstable
812 filename_testing = fallback_testing
813 self._bugs_source = self._read_bugs(filename_unstable)
814 self._bugs_target = self._read_bugs(filename_testing)
816 def apply_src_policy_impl(
817 self,
818 rcbugs_info: dict[str, Any],
819 source_data_tdist: SourcePackage | None,
820 source_data_srcdist: SourcePackage,
821 excuse: "Excuse",
822 ) -> PolicyVerdict:
823 assert self._bugs_source is not None # for type checking
824 assert self._bugs_target is not None # for type checking
825 bugs_t = set()
826 bugs_s = set()
827 source_name = excuse.item.package
828 binaries_s = {x.package_name for x in source_data_srcdist.binaries}
829 try:
830 binaries_t = {x.package_name for x in source_data_tdist.binaries} # type: ignore[union-attr]
831 except AttributeError:
832 binaries_t = set()
834 src_key = f"src:{source_name}"
835 if source_data_tdist and src_key in self._bugs_target:
836 bugs_t.update(self._bugs_target[src_key])
837 if src_key in self._bugs_source:
838 bugs_s.update(self._bugs_source[src_key])
840 for pkg in binaries_s:
841 if pkg in self._bugs_source:
842 bugs_s |= self._bugs_source[pkg]
843 for pkg in binaries_t:
844 if pkg in self._bugs_target:
845 bugs_t |= self._bugs_target[pkg]
847 # The bts seems to support filing source bugs against a binary of the
848 # same name if that binary isn't built by any source. An example is bug
849 # 820347 against Package: juce (in the live-2016-04-11 test). Add those
850 # bugs too.
851 if (
852 source_name not in (binaries_s | binaries_t)
853 and source_name
854 not in {
855 x.package_name
856 for x in self.suite_info.primary_source_suite.all_binaries_in_suite.keys()
857 }
858 and source_name
859 not in {
860 x.package_name
861 for x in self.suite_info.target_suite.all_binaries_in_suite.keys()
862 }
863 ):
864 if source_name in self._bugs_source:
865 bugs_s |= self._bugs_source[source_name]
866 if source_name in self._bugs_target: 866 ↛ 867line 866 didn't jump to line 867 because the condition on line 866 was never true
867 bugs_t |= self._bugs_target[source_name]
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 not bugs_t or source_data_tdist, (
878 "%s had bugs in the target suite but is not present" % source_name
879 )
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-bugs" in rcbugs_info:
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 rcbugs_info["ignored-bugs"]["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 if source_name in self._piuparts_target:
1053 testing_state = self._piuparts_target[source_name]
1054 else:
1055 testing_state = PiupartsState.UNKNOWN
1056 url: str | None
1057 if source_name in self._piuparts_source:
1058 unstable_state, url = self._piuparts_source[source_name]
1059 else:
1060 unstable_state = PiupartsState.UNKNOWN
1061 url = None
1062 url_html = "(no link yet)"
1063 if url is not None:
1064 url_html = '<a href="{0}">{0}</a>'.format(url)
1066 match unstable_state:
1067 case PiupartsState.PASS:
1068 # Not a regression
1069 msg = f"Piuparts tested OK - {url_html}"
1070 result = PolicyVerdict.PASS
1071 piuparts_info["test-results"] = PiupartsResult.PASS
1072 case PiupartsState.FAIL if testing_state is not PiupartsState.FAIL:
1073 piuparts_info["test-results"] = PiupartsResult.REGRESSION
1074 msg = f"Piuparts regression - {url_html}"
1075 result = PolicyVerdict.REJECTED_PERMANENTLY
1076 case PiupartsState.FAIL:
1077 piuparts_info["test-results"] = PiupartsResult.FAILED
1078 msg = f"Piuparts failure (not a regression) - {url_html}"
1079 result = PolicyVerdict.PASS
1080 case PiupartsState.WAITING:
1081 msg = f"Piuparts check waiting for test results - {url_html}"
1082 result = PolicyVerdict.REJECTED_TEMPORARILY
1083 piuparts_info["test-results"] = PiupartsResult.WAITING_FOR_TESTS
1084 case _:
1085 msg = f"Piuparts can't test {source_name} (not a blocker) - {url_html}"
1086 piuparts_info["test-results"] = PiupartsResult.CANNOT_BE_TESTED
1087 result = PolicyVerdict.PASS
1089 if url is not None:
1090 piuparts_info["piuparts-test-url"] = url
1091 if result.is_rejected:
1092 excuse.add_verdict_info(result, msg)
1093 else:
1094 excuse.addinfo(msg)
1096 if result.is_rejected:
1097 assert self.hints is not None
1098 if (
1099 ignore_hint := self.hints.search_first(
1100 "ignore-piuparts",
1101 package=source_name,
1102 version=source_data_srcdist.version,
1103 )
1104 ) is not None:
1105 piuparts_info["ignored-piuparts"] = {"issued-by": ignore_hint.user}
1106 result = PolicyVerdict.PASS_HINTED
1107 excuse.addinfo(
1108 f"Piuparts issue ignored as requested by {ignore_hint.user}"
1109 )
1111 return result
1113 def _read_piuparts_summary_gen(
1114 self, filename: str
1115 ) -> Iterator[tuple[str, PiupartsState, str]]:
1116 self.logger.info("Loading piuparts report from %s", filename)
1117 with open(filename) as fd: 1117 ↛ exitline 1117 didn't return from function '_read_piuparts_summary_gen' because the return on line 1119 wasn't executed
1118 if os.fstat(fd.fileno()).st_size < 1: 1118 ↛ 1119line 1118 didn't jump to line 1119 because the condition on line 1118 was never true
1119 return
1120 data = json.load(fd)
1121 try:
1122 if (
1123 data["_id"] != "Piuparts Package Test Results Summary"
1124 or data["_version"] != "1.0"
1125 ): # pragma: no cover
1126 raise ValueError(
1127 f"Piuparts results in {filename} does not have the correct ID or version"
1128 )
1129 except KeyError as e: # pragma: no cover
1130 raise ValueError(
1131 f"Piuparts results in {filename} is missing id or version field"
1132 ) from e
1133 for source, suite_data in data["packages"].items():
1134 if len(suite_data) != 1: # pragma: no cover
1135 raise ValueError(
1136 f"Piuparts results in {filename}, the source {source} does not have "
1137 "exactly one result set"
1138 )
1139 item = next(iter(suite_data.values()))
1140 state, _, url = item
1141 yield (source, PiupartsState.from_str(state), url)
1143 def _read_piuparts_summary(
1144 self, filename: str
1145 ) -> dict[str, tuple[PiupartsState, str]]:
1146 return {
1147 source: (state, url)
1148 for (source, state, url) in self._read_piuparts_summary_gen(filename)
1149 }
1151 def _read_piuparts_summary_without_url(
1152 self, filename: str
1153 ) -> dict[str, PiupartsState]:
1154 return {
1155 source: state
1156 for (source, state, _) in self._read_piuparts_summary_gen(filename)
1157 }
1160class DependsPolicy(AbstractBasePolicy):
1161 pkg_universe: "BinaryPackageUniverse"
1162 broken_packages: frozenset["BinaryPackageId"]
1163 all_binaries: dict["BinaryPackageId", "BinaryPackage"]
1164 allow_uninst: dict[str, set[str | None]]
1166 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
1167 super().__init__(
1168 "depends",
1169 options,
1170 suite_info,
1171 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
1172 ApplySrcPolicy.RUN_ON_EVERY_ARCH_ONLY,
1173 )
1174 self.nobreakall_arches = None
1175 self.new_arches = None
1176 self.break_arches = None
1178 def initialise(self, britney: "Britney") -> None:
1179 super().initialise(britney)
1180 self.pkg_universe = britney.pkg_universe
1181 self.broken_packages = self.pkg_universe.broken_packages
1182 self.all_binaries = britney.all_binaries
1183 self.nobreakall_arches = self.options.nobreakall_arches
1184 self.new_arches = self.options.new_arches
1185 self.break_arches = self.options.break_arches
1186 self.allow_uninst = britney.allow_uninst
1188 def apply_srcarch_policy_impl(
1189 self,
1190 deps_info: dict[str, Any],
1191 arch: str,
1192 source_data_tdist: SourcePackage | None,
1193 source_data_srcdist: SourcePackage,
1194 excuse: "Excuse",
1195 ) -> PolicyVerdict:
1196 verdict = PolicyVerdict.PASS
1198 assert self.break_arches is not None
1199 assert self.new_arches is not None
1200 if arch in self.break_arches or arch in self.new_arches:
1201 # we don't check these in the policy (TODO - for now?)
1202 return verdict
1204 item = excuse.item
1205 source_suite = item.suite
1206 target_suite = self.suite_info.target_suite
1208 packages_s_a = source_suite.binaries[arch]
1209 packages_t_a = target_suite.binaries[arch]
1211 my_bins = sorted(filter_out_faux_gen(excuse.packages[arch]))
1213 arch_all_installable = set()
1214 arch_arch_installable = set()
1215 consider_it_regression = True
1217 for pkg_id in my_bins:
1218 pkg_name = pkg_id.package_name
1219 binary_u = packages_s_a[pkg_name]
1220 pkg_arch = binary_u.architecture
1222 # in some cases, we want to track the uninstallability of a
1223 # package (because the autopkgtest policy uses this), but we still
1224 # want to allow the package to be uninstallable
1225 skip_dep_check = False
1227 if binary_u.source_version != source_data_srcdist.version:
1228 # don't check cruft in unstable
1229 continue
1231 if item.architecture != "source" and pkg_arch == "all":
1232 # we don't care about the existing arch: all binaries when
1233 # checking a binNMU item, because the arch: all binaries won't
1234 # migrate anyway
1235 skip_dep_check = True
1237 if pkg_arch == "all" and arch not in self.nobreakall_arches:
1238 skip_dep_check = True
1240 if pkg_name in self.allow_uninst[arch]: 1240 ↛ 1243line 1240 didn't jump to line 1243 because the condition on line 1240 was never true
1241 # this binary is allowed to become uninstallable, so we don't
1242 # need to check anything
1243 skip_dep_check = True
1245 if pkg_name in packages_t_a:
1246 oldbin = packages_t_a[pkg_name]
1247 if not target_suite.is_installable(oldbin.pkg_id):
1248 # as the current binary in testing is already
1249 # uninstallable, the newer version is allowed to be
1250 # uninstallable as well, so we don't need to check
1251 # anything
1252 skip_dep_check = True
1253 consider_it_regression = False
1255 if pkg_id in self.broken_packages:
1256 if pkg_arch == "all":
1257 arch_all_installable.add(False)
1258 else:
1259 arch_arch_installable.add(False)
1260 # dependencies can't be satisfied by all the known binaries -
1261 # this certainly won't work...
1262 excuse.add_unsatisfiable_on_arch(arch)
1263 if skip_dep_check:
1264 # ...but if the binary is allowed to become uninstallable,
1265 # we don't care
1266 # we still want the binary to be listed as uninstallable,
1267 continue
1268 verdict = PolicyVerdict.REJECTED_PERMANENTLY
1269 if pkg_name.endswith("-faux-build-depends"): 1269 ↛ 1270line 1269 didn't jump to line 1270 because the condition on line 1269 was never true
1270 name = pkg_name.removesuffix("-faux-build-depends")
1271 excuse.add_verdict_info(
1272 verdict,
1273 f"src:{name} has unsatisfiable build dependency",
1274 )
1275 else:
1276 excuse.add_verdict_info(
1277 verdict, f"{pkg_name}/{arch} has unsatisfiable dependency"
1278 )
1279 excuse.addreason("depends")
1280 else:
1281 if pkg_arch == "all":
1282 arch_all_installable.add(True)
1283 else:
1284 arch_arch_installable.add(True)
1286 if skip_dep_check:
1287 continue
1289 deps = self.pkg_universe.dependencies_of(pkg_id)
1291 for dep in deps:
1292 # dep is a list of packages, each of which satisfy the
1293 # dependency
1295 if not dep:
1296 continue
1297 is_ok = False
1298 needed_for_dep = set()
1300 for alternative in dep:
1301 if target_suite.is_pkg_in_the_suite(alternative):
1302 # dep can be satisfied in testing - ok
1303 is_ok = True
1304 elif alternative in my_bins:
1305 # can be satisfied by binary from same item: will be
1306 # ok if item migrates
1307 is_ok = True
1308 else:
1309 needed_for_dep.add(alternative)
1311 if not is_ok:
1312 spec = DependencySpec(DependencyType.DEPENDS, arch)
1313 excuse.add_package_depends(spec, needed_for_dep)
1315 # The autopkgtest policy needs delicate trade offs for
1316 # non-installability. The current choice (considering source
1317 # migration and only binaries built by the version of the
1318 # source):
1319 #
1320 # * Run autopkgtest if all arch:$arch binaries are installable
1321 # (but some or all arch:all binaries are not)
1322 #
1323 # * Don't schedule nor wait for not installable arch:all only package
1324 # on ! NOBREAKALL_ARCHES
1325 #
1326 # * Run autopkgtest if installability isn't a regression (there are (or
1327 # rather, should) not be a lot of packages in this state, and most
1328 # likely they'll just fail quickly)
1329 #
1330 # * Don't schedule, but wait otherwise
1331 if arch_arch_installable == {True} and False in arch_all_installable:
1332 deps_info.setdefault("autopkgtest_run_anyways", []).append(arch)
1333 elif (
1334 arch not in self.nobreakall_arches
1335 and not arch_arch_installable
1336 and False in arch_all_installable
1337 ):
1338 deps_info.setdefault("arch_all_not_installable", []).append(arch)
1339 elif not consider_it_regression:
1340 deps_info.setdefault("autopkgtest_run_anyways", []).append(arch)
1342 return verdict
1345@unique
1346class BuildDepResult(IntEnum):
1347 # relation is satisfied in target
1348 OK = 1
1349 # relation can be satisfied by other packages in source
1350 DEPENDS = 2
1351 # relation cannot be satisfied
1352 FAILED = 3
1355class BuildDependsPolicy(AbstractBasePolicy):
1357 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
1358 super().__init__(
1359 "build-depends",
1360 options,
1361 suite_info,
1362 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
1363 )
1364 self._all_buildarch: list[str] = []
1366 parse_option(options, "all_buildarch")
1368 def initialise(self, britney: "Britney") -> None:
1369 super().initialise(britney)
1370 if self.options.all_buildarch:
1371 self._all_buildarch = SuiteContentLoader.config_str_as_list(
1372 self.options.all_buildarch, []
1373 )
1375 def apply_src_policy_impl(
1376 self,
1377 build_deps_info: dict[str, Any],
1378 source_data_tdist: SourcePackage | None,
1379 source_data_srcdist: SourcePackage,
1380 excuse: "Excuse",
1381 get_dependency_solvers: GetDependencySolversProto = get_dependency_solvers,
1382 ) -> PolicyVerdict:
1383 verdict = PolicyVerdict.PASS
1385 # analyze the dependency fields (if present)
1386 if deps := source_data_srcdist.build_deps_arch:
1387 v = self._check_build_deps(
1388 deps,
1389 DependencyType.BUILD_DEPENDS,
1390 build_deps_info,
1391 source_data_srcdist,
1392 excuse,
1393 get_dependency_solvers=get_dependency_solvers,
1394 )
1395 verdict = PolicyVerdict.worst_of(verdict, v)
1397 if ideps := source_data_srcdist.build_deps_indep:
1398 v = self._check_build_deps(
1399 ideps,
1400 DependencyType.BUILD_DEPENDS_INDEP,
1401 build_deps_info,
1402 source_data_srcdist,
1403 excuse,
1404 get_dependency_solvers=get_dependency_solvers,
1405 )
1406 verdict = PolicyVerdict.worst_of(verdict, v)
1408 return verdict
1410 def _get_check_archs(
1411 self, archs: Container[str], dep_type: DependencyType
1412 ) -> list[str]:
1413 oos = self.options.outofsync_arches
1415 if dep_type is DependencyType.BUILD_DEPENDS:
1416 return [
1417 arch
1418 for arch in self.options.architectures
1419 if arch in archs and arch not in oos
1420 ]
1422 # first try the all buildarch
1423 checkarchs = list(self._all_buildarch)
1424 # then try the architectures where this source has arch specific
1425 # binaries (in the order of the architecture config file)
1426 checkarchs.extend(
1427 arch
1428 for arch in self.options.architectures
1429 if arch in archs and arch not in checkarchs
1430 )
1431 # then try all other architectures
1432 checkarchs.extend(
1433 arch for arch in self.options.architectures if arch not in checkarchs
1434 )
1436 # and drop OUTOFSYNC_ARCHES
1437 return [arch for arch in checkarchs if arch not in oos]
1439 def _add_info_for_arch(
1440 self,
1441 arch: str,
1442 excuses_info: dict[str, list[str]],
1443 blockers: dict[str, set[BinaryPackageId]],
1444 results: dict[str, BuildDepResult],
1445 dep_type: DependencyType,
1446 excuse: "Excuse",
1447 verdict: PolicyVerdict,
1448 ) -> PolicyVerdict:
1449 if arch in blockers:
1450 packages = blockers[arch]
1452 # for the solving packages, update the excuse to add the dependencies
1453 for p in packages:
1454 if arch not in self.options.break_arches: 1454 ↛ 1453line 1454 didn't jump to line 1453 because the condition on line 1454 was always true
1455 spec = DependencySpec(dep_type, arch)
1456 excuse.add_package_depends(spec, {p})
1458 if arch in results and results[arch] is BuildDepResult.FAILED:
1459 verdict = PolicyVerdict.worst_of(
1460 verdict, PolicyVerdict.REJECTED_PERMANENTLY
1461 )
1463 if arch in excuses_info:
1464 for excuse_text in excuses_info[arch]:
1465 if verdict.is_rejected: 1465 ↛ 1468line 1465 didn't jump to line 1468 because the condition on line 1465 was always true
1466 excuse.add_verdict_info(verdict, excuse_text)
1467 else:
1468 excuse.addinfo(excuse_text)
1470 return verdict
1472 def _check_build_deps(
1473 self,
1474 deps: str,
1475 dep_type: DependencyType,
1476 build_deps_info: dict[str, Any],
1477 source_data_srcdist: SourcePackage,
1478 excuse: "Excuse",
1479 get_dependency_solvers: GetDependencySolversProto = get_dependency_solvers,
1480 ) -> PolicyVerdict:
1481 verdict = PolicyVerdict.PASS
1482 any_arch_ok = dep_type is DependencyType.BUILD_DEPENDS_INDEP
1484 britney = self.britney
1486 # local copies for better performance
1487 parse_src_depends = apt_pkg.parse_src_depends
1489 source_name = excuse.item.package
1490 source_suite = excuse.item.suite
1491 target_suite = self.suite_info.target_suite
1492 binaries_s = source_suite.binaries
1493 provides_s = source_suite.provides_table
1494 binaries_t = target_suite.binaries
1495 provides_t = target_suite.provides_table
1496 unsat_bd: dict[str, list[str]] = {}
1497 relevant_archs: set[str] = {
1498 binary.architecture
1499 for binary in filter_out_faux_gen(source_data_srcdist.binaries)
1500 if britney.all_binaries[binary].architecture != "all"
1501 }
1503 excuses_info: dict[str, list[str]] = defaultdict(list)
1504 blockers: dict[str, set[BinaryPackageId]] = defaultdict(set)
1505 arch_results = {}
1506 result_archs = defaultdict(list)
1507 bestresult = BuildDepResult.FAILED
1508 check_archs = self._get_check_archs(relevant_archs, dep_type)
1509 if not check_archs:
1510 # when the arch list is empty, we check the b-d on any arch, instead of all archs
1511 # this happens for Build-Depens on a source package that only produces arch: all binaries
1512 any_arch_ok = True
1513 check_archs = self._get_check_archs(
1514 self.options.architectures, DependencyType.BUILD_DEPENDS_INDEP
1515 )
1517 for arch in check_archs:
1518 # retrieve the binary package from the specified suite and arch
1519 binaries_s_a = binaries_s[arch]
1520 provides_s_a = provides_s[arch]
1521 binaries_t_a = binaries_t[arch]
1522 provides_t_a = provides_t[arch]
1523 arch_results[arch] = BuildDepResult.OK
1524 # for every dependency block (formed as conjunction of disjunction)
1525 for block_txt in deps.split(","):
1526 block_list = parse_src_depends(block_txt, False, arch)
1527 # Unlike regular dependencies, some clauses of the Build-Depends(-Arch|-Indep) can be
1528 # filtered out by (e.g.) architecture restrictions. We need to cope with this while
1529 # keeping block_txt and block aligned.
1530 if not block_list:
1531 # Relation is not relevant for this architecture.
1532 continue
1533 block = block_list[0]
1534 # if the block is satisfied in the target suite, then skip the block
1535 if get_dependency_solvers(
1536 block, binaries_t_a, provides_t_a, build_depends=True
1537 ):
1538 # Satisfied in the target suite; all ok.
1539 continue
1541 # check if the block can be satisfied in the source suite, and list the solving packages
1542 packages = get_dependency_solvers(
1543 block, binaries_s_a, provides_s_a, build_depends=True
1544 )
1546 # if the dependency can be satisfied by the same source package, skip the block:
1547 # obviously both binary packages will enter the target suite together
1548 if any(source_name == p.source for p in packages): 1548 ↛ 1549line 1548 didn't jump to line 1549 because the condition on line 1548 was never true
1549 continue
1551 # if no package can satisfy the dependency, add this information to the excuse
1552 if not packages:
1553 excuses_info[arch].append(
1554 "%s unsatisfiable %s on %s: %s"
1555 % (source_name, dep_type, arch, block_txt.strip())
1556 )
1557 if arch not in unsat_bd: 1557 ↛ 1559line 1557 didn't jump to line 1559 because the condition on line 1557 was always true
1558 unsat_bd[arch] = []
1559 unsat_bd[arch].append(block_txt.strip())
1560 arch_results[arch] = BuildDepResult.FAILED
1561 continue
1563 blockers[arch].update(p.pkg_id for p in packages)
1564 if arch_results[arch] < BuildDepResult.DEPENDS:
1565 arch_results[arch] = BuildDepResult.DEPENDS
1567 if any_arch_ok:
1568 if arch_results[arch] < bestresult:
1569 bestresult = arch_results[arch]
1570 result_archs[arch_results[arch]].append(arch)
1571 if bestresult is BuildDepResult.OK:
1572 # we found an architecture where the b-deps-indep are
1573 # satisfied in the target suite, so we can stop
1574 break
1576 if any_arch_ok:
1577 arch = result_archs[bestresult][0]
1578 excuse.add_detailed_info(f"Checking {dep_type.get_description()} on {arch}")
1579 key = "check-%s-on-arch" % dep_type.get_reason()
1580 build_deps_info[key] = arch
1581 verdict = self._add_info_for_arch(
1582 arch,
1583 excuses_info,
1584 blockers,
1585 arch_results,
1586 dep_type,
1587 excuse,
1588 verdict,
1589 )
1591 else:
1592 for arch in check_archs:
1593 verdict = self._add_info_for_arch(
1594 arch,
1595 excuses_info,
1596 blockers,
1597 arch_results,
1598 dep_type,
1599 excuse,
1600 verdict,
1601 )
1603 if unsat_bd:
1604 build_deps_info["unsatisfiable-arch-build-depends"] = unsat_bd
1606 return verdict
1609class BuiltUsingPolicy(AbstractBasePolicy):
1610 """Built-Using policy
1612 Binaries that incorporate (part of) another source package must list these
1613 sources under 'Built-Using'.
1615 This policy checks if the corresponding sources are available in the
1616 target suite. If they are not, but they are candidates for migration, a
1617 dependency is added.
1619 If the binary incorporates a newer version of a source, that is not (yet)
1620 a candidate, we don't want to accept that binary. A rebuild later in the
1621 primary suite wouldn't fix the issue, because that would incorporate the
1622 newer version again.
1624 If the binary incorporates an older version of the source, a newer version
1625 will be accepted as a replacement. We assume that this can be fixed by
1626 rebuilding the binary at some point during the development cycle.
1628 Requiring exact version of the source would not be useful in practice. A
1629 newer upload of that source wouldn't be blocked by this policy, so the
1630 built-using would be outdated anyway.
1632 """
1634 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
1635 super().__init__(
1636 "built-using",
1637 options,
1638 suite_info,
1639 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
1640 ApplySrcPolicy.RUN_ON_EVERY_ARCH_ONLY,
1641 )
1643 def initialise(self, britney: "Britney") -> None:
1644 super().initialise(britney)
1646 def apply_srcarch_policy_impl(
1647 self,
1648 build_deps_info: dict[str, Any],
1649 arch: str,
1650 source_data_tdist: SourcePackage | None,
1651 source_data_srcdist: SourcePackage,
1652 excuse: "Excuse",
1653 ) -> PolicyVerdict:
1654 verdict = PolicyVerdict.PASS
1656 source_suite = excuse.item.suite
1657 target_suite = self.suite_info.target_suite
1658 binaries_s = source_suite.binaries
1660 def check_bu_in_suite(
1661 bu_source: str, bu_version: str, source_suite: Suite
1662 ) -> bool:
1663 if bu_source not in source_suite.sources:
1664 return False
1665 s_source = source_suite.sources[bu_source]
1666 s_ver = s_source.version
1667 if apt_pkg.version_compare(s_ver, bu_version) >= 0:
1668 dep = PackageId(bu_source, s_ver, "source")
1669 if arch in self.options.break_arches:
1670 excuse.add_detailed_info(
1671 "Ignoring Built-Using for %s/%s on %s"
1672 % (pkg_name, arch, dep.uvname)
1673 )
1674 else:
1675 spec = DependencySpec(DependencyType.BUILT_USING, arch)
1676 excuse.add_package_depends(spec, {dep})
1677 excuse.add_detailed_info(
1678 f"{pkg_name}/{arch} has Built-Using on {dep.uvname}"
1679 )
1680 return True
1682 return False
1684 for pkg_id in sorted(
1685 x
1686 for x in filter_out_faux_gen(source_data_srcdist.binaries)
1687 if x.architecture == arch
1688 ):
1689 pkg_name = pkg_id.package_name
1691 # retrieve the testing (if present) and unstable corresponding binary packages
1692 binary_s = binaries_s[arch][pkg_name]
1693 if binary_s.builtusing is None:
1694 continue
1696 for bu in binary_s.builtusing:
1697 bu_source = bu[0]
1698 bu_version = bu[1]
1699 found = False
1700 if bu_source in target_suite.sources:
1701 t_source = target_suite.sources[bu_source]
1702 t_ver = t_source.version
1703 if apt_pkg.version_compare(t_ver, bu_version) >= 0:
1704 found = True
1706 if not found:
1707 found = check_bu_in_suite(bu_source, bu_version, source_suite)
1709 if not found and source_suite.suite_class.is_additional_source:
1710 found = check_bu_in_suite(
1711 bu_source, bu_version, self.suite_info.primary_source_suite
1712 )
1714 if not found:
1715 if arch in self.options.break_arches:
1716 excuse.add_detailed_info(
1717 "Ignoring unsatisfiable Built-Using for %s/%s on %s %s"
1718 % (pkg_name, arch, bu_source, bu_version)
1719 )
1720 else:
1721 verdict = PolicyVerdict.worst_of(
1722 verdict, PolicyVerdict.REJECTED_PERMANENTLY
1723 )
1724 excuse.add_verdict_info(
1725 verdict,
1726 "%s/%s has unsatisfiable Built-Using on %s %s"
1727 % (pkg_name, arch, bu_source, bu_version),
1728 )
1730 return verdict
1733class BlockPolicy(AbstractBasePolicy):
1734 BLOCK_HINT_REGEX = re.compile("^(un)?(block-?.*)$")
1736 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
1737 super().__init__(
1738 "block",
1739 options,
1740 suite_info,
1741 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
1742 )
1743 self._blockall: dict[str | None, Hint] = {}
1745 def initialise(self, britney: "Britney") -> None:
1746 super().initialise(britney)
1747 assert self.hints is not None
1748 for hint in self.hints.search(type="block-all"):
1749 self._blockall[hint.package] = hint
1751 self._key_packages: frozenset[str] = frozenset()
1752 if "key" in self._blockall:
1753 self._key_packages = self._read_key_packages()
1755 def _read_key_packages(self) -> frozenset[str]:
1756 """Read the list of key packages
1758 The file contains data in the yaml format :
1760 - reason: <something>
1761 source: <package>
1763 The method returns a list of all key packages.
1764 """
1765 filename = os.path.join(self.state_dir, "key_packages.yaml")
1766 self.logger.info("Loading key packages from %s", filename)
1767 if os.path.exists(filename): 1767 ↛ 1772line 1767 didn't jump to line 1772 because the condition on line 1767 was always true
1768 with open(filename) as f:
1769 data = yaml.safe_load(f)
1770 key_packages = frozenset(item["source"] for item in data)
1771 else:
1772 self.logger.error(
1773 "Britney was asked to block key packages, "
1774 + "but no key_packages.yaml file was found."
1775 )
1776 sys.exit(1)
1778 return key_packages
1780 def register_hints(self, hint_parser: HintParser) -> None:
1781 # block related hints are currently defined in hint.py
1782 pass
1784 def _check_blocked(
1785 self, arch: str, version: str, excuse: "Excuse"
1786 ) -> PolicyVerdict:
1787 verdict = PolicyVerdict.PASS
1788 blocked = {}
1789 unblocked = {}
1790 block_info = {}
1791 source_suite = excuse.item.suite
1792 suite_name = source_suite.name
1793 src = excuse.item.package
1794 is_primary = source_suite.suite_class is SuiteClass.PRIMARY_SOURCE_SUITE
1796 tooltip = (
1797 f"please contact {self.options.distribution}-release if update is needed"
1798 )
1800 assert self.hints is not None
1801 mismatches = False
1802 r = self.BLOCK_HINT_REGEX
1803 for hint in self.hints.search(package=src):
1804 m = r.match(hint.type)
1805 if m:
1806 if m.group(1) == "un":
1807 assert hint.suite is not None
1808 if (
1809 hint.version != version
1810 or hint.suite.name != suite_name
1811 or (hint.architecture != arch and hint.architecture != "source")
1812 ):
1813 self.logger.info(
1814 "hint mismatch: %s %s %s", version, arch, suite_name
1815 )
1816 mismatches = True
1817 else:
1818 unblocked[m.group(2)] = hint.user
1819 excuse.add_hint(hint)
1820 else:
1821 # block(-*) hint: only accepts a source, so this will
1822 # always match
1823 blocked[m.group(2)] = hint.user
1824 excuse.add_hint(hint)
1826 if "block" not in blocked and is_primary:
1827 # if there is a specific block hint for this package, we don't
1828 # check for the general hints
1830 if self.options.distribution == "debian": 1830 ↛ 1834line 1830 didn't jump to line 1834 because the condition on line 1830 was always true
1831 url = "https://release.debian.org/testing/freeze_policy.html"
1832 tooltip = f'Follow the <a href="{url}">freeze policy</a> when applying for an unblock'
1834 if "source" in self._blockall:
1835 blocked["block"] = self._blockall["source"].user
1836 excuse.add_hint(self._blockall["source"])
1837 elif (
1838 "new-source" in self._blockall
1839 and src not in self.suite_info.target_suite.sources
1840 ):
1841 blocked["block"] = self._blockall["new-source"].user
1842 excuse.add_hint(self._blockall["new-source"])
1843 # no tooltip: new sources will probably not be accepted anyway
1844 block_info["block"] = "blocked by {}: is not in {}".format(
1845 self._blockall["new-source"].user,
1846 self.suite_info.target_suite.name,
1847 )
1848 elif "key" in self._blockall and src in self._key_packages:
1849 blocked["block"] = self._blockall["key"].user
1850 excuse.add_hint(self._blockall["key"])
1851 block_info["block"] = "blocked by {}: is a key package ({})".format(
1852 self._blockall["key"].user,
1853 tooltip,
1854 )
1855 elif "no-autopkgtest" in self._blockall:
1856 if excuse.autopkgtest_results == {"PASS"}:
1857 if not blocked: 1857 ↛ 1883line 1857 didn't jump to line 1883 because the condition on line 1857 was always true
1858 excuse.addinfo("not blocked: has successful autopkgtest")
1859 else:
1860 blocked["block"] = self._blockall["no-autopkgtest"].user
1861 excuse.add_hint(self._blockall["no-autopkgtest"])
1862 if not excuse.autopkgtest_results:
1863 block_info["block"] = (
1864 "blocked by %s: does not have autopkgtest (%s)"
1865 % (
1866 self._blockall["no-autopkgtest"].user,
1867 tooltip,
1868 )
1869 )
1870 else:
1871 block_info["block"] = (
1872 "blocked by %s: autopkgtest not fully successful (%s)"
1873 % (
1874 self._blockall["no-autopkgtest"].user,
1875 tooltip,
1876 )
1877 )
1879 elif not is_primary:
1880 blocked["block"] = suite_name
1881 excuse.needs_approval = True
1883 for block_cmd in blocked:
1884 unblock_cmd = "un" + block_cmd
1885 if block_cmd in unblocked:
1886 if is_primary or block_cmd == "block-udeb":
1887 excuse.addinfo(
1888 "Ignoring %s request by %s, due to %s request by %s"
1889 % (
1890 block_cmd,
1891 blocked[block_cmd],
1892 unblock_cmd,
1893 unblocked[block_cmd],
1894 )
1895 )
1896 else:
1897 excuse.addinfo("Approved by %s" % (unblocked[block_cmd]))
1898 else:
1899 verdict = PolicyVerdict.REJECTED_NEEDS_APPROVAL
1900 if is_primary or block_cmd == "block-udeb":
1901 # redirect people to d-i RM for udeb things:
1902 if block_cmd == "block-udeb":
1903 tooltip = "please contact the d-i release manager if an update is needed"
1904 if block_cmd in block_info:
1905 info = block_info[block_cmd]
1906 else:
1907 info = (
1908 "Not touching package due to {} request by {} ({})".format(
1909 block_cmd,
1910 blocked[block_cmd],
1911 tooltip,
1912 )
1913 )
1914 excuse.add_verdict_info(verdict, info)
1915 else:
1916 excuse.add_verdict_info(verdict, "NEEDS APPROVAL BY RM")
1917 excuse.addreason("block")
1918 if mismatches:
1919 excuse.add_detailed_info(
1920 f"Some hints for {src} do not match this item"
1921 )
1922 return verdict
1924 def apply_src_policy_impl(
1925 self,
1926 block_info: dict[str, Any],
1927 source_data_tdist: SourcePackage | None,
1928 source_data_srcdist: SourcePackage,
1929 excuse: "Excuse",
1930 ) -> PolicyVerdict:
1931 return self._check_blocked("source", source_data_srcdist.version, excuse)
1933 def apply_srcarch_policy_impl(
1934 self,
1935 block_info: dict[str, Any],
1936 arch: str,
1937 source_data_tdist: SourcePackage | None,
1938 source_data_srcdist: SourcePackage,
1939 excuse: "Excuse",
1940 ) -> PolicyVerdict:
1941 return self._check_blocked(arch, source_data_srcdist.version, excuse)
1944class BuiltOnBuilddPolicy(AbstractBasePolicy):
1946 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
1947 super().__init__(
1948 "builtonbuildd",
1949 options,
1950 suite_info,
1951 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
1952 ApplySrcPolicy.RUN_ON_EVERY_ARCH_ONLY,
1953 )
1954 self._signer_info: dict[str, Any] = {}
1956 def register_hints(self, hint_parser: HintParser) -> None:
1957 hint_parser.register_hint_type(
1958 HintType(
1959 "allow-archall-maintainer-upload",
1960 versioned=HintAnnotate.FORBIDDEN,
1961 )
1962 )
1964 def initialise(self, britney: "Britney") -> None:
1965 super().initialise(britney)
1966 try:
1967 filename_signerinfo = os.path.join(self.state_dir, "signers.json")
1968 except AttributeError as e: # pragma: no cover
1969 raise RuntimeError(
1970 "Please set STATE_DIR in the britney configuration"
1971 ) from e
1972 self._signer_info = self._read_signerinfo(filename_signerinfo)
1974 def apply_srcarch_policy_impl(
1975 self,
1976 buildd_info: dict[str, Any],
1977 arch: str,
1978 source_data_tdist: SourcePackage | None,
1979 source_data_srcdist: SourcePackage,
1980 excuse: "Excuse",
1981 ) -> PolicyVerdict:
1982 verdict = PolicyVerdict.PASS
1983 signers = self._signer_info
1985 if "signed-by" not in buildd_info:
1986 buildd_info["signed-by"] = {}
1988 item = excuse.item
1989 source_suite = item.suite
1991 # we use the source component, because a binary in contrib can
1992 # belong to a source in main
1993 component = get_component(source_data_srcdist.section)
1995 packages_s_a = source_suite.binaries[arch]
1996 assert self.hints is not None
1998 for pkg_id in sorted(
1999 x
2000 for x in filter_out_faux_gen(source_data_srcdist.binaries)
2001 if x.architecture == arch
2002 ):
2003 pkg_name = pkg_id.package_name
2004 binary_u = packages_s_a[pkg_name]
2005 pkg_arch = binary_u.architecture
2007 if binary_u.source_version != source_data_srcdist.version: 2007 ↛ 2008line 2007 didn't jump to line 2008 because the condition on line 2007 was never true
2008 continue
2010 if item.architecture != "source" and pkg_arch == "all":
2011 # we don't care about the existing arch: all binaries when
2012 # checking a binNMU item, because the arch: all binaries won't
2013 # migrate anyway
2014 continue
2016 signer = None
2017 uid = None
2018 uidinfo = ""
2019 buildd_ok = False
2020 failure_verdict = PolicyVerdict.REJECTED_PERMANENTLY
2021 try:
2022 signer = signers[pkg_name][pkg_id.version][pkg_arch]
2023 if signer["buildd"]:
2024 buildd_ok = True
2025 uid = signer["uid"]
2026 uidinfo = f"arch {pkg_arch} binaries uploaded by {uid}"
2027 except KeyError:
2028 self.logger.info(
2029 "signer info for %s %s (%s) on %s not found",
2030 pkg_name,
2031 binary_u.version,
2032 pkg_arch,
2033 arch,
2034 )
2035 uidinfo = f"upload info for arch {pkg_arch} binaries not found"
2036 failure_verdict = PolicyVerdict.REJECTED_CANNOT_DETERMINE_IF_PERMANENT
2037 if not buildd_ok:
2038 if component != "main":
2039 if not buildd_ok and pkg_arch not in buildd_info["signed-by"]: 2039 ↛ 2043line 2039 didn't jump to line 2043 because the condition on line 2039 was always true
2040 excuse.add_detailed_info(
2041 f"{uidinfo}, but package in {component}"
2042 )
2043 buildd_ok = True
2044 elif pkg_arch == "all":
2045 if (
2046 allow_hint := self.hints.search_first(
2047 "allow-archall-maintainer-upload", package=item.package
2048 )
2049 ) is not None:
2050 buildd_ok = True
2051 verdict = PolicyVerdict.worst_of(
2052 verdict, PolicyVerdict.PASS_HINTED
2053 )
2054 if pkg_arch not in buildd_info["signed-by"]:
2055 excuse.addinfo(
2056 f"{uidinfo}, but whitelisted by {allow_hint.user}"
2057 )
2058 if not buildd_ok:
2059 verdict = failure_verdict
2060 if pkg_arch not in buildd_info["signed-by"]:
2061 if pkg_arch == "all":
2062 uidinfo += (
2063 ", a new source-only upload is needed to allow migration"
2064 )
2065 excuse.add_verdict_info(verdict, f"Not built on buildd: {uidinfo}")
2067 if ( 2067 ↛ 2071line 2067 didn't jump to line 2071
2068 pkg_arch in buildd_info["signed-by"]
2069 and buildd_info["signed-by"][pkg_arch] != uid
2070 ):
2071 self.logger.info(
2072 "signer mismatch for %s (%s %s) on %s: %s, while %s already listed",
2073 pkg_name,
2074 binary_u.source,
2075 binary_u.source_version,
2076 pkg_arch,
2077 uid,
2078 buildd_info["signed-by"][pkg_arch],
2079 )
2081 buildd_info["signed-by"][pkg_arch] = uid
2083 return verdict
2085 def _read_signerinfo(self, filename: str) -> dict[str, Any]:
2086 signerinfo: dict[str, Any] = {}
2087 self.logger.info("Loading signer info from %s", filename)
2088 with open(filename) as fd: 2088 ↛ exitline 2088 didn't return from function '_read_signerinfo' because the return on line 2090 wasn't executed
2089 if os.fstat(fd.fileno()).st_size < 1: 2089 ↛ 2090line 2089 didn't jump to line 2090 because the condition on line 2089 was never true
2090 return signerinfo
2091 signerinfo = json.load(fd)
2093 return signerinfo
2096class ImplicitDependencyPolicy(AbstractBasePolicy):
2097 """Implicit Dependency policy
2099 Upgrading a package pkg-a can break the installability of a package pkg-b.
2100 A newer version (or the removal) of pkg-b might fix the issue. In that
2101 case, pkg-a has an 'implicit dependency' on pkg-b, because pkg-a can only
2102 migrate if pkg-b also migrates.
2104 This policy tries to discover a few common cases, and adds the relevant
2105 info to the excuses. If another item is needed to fix the
2106 uninstallability, a dependency is added. If no newer item can fix it, this
2107 excuse will be blocked.
2109 Note that the migration step will check the installability of every
2110 package, so this policy doesn't need to handle every corner case. It
2111 must, however, make sure that no excuse is unnecessarily blocked.
2113 Some cases that should be detected by this policy:
2115 * pkg-a is upgraded from 1.0-1 to 2.0-1, while
2116 pkg-b has "Depends: pkg-a (<< 2.0)"
2117 This typically happens if pkg-b has a strict dependency on pkg-a because
2118 it uses some non-stable internal interface (examples are glibc,
2119 binutils, python3-defaults, ...)
2121 * pkg-a is upgraded from 1.0-1 to 2.0-1, and
2122 pkg-a 1.0-1 has "Provides: provides-1",
2123 pkg-a 2.0-1 has "Provides: provides-2",
2124 pkg-b has "Depends: provides-1"
2125 This typically happens when pkg-a has an interface that changes between
2126 versions, and a virtual package is used to identify the version of this
2127 interface (e.g. perl-api-x.y)
2129 """
2131 _pkg_universe: "BinaryPackageUniverse"
2132 _all_binaries: dict["BinaryPackageId", "BinaryPackage"]
2133 _allow_uninst: dict[str, set[str | None]]
2134 _nobreakall_arches: list[str]
2136 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
2137 super().__init__(
2138 "implicit-deps",
2139 options,
2140 suite_info,
2141 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
2142 ApplySrcPolicy.RUN_ON_EVERY_ARCH_ONLY,
2143 )
2145 def initialise(self, britney: "Britney") -> None:
2146 super().initialise(britney)
2147 self._pkg_universe = britney.pkg_universe
2148 self._all_binaries = britney.all_binaries
2149 self._smooth_updates = britney.options.smooth_updates
2150 self._nobreakall_arches = self.options.nobreakall_arches
2151 self._new_arches = self.options.new_arches
2152 self._break_arches = self.options.break_arches
2153 self._allow_uninst = britney.allow_uninst
2154 self._outofsync_arches = self.options.outofsync_arches
2156 def can_be_removed(self, pkg: BinaryPackage) -> bool:
2157 src = pkg.source
2158 target_suite = self.suite_info.target_suite
2160 # TODO these conditions shouldn't be hardcoded here
2161 # ideally, we would be able to look up excuses to see if the removal
2162 # is in there, but in the current flow, this policy is called before
2163 # all possible excuses exist, so there is no list for us to check
2165 if src not in self.suite_info.primary_source_suite.sources:
2166 # source for pkg not in unstable: candidate for removal
2167 return True
2169 source_t = target_suite.sources[src]
2170 assert self.hints is not None
2171 if self.hints.has_hint("remove", package=src, version=source_t.version):
2172 # removal hint for the source in testing: candidate for removal
2173 return True
2175 if target_suite.is_cruft(pkg):
2176 # if pkg is cruft in testing, removal will be tried
2177 return True
2179 # the case were the newer version of the source no longer includes the
2180 # binary (or includes a cruft version of the binary) will be handled
2181 # separately (in that case there might be an implicit dependency on
2182 # the newer source)
2184 return False
2186 def should_skip_rdep(
2187 self, pkg: BinaryPackage, source_name: str, myarch: str
2188 ) -> bool:
2189 target_suite = self.suite_info.target_suite
2191 if not target_suite.is_pkg_in_the_suite(pkg.pkg_id):
2192 # it is not in the target suite, migration cannot break anything
2193 return True
2195 if pkg.source == source_name:
2196 # if it is built from the same source, it will be upgraded
2197 # with the source
2198 return True
2200 if self.can_be_removed(pkg):
2201 # could potentially be removed, so if that happens, it won't be
2202 # broken
2203 return True
2205 if pkg.architecture == "all" and myarch not in self._nobreakall_arches:
2206 # arch all on non nobreakarch is allowed to become uninstallable
2207 return True
2209 if pkg.pkg_id.package_name in self._allow_uninst[myarch]:
2210 # there is a hint to allow this binary to become uninstallable
2211 return True
2213 if not target_suite.is_installable(pkg.pkg_id):
2214 # it is already uninstallable in the target suite, migration
2215 # cannot break anything
2216 return True
2218 return False
2220 def breaks_installability(
2221 self,
2222 pkg_id_t: BinaryPackageId,
2223 pkg_id_s: BinaryPackageId | None,
2224 pkg_to_check: BinaryPackageId,
2225 ) -> bool:
2226 """
2227 Check if upgrading pkg_id_t to pkg_id_s breaks the installability of
2228 pkg_to_check.
2230 To check if removing pkg_id_t breaks pkg_to_check, set pkg_id_s to
2231 None.
2232 """
2234 pkg_universe = self._pkg_universe
2235 negative_deps = pkg_universe.negative_dependencies_of(pkg_to_check)
2237 for dep in pkg_universe.dependencies_of(pkg_to_check):
2238 if pkg_id_t not in dep:
2239 # this depends doesn't have pkg_id_t as alternative, so
2240 # upgrading pkg_id_t cannot break this dependency clause
2241 continue
2243 # We check all the alternatives for this dependency, to find one
2244 # that can satisfy it when pkg_id_t is upgraded to pkg_id_s
2245 found_alternative = False
2246 for d in dep:
2247 if d in negative_deps:
2248 # If this alternative dependency conflicts with
2249 # pkg_to_check, it cannot be used to satisfy the
2250 # dependency.
2251 # This commonly happens when breaks are added to pkg_id_s.
2252 continue
2254 if d.package_name != pkg_id_t.package_name:
2255 # a binary different from pkg_id_t can satisfy the dep, so
2256 # upgrading pkg_id_t won't break this dependency
2257 found_alternative = True
2258 break
2260 if d != pkg_id_s:
2261 # We want to know the impact of the upgrade of
2262 # pkg_id_t to pkg_id_s. If pkg_id_s migrates to the
2263 # target suite, any other version of this binary will
2264 # not be there, so it cannot satisfy this dependency.
2265 # This includes pkg_id_t, but also other versions.
2266 continue
2268 # pkg_id_s can satisfy the dep
2269 found_alternative = True
2270 break
2272 if not found_alternative:
2273 return True
2274 return False
2276 def check_upgrade(
2277 self,
2278 pkg_id_t: BinaryPackageId,
2279 pkg_id_s: BinaryPackageId | None,
2280 source_name: str,
2281 myarch: str,
2282 broken_binaries: set[str],
2283 excuse: "Excuse",
2284 ) -> PolicyVerdict:
2285 verdict = PolicyVerdict.PASS
2287 pkg_universe = self._pkg_universe
2288 all_binaries = self._all_binaries
2290 # check all rdeps of the package in testing
2291 rdeps_t = pkg_universe.reverse_dependencies_of(pkg_id_t)
2293 for rdep_pkg in sorted(rdeps_t):
2294 rdep_p = all_binaries[rdep_pkg]
2296 # check some cases where the rdep won't become uninstallable, or
2297 # where we don't care if it does
2298 if self.should_skip_rdep(rdep_p, source_name, myarch):
2299 continue
2301 if not self.breaks_installability(pkg_id_t, pkg_id_s, rdep_pkg):
2302 # if upgrading pkg_id_t to pkg_id_s doesn't break rdep_pkg,
2303 # there is no implicit dependency
2304 continue
2306 # The upgrade breaks the installability of the rdep. We need to
2307 # find out if there is a newer version of the rdep that solves the
2308 # uninstallability. If that is the case, there is an implicit
2309 # dependency. If not, the upgrade will fail.
2311 # check source versions
2312 good_newer_versions = set()
2313 for npkg, suite in find_newer_binaries(
2314 self.suite_info, rdep_p, add_source_for_dropped_bin=True
2315 ):
2316 if npkg.architecture == "source":
2317 # When a newer version of the source package doesn't have
2318 # the binary, we get the source as 'newer version'. In
2319 # this case, the binary will not be uninstallable if the
2320 # newer source migrates, because it is no longer there.
2321 good_newer_versions.add(npkg)
2322 continue
2323 assert isinstance(npkg, BinaryPackageId)
2324 if not self.breaks_installability(pkg_id_t, pkg_id_s, npkg):
2325 good_newer_versions.add(npkg)
2327 if good_newer_versions:
2328 spec = DependencySpec(DependencyType.IMPLICIT_DEPENDENCY, myarch)
2329 excuse.add_package_depends(spec, good_newer_versions)
2330 else:
2331 # no good newer versions: no possible solution
2332 broken_binaries.add(rdep_pkg.name)
2333 if pkg_id_s:
2334 action = "migrating {} to {}".format(
2335 pkg_id_s.name,
2336 self.suite_info.target_suite.name,
2337 )
2338 else:
2339 action = "removing {} from {}".format(
2340 pkg_id_t.name,
2341 self.suite_info.target_suite.name,
2342 )
2343 if rdep_pkg.package_name.endswith("-faux-build-depends"):
2344 name = rdep_pkg.package_name.removesuffix("-faux-build-depends")
2345 info = f'{action} makes Build-Depends of src:<a href="#{name}">{name}</a> uninstallable'
2346 else:
2347 info = '{0} makes <a href="#{1}">{1}</a> uninstallable'.format(
2348 action, rdep_pkg.name
2349 )
2350 verdict = PolicyVerdict.REJECTED_PERMANENTLY
2351 excuse.add_verdict_info(verdict, info)
2353 return verdict
2355 def apply_srcarch_policy_impl(
2356 self,
2357 implicit_dep_info: dict[str, Any],
2358 arch: str,
2359 source_data_tdist: SourcePackage | None,
2360 source_data_srcdist: SourcePackage,
2361 excuse: "Excuse",
2362 ) -> PolicyVerdict:
2363 verdict = PolicyVerdict.PASS
2365 if not source_data_tdist:
2366 # this item is not currently in testing: no implicit dependency
2367 return verdict
2369 if excuse.hasreason("missingbuild"):
2370 # if the build is missing, the policy would treat this as if the
2371 # binaries would be removed, which would give incorrect (and
2372 # confusing) info
2373 info = "missing build, not checking implicit dependencies on %s" % (arch)
2374 excuse.add_detailed_info(info)
2375 return verdict
2377 source_suite = excuse.item.suite
2378 source_name = excuse.item.package
2379 target_suite = self.suite_info.target_suite
2380 all_binaries = self._all_binaries
2382 # we check all binaries for this excuse that are currently in testing
2383 relevant_binaries = sorted(
2384 x
2385 for x in source_data_tdist.binaries
2386 if (arch == "source" or x.architecture == arch)
2387 and x.package_name in target_suite.binaries[x.architecture]
2388 and x.architecture not in self._new_arches
2389 and x.architecture not in self._break_arches
2390 and x.architecture not in self._outofsync_arches
2391 )
2393 broken_binaries: set[str] = set()
2395 assert self.hints is not None
2396 for pkg_id_t in relevant_binaries:
2397 mypkg = pkg_id_t.package_name
2398 myarch = pkg_id_t.architecture
2399 binaries_t_a = target_suite.binaries[myarch]
2400 binaries_s_a = source_suite.binaries[myarch]
2402 if target_suite.is_cruft(all_binaries[pkg_id_t]):
2403 # this binary is cruft in testing: it will stay around as long
2404 # as necessary to satisfy dependencies, so we don't need to
2405 # care
2406 continue
2408 if mypkg in binaries_s_a:
2409 mybin = binaries_s_a[mypkg]
2410 pkg_id_s: Optional["BinaryPackageId"] = mybin.pkg_id
2411 if mybin.source != source_name:
2412 # hijack: this is too complicated to check, so we ignore
2413 # it (the migration code will check the installability
2414 # later anyway)
2415 pass
2416 elif mybin.source_version != source_data_srcdist.version:
2417 # cruft in source suite: pretend the binary doesn't exist
2418 pkg_id_s = None
2419 elif pkg_id_t == pkg_id_s:
2420 # same binary (probably arch: all from a binNMU):
2421 # 'upgrading' doesn't change anything, for this binary, so
2422 # it won't break anything
2423 continue
2424 else:
2425 pkg_id_s = None
2427 if not pkg_id_s and is_smooth_update_allowed(
2428 binaries_t_a[mypkg], self._smooth_updates, self.hints
2429 ):
2430 # the binary isn't in the new version (or is cruft there), and
2431 # smooth updates are allowed: the binary can stay around if
2432 # that is necessary to satisfy dependencies, so we don't need
2433 # to check it
2434 continue
2436 if (
2437 not pkg_id_s
2438 and source_data_tdist.version == source_data_srcdist.version
2439 and source_suite.suite_class is SuiteClass.ADDITIONAL_SOURCE_SUITE
2440 and binaries_t_a[mypkg].architecture == "all"
2441 ):
2442 # we're very probably migrating a binNMU built in tpu where the arch:all
2443 # binaries were not copied to it as that's not needed. This policy could
2444 # needlessly block.
2445 continue
2447 v = self.check_upgrade(
2448 pkg_id_t, pkg_id_s, source_name, myarch, broken_binaries, excuse
2449 )
2450 verdict = PolicyVerdict.worst_of(verdict, v)
2452 # each arch is processed separately, so if we already have info from
2453 # other archs, we need to merge the info from this arch
2454 broken_old = set(implicit_dep_info.get("broken-binaries", []))
2455 implicit_dep_info["broken-binaries"] = sorted(broken_old | broken_binaries)
2457 return verdict
2460class ReverseRemovalPolicy(AbstractBasePolicy):
2461 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
2462 super().__init__(
2463 "reverseremoval",
2464 options,
2465 suite_info,
2466 {SuiteClass.PRIMARY_SOURCE_SUITE, SuiteClass.ADDITIONAL_SOURCE_SUITE},
2467 )
2469 def register_hints(self, hint_parser: HintParser) -> None:
2470 hint_parser.register_hint_type(HintType("ignore-reverse-remove"))
2472 def initialise(self, britney: "Britney") -> None:
2473 super().initialise(britney)
2475 pkg_universe = britney.pkg_universe
2476 source_suites = britney.suite_info.source_suites
2477 target_suite = britney.suite_info.target_suite
2479 # Build set of the sources of reverse (Build-) Depends
2480 assert self.hints is not None
2482 rev_bin: dict[BinaryPackageId, set[str]] = defaultdict(set)
2483 for hint in self.hints.search("remove"):
2484 for item in hint.packages:
2485 # I think we don't need to look at the target suite
2486 for src_suite in source_suites:
2487 try:
2488 # Explicitly not running filter_out_faux here
2489 my_bins = set(src_suite.sources[item.uvname].binaries)
2490 except KeyError:
2491 continue
2492 compute_reverse_tree(pkg_universe, my_bins)
2493 for this_bin in my_bins:
2494 rev_bin.setdefault(this_bin, set()).add(item.uvname)
2496 rev_src: dict[str, set[str]] = defaultdict(set)
2497 for bin_pkg, reasons in rev_bin.items():
2498 # If the pkg is in the target suite, there's nothing this
2499 # policy wants to do.
2500 if target_suite.is_pkg_in_the_suite(bin_pkg):
2501 continue
2502 that_bin = britney.all_binaries[bin_pkg]
2503 bin_src = that_bin.source + "/" + that_bin.source_version
2504 rev_src.setdefault(bin_src, set()).update(reasons)
2505 self._block_src_for_rm_hint = rev_src
2507 def apply_src_policy_impl(
2508 self,
2509 rev_remove_info: dict[str, Any],
2510 source_data_tdist: SourcePackage | None,
2511 source_data_srcdist: SourcePackage,
2512 excuse: "Excuse",
2513 ) -> PolicyVerdict:
2514 verdict = PolicyVerdict.PASS
2516 item = excuse.item
2517 if item.name in self._block_src_for_rm_hint:
2518 reason = ", ".join(sorted(self._block_src_for_rm_hint[item.name]))
2519 assert self.hints is not None
2520 excuse.addreason("reverseremoval")
2521 if (
2522 ignore_hint := self.hints.search_first(
2523 "ignore-reverse-remove", package=item.uvname, version=item.version
2524 )
2525 ) is not None:
2526 excuse.addreason("ignore-reverse-remove")
2527 excuse.addinfo(
2528 "Should block migration because of remove hint for %s, but forced by %s"
2529 % (reason, ignore_hint.user)
2530 )
2531 verdict = PolicyVerdict.PASS_HINTED
2532 else:
2533 verdict = PolicyVerdict.REJECTED_PERMANENTLY
2534 excuse.add_verdict_info(
2535 verdict, "Remove hint for (transitive) dependency: %s" % reason
2536 )
2538 return verdict
2541class ReproducibleState(Enum):
2542 BAD = auto()
2543 FAIL = auto()
2544 GOOD = auto()
2545 UNKNOWN = auto()
2547 @staticmethod
2548 def from_str(val: str | None) -> "ReproducibleState":
2549 match val:
2550 case "BAD":
2551 return ReproducibleState.BAD
2552 case "FAIL": 2552 ↛ 2553line 2552 didn't jump to line 2553 because the pattern on line 2552 never matched
2553 return ReproducibleState.FAIL
2554 case "GOOD":
2555 return ReproducibleState.GOOD
2556 case "UNKNOWN" | None: 2556 ↛ 2558line 2556 didn't jump to line 2558 because the pattern on line 2556 always matched
2557 return ReproducibleState.UNKNOWN
2558 case _:
2559 raise ValueError(f"Invalid reproducability state f{str}")
2562@dataclass(slots=True, frozen=True)
2563class ReproducibleData:
2564 state: ReproducibleState
2565 build_id: str | None = field(default=None, kw_only=True)
2566 diffoscope_log_id: str | None = field(default=None, kw_only=True)
2567 artifact_id: str | None = field(default=None, kw_only=True)
2570class ReproduciblePolicy(AbstractBasePolicy):
2571 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
2572 super().__init__(
2573 "reproducible",
2574 options,
2575 suite_info,
2576 {SuiteClass.PRIMARY_SOURCE_SUITE},
2577 ApplySrcPolicy.RUN_ON_EVERY_ARCH_ONLY,
2578 )
2579 self._reproducible: dict[str, dict[tuple[str, str], ReproducibleData]] = {}
2580 self._components: tuple[str, ...] = ()
2582 # Default values for this policy's options
2583 parse_option(options, "repro_success_bounty", default=0, to_int=True)
2584 parse_option(options, "repro_regression_penalty", default=0, to_int=True)
2585 parse_option(options, "repro_log_url")
2586 parse_option(options, "repro_excuse_url")
2587 parse_option(options, "repro_retry_url")
2588 parse_option(options, "repro_components")
2590 def register_hints(self, hint_parser: HintParser) -> None:
2591 hint_parser.register_hint_type(
2592 HintType(
2593 "ignore-reproducible-src",
2594 versioned=HintAnnotate.OPTIONAL,
2595 architectured=HintAnnotate.OPTIONAL,
2596 )
2597 )
2598 hint_parser.register_hint_type(
2599 HintType(
2600 "ignore-reproducible",
2601 versioned=HintAnnotate.OPTIONAL,
2602 architectured=HintAnnotate.OPTIONAL,
2603 )
2604 )
2606 def initialise(self, britney: "Britney") -> None:
2607 super().initialise(britney)
2608 summary = self._reproducible
2610 valid_release_names = {
2611 suite.codename
2612 for suite in chain(
2613 (britney.suite_info.target_suite,),
2614 britney.suite_info.source_suites,
2615 )
2616 } | {
2617 suite.name
2618 for suite in chain(
2619 (britney.suite_info.target_suite,),
2620 britney.suite_info.source_suites,
2621 )
2622 }
2624 assert hasattr(
2625 self, "state_dir"
2626 ), "Please set STATE_DIR in the britney configuration"
2627 assert (
2628 self.options.repro_components
2629 ), "Please set REPRO_COMPONENTS in the britney configuration"
2630 self._components = tuple(self.options.repro_components.split())
2632 for file in os.listdir(self.state_dir):
2633 if not file.startswith("reproducible-") or not file.endswith(".json"):
2634 continue
2635 filename = os.path.join(self.state_dir, file)
2637 self.logger.info("Loading reproducibility report from %s", filename)
2638 with open(filename) as fd:
2639 if os.fstat(fd.fileno()).st_size < 1:
2640 continue
2641 data = json.load(fd)
2643 for result in data["records"]:
2644 if ( 2644 ↛ 2648line 2644 didn't jump to line 2648 because the condition on line 2644 was never true
2645 release := result.get("release")
2646 ) is not None and release not in valid_release_names:
2647 # tests do not have a release set
2648 continue
2650 state = ReproducibleState.from_str(result.get("status"))
2651 repo = {
2652 key: value
2653 for key, value in result.items()
2654 if key in ("build_id", "diffoscope_log_id", "artifact_id")
2655 }
2657 summary.setdefault(result["architecture"], {})[
2658 (result["name"], result["version"])
2659 ] = ReproducibleData(state, **repo)
2661 def _lookup_data(
2662 self, package_name: str, version: str, arch: str
2663 ) -> tuple[ReproducibleData, str] | None:
2664 key = (package_name, version)
2665 if (repo := self._reproducible[arch].get(key)) is not None:
2666 return repo, arch
2668 repo = self._reproducible["all"].get(key)
2669 return (repo, "all") if repo is not None else None
2671 def _format_link(self, bpid: BinaryPackageId, arch: str) -> str:
2672 data = self._lookup_data(bpid.package_name, bpid.version, arch)
2673 assert data is not None
2674 repo, arch = data
2675 if repo.diffoscope_log_id and (diff_id := repo.artifact_id): 2675 ↛ 2676line 2675 didn't jump to line 2676 because the condition on line 2675 was never true
2676 endpoint = f"artifacts/{diff_id}/diffoscope"
2677 else:
2678 endpoint = "log"
2679 url = self.options.repro_log_url.format(
2680 arch=arch, build_id=repo.build_id, endpoint=endpoint
2681 )
2682 return f'<a href="{url}">{bpid.package_name}</a>'
2684 def _create_link_to_log(self, arch: str, failed_bpids: set[BinaryPackageId]) -> str:
2685 if not self.options.repro_log_url: 2685 ↛ 2686line 2685 didn't jump to line 2686 because the condition on line 2685 was never true
2686 return ": " + ", ".join(bpid.package_name for bpid in sorted(failed_bpids))
2688 return ": " + ", ".join(
2689 self._format_link(bpid, arch) for bpid in sorted(failed_bpids)
2690 )
2692 def apply_srcarch_policy_impl(
2693 self,
2694 policy_info: dict[str, Any],
2695 arch: str,
2696 source_data_tdist: SourcePackage | None,
2697 source_data_srcdist: SourcePackage,
2698 excuse: "Excuse",
2699 ) -> PolicyVerdict:
2700 verdict = PolicyVerdict.PASS
2701 eligible_for_bounty = False
2702 all_hints = []
2704 assert self.hints is not None # Needed for type checking / mypy
2706 # we don't want to apply this policy (yet) on binNMUs
2707 if excuse.item.architecture != "source": 2707 ↛ 2708line 2707 didn't jump to line 2708 because the condition on line 2707 was never true
2708 return verdict
2710 # we're not supposed to judge on this arch
2711 if arch not in self.options.repro_arches: 2711 ↛ 2712line 2711 didn't jump to line 2712 because the condition on line 2711 was never true
2712 return verdict
2714 # bail out if this arch has no packages for this source (not build
2715 # here)
2716 if arch not in excuse.packages: 2716 ↛ 2717line 2716 didn't jump to line 2717 because the condition on line 2716 was never true
2717 return verdict
2719 component = get_component(source_data_srcdist.section)
2721 if self._components and component not in self._components: 2721 ↛ 2722line 2721 didn't jump to line 2722 because the condition on line 2721 was never true
2722 self.logger.debug(
2723 "%s skipping reproducible policy, component %s not requested",
2724 excuse.name,
2725 component,
2726 )
2727 return verdict
2729 source_name = excuse.item.package
2731 if self.options.repro_excuse_url:
2732 url = self.options.repro_excuse_url.format(
2733 package=quote(source_name), arch=arch
2734 )
2735 url_html = ' - <a href="%s">info</a>' % url
2736 # When run on multiple archs, the last one "wins"
2737 policy_info["status-url"] = url
2738 else:
2739 url = None
2740 url_html = ""
2742 if arch not in self._reproducible: 2742 ↛ 2743line 2742 didn't jump to line 2743 because the condition on line 2742 was never true
2743 verdict = PolicyVerdict.REJECTED_TEMPORARILY
2744 msg = f"No reproducibility data available at all for {arch}"
2745 excuse.add_verdict_info(verdict, msg)
2746 return verdict
2747 if "all" not in self._reproducible: 2747 ↛ 2748line 2747 didn't jump to line 2748 because the condition on line 2747 was never true
2748 verdict = PolicyVerdict.REJECTED_TEMPORARILY
2749 msg = "No reproducibility data available at all for arch:all"
2750 excuse.add_verdict_info(verdict, msg)
2751 return verdict
2753 # skip/delay policy until both arch:arch and arch:all builds are done
2754 if (arch or "all") in excuse.missing_builds: 2754 ↛ 2755line 2754 didn't jump to line 2755 because the condition on line 2754 was never true
2755 self.logger.debug(
2756 "%s not built for %s or all, skipping reproducible policy",
2757 excuse.name,
2758 arch,
2759 )
2760 verdict = PolicyVerdict.REJECTED_TEMPORARILY
2761 excuse.add_verdict_info(
2762 verdict,
2763 f"Reproducibility check deferred on {arch}: missing builds{url_html}",
2764 )
2765 return verdict
2767 source_suite_state = "not-unknown"
2768 failed_bpids: set[BinaryPackageId] = set()
2769 # The states should either be GOOD/BAD for all binaries, UNKNOWN for all
2770 # binaries, or missing for all binaries, but let's not assume that.
2771 # They can be from different components after all.
2772 bins_src, src_suite_name = binaries_from_source_version(
2773 source_data_srcdist, self.suite_info
2774 )
2775 for bpid in bins_src:
2776 if bpid.architecture not in ("all", arch): 2776 ↛ 2777line 2776 didn't jump to line 2777 because the condition on line 2776 was never true
2777 continue
2778 in_component = True
2779 for suite in self.suite_info.source_suites:
2780 if suite.name == src_suite_name and (
2781 (
2782 component := get_component(
2783 suite.all_binaries_in_suite[bpid].section
2784 )
2785 )
2786 not in self._components
2787 ):
2788 self.logger.debug(
2789 "repro check for %s skipped due to component %s",
2790 bpid,
2791 component,
2792 )
2793 in_component = False
2794 break
2795 if not in_component:
2796 # TODO: should we update the excuses text?
2797 continue
2799 if (
2800 data := self._lookup_data(bpid.package_name, bpid.version, arch)
2801 ) is not None:
2802 pkg_info, _ = data
2803 self.logger.debug("repro data for %s: %s", bpid, pkg_info.state)
2804 if pkg_info.state is ReproducibleState.BAD:
2805 failed_bpids.add(bpid)
2806 # not changing source_suite_state here on purpose
2807 elif ( 2807 ↛ 2811line 2807 didn't jump to line 2811
2808 pkg_info.state is ReproducibleState.FAIL
2809 or pkg_info.state is ReproducibleState.UNKNOWN
2810 ):
2811 source_suite_state = "unknown"
2812 else:
2813 self.logger.debug("No repro data found for %s", bpid)
2814 # but maybe it's hinted (e.g. at the time of writing
2815 # reproduce.debian.net has a bug where udebs go missing)
2816 if (
2817 bpid_hints := self.hints.search_first(
2818 "ignore-reproducible",
2819 package=bpid.package_name,
2820 version=bpid.version,
2821 architecture=bpid.architecture,
2822 )
2823 ) is not None:
2824 all_hints.append(bpid_hints)
2825 self.logger.debug("repro: hint found for %s: %s", source_name, bpid)
2826 else:
2827 source_suite_state = "unknown"
2828 break
2830 if source_suite_state == "not-unknown":
2831 source_suite_state = "known"
2833 excuse_info = []
2834 if source_suite_state == "unknown":
2835 verdict = PolicyVerdict.REJECTED_TEMPORARILY
2836 excuse_info.append(
2837 f"Reproducibility check waiting for results on {arch}{url_html}"
2838 )
2839 policy_info.setdefault("state", {}).setdefault(arch, "unavailable")
2840 elif failed_bpids:
2841 ignored_bpids: set[BinaryPackageId] = set()
2842 if source_data_tdist is None: 2842 ↛ 2843line 2842 didn't jump to line 2843 because the condition on line 2842 was never true
2843 target_suite_state = "new"
2844 else:
2845 target_suite_state = "reproducible"
2846 for bpid in failed_bpids:
2847 pkg_name = bpid.package_name
2848 for bpid_t in filter_out_faux_gen(source_data_tdist.binaries):
2849 if bpid_t.architecture not in ("all", arch): 2849 ↛ 2850line 2849 didn't jump to line 2850 because the condition on line 2849 was never true
2850 continue
2851 if pkg_name != bpid_t.package_name: 2851 ↛ 2852line 2851 didn't jump to line 2852 because the condition on line 2851 was never true
2852 continue
2853 if ( 2853 ↛ 2868line 2853 didn't jump to line 2868 because the condition on line 2853 was always true
2854 data := self._lookup_data(pkg_name, bpid_t.version, arch)
2855 ) is not None:
2856 pkg_info, _ = data
2857 self.logger.debug(
2858 "testing repro data for %s: %s", bpid_t, pkg_info.state
2859 )
2860 if pkg_info.state is ReproducibleState.BAD:
2861 ignored_bpids.add(bpid)
2862 elif ( 2862 ↛ 2866line 2862 didn't jump to line 2866
2863 pkg_info.state is ReproducibleState.FAIL
2864 or pkg_info.state is ReproducibleState.UNKNOWN
2865 ):
2866 target_suite_state = "unknown"
2867 else:
2868 self.logger.debug(
2869 "No testing repro data found for %s", bpid_t
2870 )
2871 # This shouldn't happen as for the past migration
2872 # to have been allowed, there should be data.
2873 target_suite_state = "unknown"
2874 break
2876 # Reminder: code here is part of the non-reproducibile source-suite branch
2877 if target_suite_state == "new": 2877 ↛ 2878line 2877 didn't jump to line 2878 because the condition on line 2877 was never true
2878 verdict = PolicyVerdict.REJECTED_PERMANENTLY
2879 excuse_info.append(
2880 f"New but not reproduced on {arch}{url_html}"
2881 f"{self._create_link_to_log(arch, failed_bpids)}"
2882 )
2883 policy_info.setdefault("state", {}).setdefault(
2884 arch, "new but not reproducible"
2885 )
2886 elif target_suite_state == "unknown": 2886 ↛ 2888line 2886 didn't jump to line 2888 because the condition on line 2886 was never true
2887 # Shouldn't happen after initial bootstrap once blocking
2888 verdict = PolicyVerdict.REJECTED_TEMPORARILY
2889 excuse_info.append(
2890 f"Reproducibility check failed and now waiting for reference "
2891 f"results on {arch}{url_html}"
2892 f"{self._create_link_to_log(arch, failed_bpids)}"
2893 )
2894 policy_info.setdefault("state", {}).setdefault(
2895 arch, "waiting for reference"
2896 )
2897 elif failed_bpids <= ignored_bpids:
2898 # For the forseeable future we want to prevent regressions, one day
2899 # we might want to even block these.
2900 # verdict = PolicyVerdict.REJECTED_PERMANENTLY
2901 excuse_info.append(
2902 f"Not reproduced on {arch} (not a regression)"
2903 f"{self._create_link_to_log(arch, failed_bpids)}"
2904 )
2905 policy_info.setdefault("state", {}).setdefault(arch, "not reproducible")
2906 else:
2907 verdict = PolicyVerdict.REJECTED_PERMANENTLY
2908 excuse_info.append(
2909 f"Reproducibility regression on {arch}"
2910 f"{self._create_link_to_log(arch, failed_bpids - ignored_bpids)}"
2911 )
2912 policy_info.setdefault("state", {}).setdefault(arch, "regression")
2914 # non-reproducible source-suite cases are handled above, so here we
2915 # handle the last of the source-suite cases
2916 else:
2917 excuse_info.append(f"Reproduced on {arch}{url_html}")
2918 policy_info.setdefault("state", {}).setdefault(arch, "reproducible")
2919 eligible_for_bounty = True
2921 if verdict.is_rejected:
2922 for hint_arch in ("source", arch):
2923 if (
2924 ignore_hint := self.hints.search_first(
2925 "ignore-reproducible-src",
2926 package=source_name,
2927 version=source_data_srcdist.version,
2928 architecture=hint_arch,
2929 )
2930 ) is not None:
2931 # one hint is enough, take the first one encountered
2932 verdict = PolicyVerdict.PASS_HINTED
2933 policy_info.setdefault("hints", {}).setdefault(arch, []).append(
2934 f"{ignore_hint.user}: {str(ignore_hint)}"
2935 )
2936 if hint_arch == arch: 2936 ↛ 2939line 2936 didn't jump to line 2939 because the condition on line 2936 was always true
2937 on_arch = f" on {arch}"
2938 else:
2939 on_arch = ""
2940 excuse_info.append(
2941 f"Reproducibility issues ignored for src:{ignore_hint.package}"
2942 f"{on_arch} as requested by {ignore_hint.user}"
2943 )
2944 break
2946 if verdict.is_rejected:
2947 if source_suite_state == "known":
2948 check_bpids = failed_bpids - ignored_bpids
2949 else:
2950 # Let's not wait for results if all binaries have a hint
2951 check_bpids = filter_out_faux(source_data_srcdist.binaries)
2952 missed_bpids = set()
2954 for bpid in check_bpids:
2955 if (
2956 bpid_hint := self.hints.search_first(
2957 "ignore-reproducible",
2958 package=bpid.package_name,
2959 version=bpid.version,
2960 architecture=bpid.architecture,
2961 )
2962 ) is not None:
2963 # one hint per binary is enough
2964 all_hints.append(bpid_hint)
2965 self.logger.debug(
2966 "repro: hint found for %s: %s", source_name, bpid
2967 )
2968 else:
2969 missed_bpids.add(bpid)
2971 if not missed_bpids:
2972 verdict = PolicyVerdict.PASS_HINTED
2973 for hint in all_hints:
2974 policy_info.setdefault("hints", {}).setdefault(arch, []).append(
2975 hint.user + ": " + str(hint)
2976 )
2977 # TODO: we're going to print this for arch:all binaries on each arch
2978 excuse_info.append(
2979 f"Reproducibility issues ignored for {hint.package} on {arch} as "
2980 f"requested by {hint.user}"
2981 )
2982 elif all_hints: 2982 ↛ 2983line 2982 didn't jump to line 2983 because the condition on line 2982 was never true
2983 self.logger.info(
2984 "repro: binary hints for %s ignored as they don't cover these binaries %s",
2985 source_name,
2986 missed_bpids,
2987 )
2989 # A binary without results got hinted
2990 if not verdict.is_rejected and all_hints:
2991 for hint in all_hints:
2992 excuse_info.append(
2993 f"Reproducibility unknown for {hint.package} but ignored on {arch} as "
2994 f"requested by {hint.user}"
2995 )
2997 if self.options.repro_success_bounty and eligible_for_bounty: 2997 ↛ 2998line 2997 didn't jump to line 2998 because the condition on line 2997 was never true
2998 excuse.add_bounty("reproducibility", self.options.repro_success_bounty)
3000 if verdict.is_rejected and self.options.repro_regression_penalty: 3000 ↛ 3002line 3000 didn't jump to line 3002 because the condition on line 3000 was never true
3001 # With a non-zero penalty, we shouldn't block on this policy
3002 verdict = PolicyVerdict.PASS
3003 if self.options.repro_regression_penalty > 0:
3004 excuse.add_penalty(
3005 "reproducibility", self.options.repro_regression_penalty
3006 )
3008 for msg in excuse_info:
3009 if verdict.is_rejected:
3010 excuse.add_verdict_info(verdict, msg)
3011 else:
3012 excuse.addinfo(msg)
3014 return verdict