Coverage for britney2/policies/autopkgtest.py: 90%
850 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-08-18 12:43 +0000
1# Copyright (C) 2013 - 2016 Canonical Ltd.
2# Authors:
3# Colin Watson <cjwatson@ubuntu.com>
4# Jean-Baptiste Lallement <jean-baptiste.lallement@canonical.com>
5# Martin Pitt <martin.pitt@ubuntu.com>
7# This program is free software; you can redistribute it and/or modify
8# it under the terms of the GNU General Public License as published by
9# the Free Software Foundation; either version 2 of the License, or
10# (at your option) any later version.
12# This program is distributed in the hope that it will be useful,
13# but WITHOUT ANY WARRANTY; without even the implied warranty of
14# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15# GNU General Public License for more details.
17import calendar
18import collections
19import http.client
20import io
21import itertools
22import json
23import optparse
24import os
25import sys
26import tarfile
27import time
28import urllib.parse
29from collections.abc import Iterator
30from copy import deepcopy
31from enum import Enum
32from functools import lru_cache, total_ordering
33from typing import TYPE_CHECKING, Any, Optional, cast
34from urllib.error import HTTPError
35from urllib.request import urlopen
36from urllib.response import addinfourl
38import apt_pkg
39from more_itertools import iter_except
41from britney2 import (
42 BinaryPackageId,
43 PackageId,
44 SourcePackage,
45 SuiteClass,
46 Suites,
47 TargetSuite,
48)
49from britney2.hints import HintAnnotate, HintType
50from britney2.migrationitem import MigrationItem
51from britney2.policies import PolicyVerdict
52from britney2.policies.policy import AbstractBasePolicy
53from britney2.utils import (
54 binaries_from_source_version,
55 filter_out_faux,
56 filter_out_faux_gen,
57 get_dependency_solvers,
58 parse_option,
59)
61if TYPE_CHECKING: 61 ↛ 62line 61 didn't jump to line 62 because the condition on line 61 was never true
62 import amqplib.client_0_8 as amqp
64 from ..britney import Britney
65 from ..excuse import Excuse
66 from ..hints import HintParser
69@total_ordering
70class Result(Enum):
71 PASS = 1
72 NEUTRAL = 2
73 FAIL = 3
74 OLD_PASS = 4
75 OLD_NEUTRAL = 5
76 OLD_FAIL = 6
77 NONE = 7
79 def __lt__(self, other: "Result") -> bool:
80 return self.value < other.value
83EXCUSES_LABELS = {
84 "PASS": '<span style="background:#87d96c">Pass</span>',
85 "OLD_PASS": '<span style="background:#87d96c">Pass</span>',
86 "NEUTRAL": "No tests, superficial or marked flaky",
87 "OLD_NEUTRAL": "No tests, superficial or marked flaky",
88 "FAIL": '<span style="background:#ff6666">Failed</span>',
89 "OLD_FAIL": '<span style="background:#ff6666">Failed</span>',
90 "ALWAYSFAIL": '<span style="background:#e5c545">Failed (not a regression)</span>',
91 "REGRESSION": '<span style="background:#ff6666">Regression</span>',
92 "IGNORE-FAIL": '<span style="background:#e5c545">Ignored failure</span>',
93 "RUNNING": '<span style="background:#99ddff">Test triggered</span>',
94 "RUNNING-ALWAYSFAIL": "Test triggered (will not be considered a regression)",
95 "RUNNING-IGNORE": "Test triggered (failure will be ignored)",
96 "RUNNING-REFERENCE": '<span style="background:#ff6666">Reference test triggered, but real test failed already</span>',
97 "DEFERRED": '<span style="background:#99ddff">Test deferred</span>',
98}
100REF_TRIG = "migration-reference/0"
102VERSION_KEY = "britney-autopkgtest-pending-file-version"
105def srchash(src: str) -> str:
106 """archive hash prefix for source package"""
108 if src.startswith("lib"): 108 ↛ 109line 108 didn't jump to line 109 because the condition on line 108 was never true
109 return src[:4]
110 else:
111 return src[0]
114def added_pkgs_compared_to_target_suite(
115 package_ids: frozenset[BinaryPackageId],
116 target_suite: TargetSuite,
117 *,
118 invert: bool = False,
119) -> Iterator[BinaryPackageId]:
120 if invert: 120 ↛ 121line 120 didn't jump to line 121 because the condition on line 120 was never true
121 pkgs_ids_to_ignore = package_ids.difference(
122 target_suite.which_of_these_are_in_the_suite(package_ids)
123 )
124 names_ignored = {p.package_name for p in pkgs_ids_to_ignore}
125 else:
126 names_ignored = {
127 p.package_name
128 for p in target_suite.which_of_these_are_in_the_suite(package_ids)
129 }
130 yield from (p for p in package_ids if p.package_name not in names_ignored)
133def all_leaf_results(
134 test_results: dict[str, dict[str, dict[str, list[Any]]]],
135) -> Iterator[list[Any]]:
136 for trigger in test_results.values():
137 for arch in trigger.values():
138 yield from arch.values()
141def mark_result_as_old(result: Result) -> Result:
142 """Convert current result into corresponding old result"""
144 if result is Result.FAIL:
145 result = Result.OLD_FAIL
146 elif result is Result.PASS:
147 result = Result.OLD_PASS
148 elif result is Result.NEUTRAL: 148 ↛ 150line 148 didn't jump to line 150 because the condition on line 148 was always true
149 result = Result.OLD_NEUTRAL
150 return result
153def parse_bdeps(
154 src_data: SourcePackage, strip_multi_arch: bool = True, architecture: str = ""
155) -> Iterator[list[tuple[str, str, str]]]:
156 for bdeps in filter(None, (src_data.build_deps_arch, src_data.build_deps_indep)):
157 yield from apt_pkg.parse_src_depends(bdeps, strip_multi_arch, architecture)
160def has_autodep8(srcinfo: SourcePackage) -> bool:
161 """Check if package is covered by autodep8"""
162 return srcinfo.testsuite is not None and any(
163 t.startswith("autopkgtest-pkg") for t in srcinfo.testsuite
164 )
167def has_autodep8_or_autopkgtest(srcinfo: SourcePackage) -> bool:
168 """Check if package is covered by autodep8 or autopkgtest"""
169 return srcinfo.testsuite is not None and any(
170 t == "autopkgtest" or t.startswith("autopkgtest-pkg") for t in srcinfo.testsuite
171 )
174class AutopkgtestPolicy(AbstractBasePolicy):
175 """autopkgtest regression policy for source migrations
177 Run autopkgtests for the excuse and all of its reverse dependencies, and
178 reject the upload if any of those regress.
179 """
181 def __init__(self, options: optparse.Values, suite_info: Suites) -> None:
182 super().__init__(
183 "autopkgtest", options, suite_info, {SuiteClass.PRIMARY_SOURCE_SUITE}
184 )
185 # tests requested in this and previous runs
186 # trigger -> src -> [arch]
187 self.pending_tests: dict[str, dict[str, dict[str, int]]] | None = None
188 self.pending_tests_file = os.path.join(
189 self.state_dir, "autopkgtest-pending.json"
190 )
191 self.testsuite_triggers: dict[str, set[str]] = {}
192 self.result_in_baseline_cache: dict[str, dict[str, list[Any]]] = (
193 collections.defaultdict(dict)
194 )
196 self.amqp_file_handle: io.TextIOWrapper | None = None
198 # Default values for this policy's options
199 parse_option(options, "adt_baseline")
200 parse_option(options, "adt_huge", to_int=True)
201 parse_option(options, "adt_ppas")
202 parse_option(options, "adt_reference_max_age", day_to_sec=True)
203 parse_option(options, "adt_pending_max_age", default=5, day_to_sec=True)
204 parse_option(options, "adt_regression_penalty", default=0, to_int=True)
205 parse_option(options, "adt_log_url") # see below for defaults
206 parse_option(options, "adt_retry_url") # see below for defaults
207 parse_option(options, "adt_retry_older_than", day_to_sec=True)
208 parse_option(options, "adt_results_cache_age", day_to_sec=True)
209 parse_option(options, "adt_shared_results_cache")
210 parse_option(options, "adt_success_bounty", default=0, to_int=True)
211 parse_option(options, "adt_ignore_failure_for_new_tests", to_bool=True)
213 # When ADT_RESULTS_CACHE_AGE is smaller than or equal to
214 # ADT_REFERENCE_MAX_AGE old reference result will be removed from cache
215 # before the newly scheduled results are in, potentially causing
216 # additional waiting. For packages like glibc this might cause an
217 # infinite delay as there will always be a package that's
218 # waiting. Similarly for ADT_RETRY_OLDER_THAN.
219 if self.options.adt_results_cache_age <= self.options.adt_reference_max_age:
220 self.logger.warning(
221 "Unexpected: ADT_REFERENCE_MAX_AGE bigger than ADT_RESULTS_CACHE_AGE"
222 )
223 if self.options.adt_results_cache_age <= self.options.adt_retry_older_than:
224 self.logger.warning(
225 "Unexpected: ADT_RETRY_OLDER_THAN bigger than ADT_RESULTS_CACHE_AGE"
226 )
228 if not self.options.adt_log_url: 228 ↛ 254line 228 didn't jump to line 254 because the condition on line 228 was always true
229 # Historical defaults
230 if self.options.adt_swift_url.startswith("file://"):
231 self.options.adt_log_url = os.path.join(
232 self.options.adt_ci_url,
233 "data",
234 "autopkgtest",
235 self.options.series,
236 "{arch}",
237 "{hash}",
238 "{package}",
239 "{run_id}",
240 "log.gz",
241 )
242 else:
243 self.options.adt_log_url = os.path.join(
244 self.options.adt_swift_url,
245 "{swift_container}",
246 self.options.series,
247 "{arch}",
248 "{hash}",
249 "{package}",
250 "{run_id}",
251 "log.gz",
252 )
254 if hasattr(self.options, "adt_retry_url_mech"): 254 ↛ 255line 254 didn't jump to line 255 because the condition on line 254 was never true
255 self.logger.warning(
256 "The ADT_RETRY_URL_MECH configuration has been deprecated."
257 )
258 self.logger.warning(
259 "Instead britney now supports ADT_RETRY_URL for more flexibility."
260 )
261 if self.options.adt_retry_url:
262 self.logger.error(
263 "Please remove the ADT_RETRY_URL_MECH as ADT_RETRY_URL will be used."
264 )
265 elif self.options.adt_retry_url_mech == "run_id":
266 self.options.adt_retry_url = (
267 self.options.adt_ci_url + "api/v1/retry/{run_id}"
268 )
269 if not self.options.adt_retry_url: 269 ↛ 286line 269 didn't jump to line 286 because the condition on line 269 was always true
270 # Historical default
271 self.options.adt_retry_url = (
272 self.options.adt_ci_url
273 + "request.cgi?"
274 + "release={release}&arch={arch}&package={package}&trigger={trigger}{ppas}"
275 )
277 # results map: trigger -> src -> arch -> [passed, version, run_id, seen]
278 # - trigger is "source/version" of an unstable package that triggered
279 # this test run.
280 # - "passed" is a Result
281 # - "version" is the package version of "src" of that test
282 # - "run_id" is an opaque ID that identifies a particular test run for
283 # a given src/arch.
284 # - "seen" is an approximate time stamp of the test run. How this is
285 # deduced depends on the interface used.
286 self.test_results: dict[str, dict[str, dict[str, list[Any]]]] = {}
287 if self.options.adt_shared_results_cache:
288 self.results_cache_file = self.options.adt_shared_results_cache
289 else:
290 self.results_cache_file = os.path.join(
291 self.state_dir, "autopkgtest-results.cache"
292 )
294 try:
295 self.options.adt_ppas = self.options.adt_ppas.strip().split()
296 except AttributeError:
297 self.options.adt_ppas = []
299 self.swift_container = "autopkgtest-" + options.series
300 if self.options.adt_ppas:
301 self.swift_container += "-" + options.adt_ppas[-1].replace("/", "-")
303 # restrict adt_arches to architectures we actually run for
304 self.adt_arches = []
305 for arch in self.options.adt_arches.split():
306 if arch in self.options.architectures:
307 self.adt_arches.append(arch)
308 else:
309 self.logger.info(
310 "Ignoring ADT_ARCHES %s as it is not in architectures list", arch
311 )
313 def __del__(self) -> None:
314 if self.amqp_file_handle: 314 ↛ exitline 314 didn't return from function '__del__' because the condition on line 314 was always true
315 try:
316 self.amqp_file_handle.close()
317 except AttributeError:
318 pass
320 def register_hints(self, hint_parser: "HintParser") -> None:
321 hint_parser.register_hint_type(
322 HintType(
323 "force-badtest",
324 versioned=HintAnnotate.OPTIONAL,
325 architectured=HintAnnotate.OPTIONAL,
326 )
327 )
328 hint_parser.register_hint_type(
329 HintType(
330 "force-skiptest",
331 architectured=HintAnnotate.OPTIONAL,
332 )
333 )
335 def initialise(self, britney: "Britney") -> None:
336 super().initialise(britney)
337 # We want to use the "current" time stamp in multiple locations
338 time_now = round(time.time())
339 if hasattr(self.options, "fake_runtime"):
340 time_now = int(self.options.fake_runtime)
341 self._now = time_now
343 builddeps = sys.intern("@builddeps@")
345 # compute inverse Testsuite-Triggers: map, unifying all series
346 self.logger.info("Building inverse testsuite_triggers map")
347 for suite in self.suite_info:
348 for src, data in suite.sources.items():
349 # for now, let's assume that autodep8 uses builddeps (most do)
350 testsuite_triggers = data.testsuite_triggers or []
351 if has_autodep8(data) and builddeps not in testsuite_triggers:
352 testsuite_triggers.append(builddeps)
353 for trigger in testsuite_triggers:
354 if trigger == builddeps:
355 for arch in self.adt_arches:
356 for block in parse_bdeps(data, True, arch):
357 self.testsuite_triggers.setdefault(
358 block[0][0], set()
359 ).add(src)
360 else:
361 self.testsuite_triggers.setdefault(trigger, set()).add(src)
362 target_suite_name = self.suite_info.target_suite.name
364 os.makedirs(self.state_dir, exist_ok=True)
365 self.read_pending_tests()
367 # read the cached results that we collected so far
368 if os.path.exists(self.results_cache_file):
369 with open(self.results_cache_file) as f:
370 test_results = json.load(f)
371 self.test_results = self.check_and_upgrade_cache(test_results)
372 self.logger.info("Read previous results from %s", self.results_cache_file)
373 else:
374 self.logger.info(
375 "%s does not exist, re-downloading all results from swift",
376 self.results_cache_file,
377 )
379 # read in the new results
380 if self.options.adt_swift_url.startswith("file://"):
381 debci_file = self.options.adt_swift_url[7:]
382 if os.path.exists(debci_file):
383 with open(debci_file) as f:
384 test_results = json.load(f)
385 self.logger.info("Read new results from %s", debci_file)
386 for res in test_results["results"]:
387 # if there's no date, the test didn't finish yet
388 if res["date"] is None: 388 ↛ 389line 388 didn't jump to line 389 because the condition on line 388 was never true
389 continue
390 test_suite = res["suite"]
391 if test_suite != target_suite_name: 391 ↛ 393line 391 didn't jump to line 393 because the condition on line 391 was never true
392 # not requested for this target suite, so ignore
393 continue
394 triggers = res["trigger"]
395 if triggers is None: 395 ↛ 397line 395 didn't jump to line 397 because the condition on line 395 was never true
396 # not requested for this policy, so ignore
397 continue
398 status = res["status"]
399 if status is None:
400 # still running => pending
401 continue
402 src = res["package"]
403 arch = res["arch"]
404 ver = res["version"]
405 run_id = str(res["run_id"])
406 seen = round(
407 calendar.timegm(
408 time.strptime(res["date"][0:-5], "%Y-%m-%dT%H:%M:%S")
409 )
410 )
411 for trigger in triggers.split():
412 # remove matching test requests
413 self.remove_from_pending(trigger, src, arch, seen)
414 if status == "tmpfail": 414 ↛ 416line 414 didn't jump to line 416 because the condition on line 414 was never true
415 # let's see if we still need it
416 continue
417 self.logger.debug(
418 "Results %s %s %s added", src, trigger, status
419 )
420 self.add_trigger_to_results(
421 trigger,
422 src,
423 ver,
424 arch,
425 run_id,
426 seen,
427 Result[status.upper()],
428 )
429 else:
430 self.logger.info(
431 "%s does not exist, no new data will be processed", debci_file
432 )
434 # The cache can contain results against versions of packages that
435 # are not in any suite anymore. Strip those out, as we don't want
436 # to use those results. Additionally, old references may be
437 # filtered out.
438 if self.options.adt_baseline == "reference":
439 self.filter_old_results()
441 # we need sources, binaries, and installability tester, so for now
442 # remember the whole britney object
443 self.britney = britney
445 # Initialize AMQP connection
446 self.amqp_channel: Optional["amqp.channel.Channel"] = None
447 self.amqp_file_handle = None
448 if self.options.dry_run: 448 ↛ 449line 448 didn't jump to line 449 because the condition on line 448 was never true
449 return
451 amqp_url = self.options.adt_amqp
453 if amqp_url.startswith("amqp://"): 453 ↛ 454line 453 didn't jump to line 454 because the condition on line 453 was never true
454 import amqplib.client_0_8 as amqp
456 # depending on the setup we connect to a AMQP server
457 creds = urllib.parse.urlsplit(amqp_url, allow_fragments=False)
458 self.amqp_con = amqp.Connection(
459 creds.hostname, userid=creds.username, password=creds.password
460 )
461 self.amqp_channel = self.amqp_con.channel()
462 self.logger.info("Connected to AMQP server")
463 elif amqp_url.startswith("file://"): 463 ↛ 468line 463 didn't jump to line 468 because the condition on line 463 was always true
464 # or in Debian and in testing mode, adt_amqp will be a file:// URL
465 amqp_file = amqp_url[7:]
466 self.amqp_file_handle = open(amqp_file, "w", 1)
467 else:
468 raise RuntimeError("Unknown ADT_AMQP schema %s" % amqp_url.split(":", 1)[0])
470 def check_and_upgrade_cache(
471 self, test_results: dict[str, dict[str, dict[str, list[Any]]]]
472 ) -> dict[str, dict[str, dict[str, list[Any]]]]:
473 # Drop results older than ADT_RESULTS_CACHE_AGE
475 # Collect keys to delete instead of copying the full list of keys and
476 # changing the dicts on the fly. The lists containing the keys to delete
477 # only reaches the upper bound if all entries are too old.
478 to_delete_trigger = []
479 for trigger, trigger_data in test_results.items():
480 to_delete_pkg = []
481 for pkg, results in trigger_data.items():
482 to_delete_arch = []
483 for arch, arch_result in results.items():
484 arch_result[0] = Result[arch_result[0]]
485 if self._now - arch_result[3] > self.options.adt_results_cache_age: 485 ↛ 486line 485 didn't jump to line 486 because the condition on line 485 was never true
486 to_delete_arch.append(arch)
488 for arch in to_delete_arch: 488 ↛ 489line 488 didn't jump to line 489 because the loop on line 488 never started
489 del results[arch]
490 if not results: 490 ↛ 491line 490 didn't jump to line 491 because the condition on line 490 was never true
491 to_delete_pkg.append(pkg)
492 for pkg in to_delete_pkg: 492 ↛ 493line 492 didn't jump to line 493 because the loop on line 492 never started
493 del trigger_data[pkg]
494 if not trigger_data: 494 ↛ 495line 494 didn't jump to line 495 because the condition on line 494 was never true
495 to_delete_trigger.append(trigger)
496 for trigger in to_delete_trigger: 496 ↛ 497line 496 didn't jump to line 497 because the loop on line 496 never started
497 del test_results[trigger]
499 return test_results
501 def filter_old_results(self) -> None:
502 """Remove results for old versions and reference runs from the cache.
504 For now, only delete reference runs. If we delete regular
505 results after a while, packages with lots of triggered tests may
506 never have all the results at the same time."""
508 test_results = self.test_results
510 for trigger, trigger_data in test_results.items():
511 for src, results in trigger_data.items():
512 for result in results.values():
513 if (
514 trigger == REF_TRIG
515 and self._now - result[3] > self.options.adt_reference_max_age
516 ):
517 result[0] = mark_result_as_old(result[0])
518 elif not self.test_version_in_any_suite(src, result[1]):
519 result[0] = mark_result_as_old(result[0])
521 def test_version_in_any_suite(self, src: str, version: str) -> bool:
522 """Check if the mentioned version of src is found in a suite
524 To prevent regressions in the target suite, the result should be
525 from a test with the version of the package in either the source
526 suite or the target suite. The source suite is also valid,
527 because due to versioned test dependencies and Breaks/Conflicts
528 relations, regularly the version in the source suite is used
529 during testing.
530 """
532 versions = {
533 suite.sources[src].version
534 for suite in self.suite_info
535 if src in suite.sources
536 }
538 valid_version = False
539 for ver in versions:
540 if apt_pkg.version_compare(ver, version) == 0:
541 valid_version = True
542 break
544 return valid_version
546 def save_pending_json(self) -> None:
547 # update the pending tests on-disk cache
548 self.logger.info(
549 "Updating pending requested tests in %s", self.pending_tests_file
550 )
551 # Shallow clone pending_tests as we only modify the toplevel and change its type.
552 pending_tests: dict[str, Any] = {}
553 if self.pending_tests:
554 pending_tests = dict(self.pending_tests)
555 # Avoid adding if there are no pending results at all (eases testing)
556 pending_tests[VERSION_KEY] = 1
557 with open(self.pending_tests_file + ".new", "w") as f:
558 json.dump(pending_tests, f, indent=2)
559 os.rename(self.pending_tests_file + ".new", self.pending_tests_file)
561 def save_state(self, britney: "Britney") -> None:
562 super().save_state(britney)
564 # update the results on-disk cache, unless we are using a r/o shared one
565 if not self.options.adt_shared_results_cache:
566 self.logger.info("Updating results cache")
567 test_results = deepcopy(self.test_results)
568 for result in all_leaf_results(test_results):
569 result[0] = result[0].name
570 with open(self.results_cache_file + ".new", "w") as f:
571 json.dump(test_results, f, indent=2)
572 os.rename(self.results_cache_file + ".new", self.results_cache_file)
574 self.save_pending_json()
576 def format_retry_url(
577 self, run_id: str | None, arch: str, testsrc: str, trigger: str
578 ) -> str:
579 if self.options.adt_ppas:
580 ppas = "&" + urllib.parse.urlencode(
581 [("ppa", p) for p in self.options.adt_ppas]
582 )
583 else:
584 ppas = ""
585 return cast(str, self.options.adt_retry_url).format(
586 run_id=run_id,
587 release=self.options.series,
588 arch=arch,
589 package=testsrc,
590 trigger=urllib.parse.quote_plus(trigger),
591 ppas=ppas,
592 )
594 def format_log_url(self, testsrc: str, arch: str, run_id: str) -> str:
595 return cast(str, self.options.adt_log_url).format(
596 release=self.options.series,
597 swift_container=self.swift_container,
598 hash=srchash(testsrc),
599 package=testsrc,
600 arch=arch,
601 run_id=run_id,
602 )
604 def apply_src_policy_impl(
605 self,
606 tests_info: dict[str, Any],
607 source_data_tdist: SourcePackage | None,
608 source_data_srcdist: SourcePackage,
609 excuse: "Excuse",
610 ) -> PolicyVerdict:
612 # initialize
613 verdict = PolicyVerdict.PASS
614 source_name = excuse.item.package
616 # skip/delay autopkgtests until new package is built somewhere
617 if not binaries_from_source_version(source_data_srcdist, self.suite_info)[0]:
618 self.logger.debug(
619 "%s hasnot been built anywhere, skipping autopkgtest policy",
620 excuse.name,
621 )
622 verdict = PolicyVerdict.REJECTED_TEMPORARILY
623 excuse.add_verdict_info(verdict, "Autopkgtest deferred: missing builds")
625 elif "all" in excuse.missing_builds:
626 self.logger.debug(
627 "%s hasnot been built for arch:all, skipping autopkgtest policy",
628 source_name,
629 )
630 verdict = PolicyVerdict.REJECTED_TEMPORARILY
631 excuse.add_verdict_info(
632 verdict, "Autopkgtest deferred: missing arch:all build"
633 )
635 all_self_tests_pass = False
636 results_info: list[str] = []
637 if not verdict.is_rejected:
638 self.logger.debug("Checking autopkgtests for %s", source_name)
639 trigger = source_name + "/" + source_data_srcdist.version
641 # build a (testsrc, testver) → arch → (status, run_id, log_url) map; we trigger/check test
642 # results per architecture for technical/efficiency reasons, but we
643 # want to evaluate and present the results by tested source package
644 # first
645 pkg_arch_result: dict[
646 tuple[str, str], dict[str, tuple[str, str | None, str]]
647 ] = collections.defaultdict(dict)
648 for arch in self.adt_arches:
649 if arch in excuse.missing_builds:
650 verdict = PolicyVerdict.REJECTED_TEMPORARILY
651 self.logger.debug(
652 "%s hasnot been built on arch %s, delay autopkgtest there",
653 source_name,
654 arch,
655 )
656 excuse.add_verdict_info(
657 verdict,
658 f"Autopkgtest deferred on {arch}: missing arch:{arch} build",
659 )
660 else:
661 verdict = self.check_and_request_arch(
662 excuse,
663 arch,
664 source_data_srcdist,
665 pkg_arch_result,
666 trigger,
667 verdict,
668 )
670 verdict, results_info, all_self_tests_pass = self.process_pkg_arch_results(
671 tests_info, excuse, pkg_arch_result, verdict, trigger
672 )
674 verdict = self.finalize_excuse(
675 excuse, verdict, all_self_tests_pass, results_info
676 )
677 return verdict
679 def apply_srcarch_policy_impl(
680 self,
681 tests_info: dict[str, Any],
682 arch: str,
683 source_data_tdist: SourcePackage | None,
684 source_data_srcdist: SourcePackage,
685 excuse: "Excuse",
686 ) -> PolicyVerdict:
688 # TODO: disable this for now, since it causes issues as the autopkgtests
689 # are not scheduled with recursive dependencies as needed, which means
690 # transitions that use provides are not detected and the autopkgtests fail
691 # during installation.
692 return PolicyVerdict.PASS
694 assert self.hints is not None # for type checking
695 # initialize
696 verdict = PolicyVerdict.PASS
698 str_excuses = str(excuse.item)
699 self.logger.debug("Checking autopkgtests for binNMU %s/%s", str_excuses, arch)
701 if arch not in self.adt_arches:
702 return verdict
704 # find the binNMU version
705 versions = set()
706 for bin_pkg in source_data_srcdist.binaries:
707 if bin_pkg.architecture == arch:
708 if (
709 len(parts := bin_pkg.version.split("+b")) > 1
710 and parts[-1].isdigit()
711 ):
712 versions.add(parts[-1])
713 else:
714 self.logger.debug(
715 "Version %s doesn't end with '+b#', skipping", bin_pkg.version
716 )
717 if not versions or len(versions) > 1:
718 self.logger.debug("This migration item doesn't look like a binNMU")
719 return verdict
721 trigger = str_excuses + "/" + versions.pop()
723 # While we don't need the arch here, this is common with apply_src_policy_impl()
724 # (testsrc, testver) → arch → (status, run_id, log_url) map
725 pkg_arch_result: dict[
726 tuple[str, str], dict[str, tuple[str, str | None, str]]
727 ] = collections.defaultdict(dict)
729 verdict = self.check_and_request_arch(
730 excuse, arch, source_data_srcdist, pkg_arch_result, trigger, verdict
731 )
733 verdict, results_info, all_self_tests_pass = self.process_pkg_arch_results(
734 tests_info, excuse, pkg_arch_result, verdict, trigger
735 )
737 verdict = self.finalize_excuse(
738 excuse, verdict, all_self_tests_pass, results_info
739 )
740 return verdict
742 def check_and_request_arch(
743 self,
744 excuse: "Excuse",
745 arch: str,
746 source_data_srcdist: SourcePackage,
747 pkg_arch_result: dict[tuple[str, str], dict[str, tuple[str, str | None, str]]],
748 trigger: str,
749 verdict: PolicyVerdict,
750 ) -> PolicyVerdict:
751 """Perform sanity checks and request test/results when they pass"""
753 source_name = excuse.item.package
754 if arch in excuse.policy_info["depends"].get("arch_all_not_installable", []):
755 self.logger.debug(
756 "%s is uninstallable on arch %s (which is allowed), not running autopkgtest there",
757 source_name,
758 arch,
759 )
760 excuse.addinfo(
761 f"Autopkgtest skipped on {arch}: not installable (which is allowed)"
762 )
763 elif arch in excuse.unsatisfiable_on_archs and arch not in excuse.policy_info[
764 "depends"
765 ].get("autopkgtest_run_anyways", []):
766 verdict = PolicyVerdict.REJECTED_TEMPORARILY
767 self.logger.debug(
768 "%s is uninstallable on arch %s, not running autopkgtest there",
769 source_name,
770 arch,
771 )
772 excuse.addinfo(f"Autopkgtest skipped on {arch}: not installable")
773 else:
774 self.request_tests_for_source(
775 arch, source_data_srcdist, pkg_arch_result, excuse, trigger
776 )
778 return verdict
780 def process_pkg_arch_results(
781 self,
782 tests_info: dict[str, Any],
783 excuse: "Excuse",
784 pkg_arch_result: dict[tuple[str, str], dict[str, tuple[str, str | None, str]]],
785 verdict: PolicyVerdict,
786 trigger: str,
787 ) -> tuple[PolicyVerdict, list[str], bool]:
788 """Calculate verdict based on results and render excuse text"""
790 source_name = excuse.item.package
791 all_self_tests_pass = False
792 results_info = []
794 # add test result details to Excuse
795 cloud_url = self.options.adt_ci_url + "packages/%(h)s/%(s)s/%(r)s/%(a)s"
796 testver: str | None
797 for testsrc, testver in sorted(pkg_arch_result):
798 assert testver is not None
799 arch_results = pkg_arch_result[(testsrc, testver)]
800 r = {v[0] for v in arch_results.values()}
801 if r & {"FAIL", "OLD_FAIL", "REGRESSION"}:
802 verdict = PolicyVerdict.REJECTED_PERMANENTLY
803 elif (
804 r & {"DEFERRED", "RUNNING", "RUNNING-REFERENCE"}
805 and not verdict.is_rejected
806 ):
807 verdict = PolicyVerdict.REJECTED_TEMPORARILY
808 # skip version if still running on all arches
809 if not r - {"DEFERRED", "RUNNING", "RUNNING-ALWAYSFAIL", "RUNNING-IGNORE"}:
810 testver = None
812 # A source package is eligible for the bounty if it has tests
813 # of its own that pass on all tested architectures.
814 if testsrc == source_name:
815 excuse.autopkgtest_results = r
816 if r == {"PASS"}:
817 all_self_tests_pass = True
819 if testver:
820 testname = f"{testsrc}/{testver}"
821 else:
822 testname = testsrc
824 html_archmsg = []
825 for arch in sorted(arch_results):
826 status, run_id, log_url = arch_results[arch]
827 artifact_url = None
828 retry_url = None
829 reference_url = None
830 reference_retry_url = None
831 history_url = None
832 if self.options.adt_ppas:
833 if log_url.endswith("log.gz"):
834 artifact_url = log_url.replace("log.gz", "artifacts.tar.gz")
835 else:
836 history_url = cloud_url % {
837 "h": srchash(testsrc),
838 "s": testsrc,
839 "r": self.options.series,
840 "a": arch,
841 }
842 if status not in ("DEFERRED", "PASS", "RUNNING", "RUNNING-IGNORE"):
843 retry_url = self.format_retry_url(run_id, arch, testsrc, trigger)
845 baseline_result = self.result_in_baseline(testsrc, arch)
846 if baseline_result and baseline_result[0] is not Result.NONE:
847 baseline_run_id = str(baseline_result[2])
848 reference_url = self.format_log_url(
849 testsrc, arch, baseline_run_id
850 )
851 if self.options.adt_baseline == "reference":
852 reference_retry_url = self.format_retry_url(
853 baseline_run_id, arch, testsrc, REF_TRIG
854 )
855 tests_info.setdefault(testname, {})[arch] = [
856 status,
857 log_url,
858 history_url,
859 artifact_url,
860 retry_url,
861 ]
863 # render HTML snippet for testsrc entry for current arch
864 if history_url:
865 message = f'<a href="{history_url}">{arch}</a>'
866 else:
867 message = arch
868 message += ': <a href="{}">{}</a>'.format(
869 log_url,
870 EXCUSES_LABELS[status],
871 )
872 if retry_url:
873 message += (
874 '<a href="%s" style="text-decoration: none;"> ♻</a>' % retry_url
875 )
876 if reference_url:
877 message += ' (<a href="%s">reference</a>' % reference_url
878 if reference_retry_url:
879 message += (
880 '<a href="%s" style="text-decoration: none;"> ♻</a>'
881 % reference_retry_url
882 )
883 message += ")"
884 if artifact_url:
885 message += ' <a href="%s">[artifacts]</a>' % artifact_url
886 html_archmsg.append(message)
888 # render HTML line for testsrc entry
889 # - if action is or may be required
890 # - for ones own package
891 if (
892 r
893 - {
894 "PASS",
895 "NEUTRAL",
896 "RUNNING-ALWAYSFAIL",
897 "ALWAYSFAIL",
898 "IGNORE-FAIL",
899 }
900 or testsrc == source_name
901 ):
902 if testver:
903 pkg = '<a href="#{0}">{0}</a>/{1}'.format(testsrc, testver)
904 else:
905 pkg = '<a href="#{0}">{0}</a>'.format(testsrc)
906 results_info.append(
907 "Autopkgtest for {}: {}".format(pkg, ", ".join(html_archmsg))
908 )
910 return (verdict, results_info, all_self_tests_pass)
912 def finalize_excuse(
913 self,
914 excuse: "Excuse",
915 verdict: PolicyVerdict,
916 all_self_tests_pass: bool,
917 results_info: list[str],
918 ) -> PolicyVerdict:
919 """Updates excuses and verdict for hints and bounty/penalty config
921 Given the verdict so far, hints and configuration, the verdict may be
922 updated. Depending of the end verdict, the content of results_info is
923 added as info or as excuse.
924 """
926 package = excuse.item.package
927 version = excuse.item.version
929 assert self.hints is not None # for type checking
930 if verdict.is_rejected:
931 # check for force-skiptest hint
932 if (
933 hint := self.hints.search_first(
934 "force-skiptest",
935 package=package,
936 version=version,
937 )
938 ) is not None:
939 excuse.addreason("skiptest")
940 excuse.addinfo(
941 "Not waiting for autopkgtest results and failures are "
942 f"ignored because of hint by {hint.user}"
943 )
944 verdict = PolicyVerdict.PASS_HINTED
945 else:
946 excuse.addreason("autopkgtest")
948 if (
949 self.options.adt_success_bounty
950 and verdict is PolicyVerdict.PASS
951 and all_self_tests_pass
952 ):
953 excuse.add_bounty("autopkgtest", self.options.adt_success_bounty)
954 if self.options.adt_regression_penalty and verdict in {
955 PolicyVerdict.REJECTED_PERMANENTLY,
956 PolicyVerdict.REJECTED_TEMPORARILY,
957 }:
958 if self.options.adt_regression_penalty > 0: 958 ↛ 961line 958 didn't jump to line 961 because the condition on line 958 was always true
959 excuse.add_penalty("autopkgtest", self.options.adt_regression_penalty)
960 # In case we give penalties instead of blocking, we must always pass
961 verdict = PolicyVerdict.PASS
962 for i in results_info:
963 if verdict.is_rejected:
964 excuse.add_verdict_info(verdict, i)
965 else:
966 excuse.addinfo(i)
968 return verdict
970 def request_tests_for_source(
971 self,
972 arch: str,
973 source_data_srcdist: SourcePackage,
974 pkg_arch_result: dict[tuple[str, str], dict[str, tuple[str, str | None, str]]],
975 excuse: "Excuse",
976 trigger: str,
977 ) -> None:
978 pkg_universe = self.britney.pkg_universe
979 target_suite = self.suite_info.target_suite
980 source_suite = excuse.item.suite
981 sources_t = target_suite.sources
982 sources_s = excuse.item.suite.sources
983 packages_s_a = excuse.item.suite.binaries[arch]
984 source_name = excuse.item.package
985 source_version = source_data_srcdist.version
986 # request tests (unless they were already requested earlier or have a result)
987 tests = self.tests_for_source(source_name, source_version, arch, excuse)
988 is_huge = len(tests) > self.options.adt_huge
990 # local copies for better performance
991 parse_src_depends = apt_pkg.parse_src_depends
993 # Here we figure out what is required from the source suite
994 # for the test to install successfully.
995 #
996 # The ImplicitDependencyPolicy does a similar calculation, but
997 # if I (elbrus) understand correctly, only in the reverse
998 # dependency direction. We are doing something similar here
999 # but in the dependency direction (note: this code is older).
1000 # We use the ImplicitDependencyPolicy result for the reverse
1001 # dependencies and we keep the code below for the
1002 # dependencies. Using the ImplicitDependencyPolicy results
1003 # also in the reverse direction seems to require quite some
1004 # reorganisation to get that information available here, as in
1005 # the current state only the current excuse is available here
1006 # and the required other excuses may not be calculated yet.
1007 #
1008 # Loop over all binary packages from trigger and
1009 # recursively look up which *versioned* dependencies are
1010 # only satisfied in the source suite.
1011 #
1012 # For all binaries found, look up which packages they
1013 # break/conflict with in the target suite, but not in the
1014 # source suite. The main reason to do this is to cover test
1015 # dependencies, so we will check Testsuite-Triggers as
1016 # well.
1017 #
1018 # OI: do we need to do the first check in a smart way
1019 # (i.e. only for the packages that are actually going to be
1020 # installed) for the breaks/conflicts set as well, i.e. do
1021 # we need to check if any of the packages that we now
1022 # enforce being from the source suite, actually have new
1023 # versioned depends and new breaks/conflicts.
1024 #
1025 # For all binaries found, add the set of unique source
1026 # packages to the list of triggers.
1028 bin_triggers: set[PackageId] = set()
1029 bin_new = filter_out_faux(source_data_srcdist.binaries)
1030 # For each build-depends block (if any) check if the first alternative
1031 # is satisfiable in the target suite. If not, add it to the initial set
1032 # used for checking.
1033 for block in parse_bdeps(source_data_srcdist, True, arch):
1034 if (
1035 next(
1036 get_dependency_solvers(
1037 (block[0],),
1038 target_suite.binaries[arch],
1039 target_suite.provides_table[arch],
1040 build_depends=True,
1041 ),
1042 None,
1043 )
1044 is None
1045 and (
1046 solvers := next(
1047 get_dependency_solvers(
1048 (block[0],),
1049 packages_s_a,
1050 source_suite.provides_table[arch],
1051 build_depends=True,
1052 ),
1053 None,
1054 )
1055 )
1056 is not None
1057 ):
1058 bin_new.add(solvers.pkg_id)
1059 for n_binary in iter_except(bin_new.pop, KeyError):
1060 if n_binary in bin_triggers:
1061 continue
1062 bin_triggers.add(n_binary)
1064 # Check if there is a dependency that is not
1065 # available in the target suite.
1066 # We add slightly too much here, because new binaries
1067 # will also show up, but they are already properly
1068 # installed. Nevermind.
1069 depends = pkg_universe.dependencies_of(n_binary)
1070 # depends is a frozenset{frozenset{BinaryPackageId, ..}}
1071 for deps_of_bin in depends:
1072 if target_suite.any_of_these_are_in_the_suite(deps_of_bin):
1073 # if any of the alternative dependencies is already
1074 # satisfied in the target suite, we can just ignore it
1075 continue
1076 # We'll figure out which version later
1077 bin_new.update(
1078 added_pkgs_compared_to_target_suite(deps_of_bin, target_suite)
1079 )
1081 # Check if the package breaks/conflicts anything. We might
1082 # be adding slightly too many source packages due to the
1083 # check here as a binary package that is broken may be
1084 # coming from a different source package in the source
1085 # suite. Nevermind.
1086 bin_broken = set()
1087 for t_binary in bin_triggers:
1088 # broken is a frozenset{BinaryPackageId, ..}
1089 broken = pkg_universe.negative_dependencies_of(
1090 cast(BinaryPackageId, t_binary)
1091 )
1092 broken_in_target = {
1093 p.package_name
1094 for p in target_suite.which_of_these_are_in_the_suite(broken)
1095 }
1096 broken_in_source = {
1097 p.package_name
1098 for p in source_suite.which_of_these_are_in_the_suite(broken)
1099 }
1100 # We want packages with a newer version in the source suite that
1101 # no longer has the conflict. This is an approximation
1102 broken_filtered = {
1103 p
1104 for p in broken
1105 if p.package_name in broken_in_target
1106 and p.package_name not in broken_in_source
1107 }
1108 # We add the version in the target suite, but the code below will
1109 # change it to the version in the source suite
1110 bin_broken.update(broken_filtered)
1111 bin_triggers.update(bin_broken)
1113 # The ImplicitDependencyPolicy also found packages that need
1114 # to migrate together, so add them to the triggers too.
1115 for bin_implicit in excuse.depends_packages_flattened:
1116 if bin_implicit.architecture == arch:
1117 bin_triggers.add(bin_implicit)
1119 triggers = set()
1120 for t_binary2 in bin_triggers:
1121 if t_binary2.architecture == arch:
1122 try:
1123 source_of_bin = packages_s_a[t_binary2.package_name].source
1124 # If the version in the target suite is the same, don't add a trigger.
1125 # Note that we looked up the source package in the source suite.
1126 # If it were a different source package in the target suite, however, then
1127 # we would not have this source package in the same version anyway.
1128 #
1129 # binNMU's exist, so let's also check if t_binary2 exists
1130 # in the target suite if the sources are the same.
1131 if ( 1131 ↛ 1147line 1131 didn't jump to line 1147
1132 sources_t.get(source_of_bin, None) is None
1133 or sources_s[source_of_bin].version
1134 != sources_t[source_of_bin].version
1135 or not target_suite.any_of_these_are_in_the_suite(
1136 {cast(BinaryPackageId, t_binary2)}
1137 )
1138 ):
1139 triggers.add(
1140 source_of_bin + "/" + sources_s[source_of_bin].version
1141 )
1142 except KeyError:
1143 # Apparently the package was removed from
1144 # unstable e.g. if packages are replaced
1145 # (e.g. -dbg to -dbgsym)
1146 pass
1147 if t_binary2 not in source_data_srcdist.binaries:
1148 for tdep_src in self.testsuite_triggers.get(
1149 t_binary2.package_name, set()
1150 ):
1151 try:
1152 # Only add trigger if versions in the target and source suites are different
1153 if ( 1153 ↛ 1148line 1153 didn't jump to line 1148
1154 sources_t.get(tdep_src, None) is None
1155 or sources_s[tdep_src].version
1156 != sources_t[tdep_src].version
1157 ):
1158 triggers.add(
1159 tdep_src + "/" + sources_s[tdep_src].version
1160 )
1161 except KeyError:
1162 # Apparently the source was removed from
1163 # unstable (testsuite_triggers are unified
1164 # over all suites)
1165 pass
1166 source_trigger = source_name + "/" + source_version
1167 triggers.discard(source_trigger)
1168 triggers_list = sorted(list(triggers))
1169 triggers_list.insert(0, trigger)
1171 impl_pids = excuse.policy_info.get("implicit-deps", {}).get(
1172 "broken-binaries", []
1173 )
1174 for testsrc, testver in tests:
1175 # Not if binaries from testsrc are not installable
1176 skip = False
1177 for bpid_s in impl_pids:
1178 bpid = BinaryPackageId(*bpid_s.split("/"))
1179 if (
1180 bpid.architecture == arch
1181 and testsrc == target_suite.all_binaries_in_suite[bpid].source
1182 ):
1183 skip = True
1184 break
1185 if skip:
1186 pkg_arch_result[(testsrc, testver)][arch] = ("DEFERRED", None, "")
1187 else:
1188 self.pkg_test_request(testsrc, arch, triggers_list, huge=is_huge)
1189 result, real_ver, run_id, url = self.pkg_test_result(
1190 testsrc, testver, arch, trigger
1191 )
1192 pkg_arch_result[(testsrc, real_ver)][arch] = (result, run_id, url)
1194 def tests_for_source(
1195 self, src: str, ver: str, arch: str, excuse: "Excuse"
1196 ) -> list[tuple[str, str]]:
1197 """Iterate over all tests that should be run for given source and arch"""
1199 source_suite = self.suite_info.primary_source_suite
1200 target_suite = self.suite_info.target_suite
1201 sources_info = target_suite.sources
1202 binaries_info = target_suite.binaries[arch]
1204 reported_pkgs = set()
1206 tests = []
1208 # Debian doesn't have linux-meta, but Ubuntu does
1209 # for linux themselves we don't want to trigger tests -- these should
1210 # all come from linux-meta*. A new kernel ABI without a corresponding
1211 # -meta won't be installed and thus we can't sensibly run tests against
1212 # it.
1213 if ( 1213 ↛ 1217line 1213 didn't jump to line 1217
1214 src.startswith("linux")
1215 and src.replace("linux", "linux-meta") in sources_info
1216 ):
1217 return []
1219 # we want to test the package itself, if it still has a test in unstable
1220 # but only if the package actually exists on this arch
1221 srcinfo = source_suite.sources[src]
1222 if has_autodep8_or_autopkgtest(srcinfo) and len(excuse.packages[arch]) > 0:
1223 reported_pkgs.add(src)
1224 tests.append((src, ver))
1226 extra_bins = []
1227 # Debian doesn't have linux-meta, but Ubuntu does
1228 # Hack: For new kernels trigger all DKMS packages by pretending that
1229 # linux-meta* builds a "dkms" binary as well. With that we ensure that we
1230 # don't regress DKMS drivers with new kernel versions.
1231 if src.startswith("linux-meta"):
1232 # does this have any image on this arch?
1233 for pkg_id in srcinfo.binaries:
1234 if pkg_id.architecture == arch and "-image" in pkg_id.package_name:
1235 try:
1236 extra_bins.append(binaries_info["dkms"].pkg_id)
1237 except KeyError:
1238 pass
1240 if not self.has_built_on_this_arch_or_is_arch_all(srcinfo, arch):
1241 return []
1243 pkg_universe = self.britney.pkg_universe
1244 # plus all direct reverse dependencies and test triggers of its
1245 # binaries which have an autopkgtest
1246 for binary in itertools.chain(srcinfo.binaries, extra_bins):
1247 for rdep in filter_out_faux_gen(
1248 pkg_universe.reverse_dependencies_of(binary)
1249 ):
1250 try:
1251 rdep_src = binaries_info[rdep.package_name].source
1252 # Don't re-trigger the package itself here; this should
1253 # have been done above if the package still continues to
1254 # have an autopkgtest in unstable.
1255 if rdep_src == src:
1256 continue
1257 except KeyError:
1258 continue
1260 rdep_src_info = sources_info[rdep_src]
1261 if has_autodep8_or_autopkgtest(rdep_src_info):
1262 if rdep_src not in reported_pkgs:
1263 tests.append((rdep_src, rdep_src_info.version))
1264 reported_pkgs.add(rdep_src)
1266 for tdep_src in self.testsuite_triggers.get(binary.package_name, set()):
1267 if tdep_src not in reported_pkgs:
1268 try:
1269 tdep_src_info = sources_info[tdep_src]
1270 except KeyError:
1271 continue
1272 if has_autodep8_or_autopkgtest(tdep_src_info): 1272 ↛ 1266line 1272 didn't jump to line 1266 because the condition on line 1272 was always true
1273 for pkg_id in tdep_src_info.binaries: 1273 ↛ 1266line 1273 didn't jump to line 1266 because the loop on line 1273 didn't complete
1274 if pkg_id.architecture == arch:
1275 tests.append((tdep_src, tdep_src_info.version))
1276 reported_pkgs.add(tdep_src)
1277 break
1279 tests.sort(key=lambda s_v: s_v[0])
1280 return tests
1282 def read_pending_tests(self) -> None:
1283 """Read pending test requests from previous britney runs
1285 Initialize self.pending_tests with that data.
1286 """
1287 assert self.pending_tests is None, "already initialized"
1288 if not os.path.exists(self.pending_tests_file):
1289 self.logger.info(
1290 "No %s, starting with no pending tests", self.pending_tests_file
1291 )
1292 self.pending_tests = {}
1293 return
1294 with open(self.pending_tests_file) as f:
1295 self.pending_tests = json.load(f)
1296 if VERSION_KEY in self.pending_tests:
1297 del self.pending_tests[VERSION_KEY]
1299 # Collect keys to delete instead of copying the full list of keys and
1300 # changing the dicts on the fly. The lists containing the keys to delete
1301 # only reaches the upper bound if all entries are too old.
1302 to_delete_trigger = []
1303 for trigger, trigger_results in self.pending_tests.items():
1304 to_delete_pkg = []
1305 for pkg, arch_dict in trigger_results.items():
1306 to_delete_arch = []
1307 for arch, pending_test in arch_dict.items():
1308 if self._now - pending_test > self.options.adt_pending_max_age:
1309 to_delete_arch.append(arch)
1311 for key in to_delete_arch:
1312 del arch_dict[key]
1313 if not arch_dict:
1314 to_delete_pkg.append(pkg)
1315 for key in to_delete_pkg:
1316 del trigger_results[key]
1317 if not trigger_results:
1318 to_delete_trigger.append(trigger)
1319 for key in to_delete_trigger:
1320 del self.pending_tests[key]
1321 else:
1322 # Migration code:
1323 for trigger_data in self.pending_tests.values(): 1323 ↛ 1324line 1323 didn't jump to line 1324 because the loop on line 1323 never started
1324 for pkg, arch_list in trigger_data.items():
1325 trigger_data[pkg] = {}
1326 for arch in arch_list:
1327 trigger_data[pkg][arch] = self._now
1329 self.logger.info(
1330 "Read pending requested tests from %s", self.pending_tests_file
1331 )
1332 self.logger.debug("%s", self.pending_tests)
1334 # this requires iterating over all triggers and thus is expensive;
1335 # cache the results
1336 @lru_cache(None)
1337 def latest_run_for_package(self, src: str, arch: str) -> str:
1338 """Return latest run ID for src on arch"""
1340 latest_run_id = ""
1341 for srcmap in self.test_results.values():
1342 try:
1343 run_id = srcmap[src][arch][2]
1344 except KeyError:
1345 continue
1346 if run_id > latest_run_id:
1347 latest_run_id = run_id
1348 return latest_run_id
1350 def urlopen_retry(self, url: str) -> http.client.HTTPResponse | addinfourl:
1351 """A urlopen() that retries on time outs or errors"""
1353 exc: Exception
1354 for retry in range(5): 1354 ↛ 1378line 1354 didn't jump to line 1378 because the loop on line 1354 didn't complete
1355 try:
1356 req = urlopen(url, timeout=30)
1357 code = req.getcode()
1358 if not code or 200 <= code < 300: 1358 ↛ 1354line 1358 didn't jump to line 1354 because the condition on line 1358 was always true
1359 return req # type: ignore[no-any-return]
1360 except TimeoutError as e: 1360 ↛ 1361line 1360 didn't jump to line 1361 because the exception caught by line 1360 didn't happen
1361 self.logger.info(
1362 "Timeout downloading '%s', will retry %d more times.",
1363 url,
1364 5 - retry - 1,
1365 )
1366 exc = e
1367 except HTTPError as e:
1368 if e.code not in (503, 502): 1368 ↛ 1370line 1368 didn't jump to line 1370 because the condition on line 1368 was always true
1369 raise
1370 self.logger.info(
1371 "Caught error %d downloading '%s', will retry %d more times.",
1372 e.code,
1373 url,
1374 5 - retry - 1,
1375 )
1376 exc = e
1377 else:
1378 raise exc
1380 @lru_cache(None)
1381 def fetch_swift_results(self, swift_url: str, src: str, arch: str) -> None:
1382 """Download new results for source package/arch from swift"""
1384 # prepare query: get all runs with a timestamp later than the latest
1385 # run_id for this package/arch; '@' is at the end of each run id, to
1386 # mark the end of a test run directory path
1387 # example: <autopkgtest-wily>wily/amd64/libp/libpng/20150630_054517@/result.tar
1388 query = {
1389 "delimiter": "@",
1390 "prefix": f"{self.options.series}/{arch}/{srchash(src)}/{src}/",
1391 }
1393 # determine latest run_id from results
1394 if not self.options.adt_shared_results_cache:
1395 latest_run_id = self.latest_run_for_package(src, arch)
1396 if latest_run_id:
1397 query["marker"] = query["prefix"] + latest_run_id
1399 # request new results from swift
1400 url = os.path.join(swift_url, self.swift_container)
1401 url += "?" + urllib.parse.urlencode(query)
1402 f = None
1403 try:
1404 f = self.urlopen_retry(url)
1405 if f.getcode() == 200:
1406 result_paths = f.read().decode().strip().splitlines()
1407 elif f.getcode() == 204: # No content 1407 ↛ 1413line 1407 didn't jump to line 1413 because the condition on line 1407 was always true
1408 result_paths = []
1409 else:
1410 # we should not ever end up here as we expect a HTTPError in
1411 # other cases; e. g. 3XX is something that tells us to adjust
1412 # our URLS, so fail hard on those
1413 raise NotImplementedError(
1414 f"fetch_swift_results({url}): cannot handle HTTP code {f.getcode()!r}"
1415 )
1416 except OSError as e:
1417 # 401 "Unauthorized" is swift's way of saying "container does not exist"
1418 if getattr(e, "code", -1) == 401: 1418 ↛ 1427line 1418 didn't jump to line 1427 because the condition on line 1418 was always true
1419 self.logger.info(
1420 "fetch_swift_results: %s does not exist yet or is inaccessible", url
1421 )
1422 return
1423 # Other status codes are usually a transient
1424 # network/infrastructure failure. Ignoring this can lead to
1425 # re-requesting tests which we already have results for, so
1426 # fail hard on this and let the next run retry.
1427 self.logger.error("Failure to fetch swift results from %s: %s", url, e)
1428 sys.exit(1)
1429 finally:
1430 if f is not None: 1430 ↛ 1433line 1430 didn't jump to line 1433 because the condition on line 1430 was always true
1431 f.close() 1431 ↛ exitline 1431 didn't return from function 'fetch_swift_results' because the return on line 1422 wasn't executed
1433 for p in result_paths:
1434 self.fetch_one_result(
1435 os.path.join(swift_url, self.swift_container, p, "result.tar"),
1436 src,
1437 arch,
1438 )
1440 def fetch_one_result(self, url: str, src: str, arch: str) -> None:
1441 """Download one result URL for source/arch
1443 Remove matching pending_tests entries.
1444 """
1445 f = None
1446 try:
1447 f = self.urlopen_retry(url)
1448 if f.getcode() == 200: 1448 ↛ 1451line 1448 didn't jump to line 1451 because the condition on line 1448 was always true
1449 tar_bytes = io.BytesIO(f.read())
1450 else:
1451 raise NotImplementedError(
1452 f"fetch_one_result({url}): cannot handle HTTP code {f.getcode()!r}"
1453 )
1454 except OSError as err:
1455 self.logger.error("Failure to fetch %s: %s", url, err)
1456 # we tolerate "not found" (something went wrong on uploading the
1457 # result), but other things indicate infrastructure problems
1458 if getattr(err, "code", -1) == 404:
1459 return
1460 sys.exit(1)
1461 finally:
1462 if f is not None: 1462 ↛ exit, 1462 ↛ 14642 missed branches: 1) line 1462 didn't return from function 'fetch_one_result' because the return on line 1459 wasn't executed, 2) line 1462 didn't jump to line 1464 because the condition on line 1462 was always true
1463 f.close() 1463 ↛ exitline 1463 didn't return from function 'fetch_one_result' because the return on line 1459 wasn't executed
1464 try:
1465 with tarfile.open(None, "r", tar_bytes) as tar:
1466 exitcode = int(tar.extractfile("exitcode").read().strip()) # type: ignore[union-attr]
1467 srcver = tar.extractfile("testpkg-version").read().decode().strip() # type: ignore[union-attr]
1468 ressrc, ver = srcver.split()
1469 testinfo = json.loads(tar.extractfile("testinfo.json").read().decode()) # type: ignore[union-attr]
1470 except (KeyError, ValueError, tarfile.TarError) as err:
1471 self.logger.error("%s is damaged, ignoring: %s", url, err)
1472 # ignore this; this will leave an orphaned request in autopkgtest-pending.json
1473 # and thus require manual retries after fixing the tmpfail, but we
1474 # can't just blindly attribute it to some pending test.
1475 return
1477 if src != ressrc: 1477 ↛ 1478line 1477 didn't jump to line 1478 because the condition on line 1477 was never true
1478 self.logger.error(
1479 "%s is a result for package %s, but expected package %s",
1480 url,
1481 ressrc,
1482 src,
1483 )
1484 return
1486 # parse recorded triggers in test result
1487 for e in testinfo.get("custom_environment", []): 1487 ↛ 1492line 1487 didn't jump to line 1492 because the loop on line 1487 didn't complete
1488 if e.startswith("ADT_TEST_TRIGGERS="): 1488 ↛ 1487line 1488 didn't jump to line 1487 because the condition on line 1488 was always true
1489 result_triggers = [i for i in e.split("=", 1)[1].split() if "/" in i]
1490 break
1491 else:
1492 self.logger.error("%s result has no ADT_TEST_TRIGGERS, ignoring")
1493 return
1495 run_id = os.path.basename(os.path.dirname(url))
1496 seen = round(calendar.timegm(time.strptime(run_id, "%Y%m%d_%H%M%S@")))
1497 # allow some skipped tests, but nothing else
1498 if exitcode in [0, 2]:
1499 result = Result.PASS
1500 elif exitcode == 8: 1500 ↛ 1501line 1500 didn't jump to line 1501 because the condition on line 1500 was never true
1501 result = Result.NEUTRAL
1502 else:
1503 result = Result.FAIL
1505 self.logger.info(
1506 "Fetched test result for %s/%s/%s %s (triggers: %s): %s",
1507 src,
1508 ver,
1509 arch,
1510 run_id,
1511 result_triggers,
1512 result.name.lower(),
1513 )
1515 # remove matching test requests
1516 for trigger in result_triggers:
1517 self.remove_from_pending(trigger, src, arch)
1519 # add this result
1520 for trigger in result_triggers:
1521 self.add_trigger_to_results(trigger, src, ver, arch, run_id, seen, result)
1523 def remove_from_pending(
1524 self, trigger: str, src: str, arch: str, timestamp: int = sys.maxsize
1525 ) -> None:
1526 assert self.pending_tests is not None # for type checking
1527 try:
1528 arch_dict = self.pending_tests[trigger][src]
1529 if timestamp < arch_dict[arch]:
1530 # The result is from before the moment of scheduling, so it's
1531 # not the one we're waiting for
1532 return
1533 del arch_dict[arch]
1534 if not arch_dict:
1535 del self.pending_tests[trigger][src]
1536 if not self.pending_tests[trigger]:
1537 del self.pending_tests[trigger]
1538 self.logger.debug(
1539 "-> matches pending request %s/%s for trigger %s", src, arch, trigger
1540 )
1541 except KeyError:
1542 self.logger.debug(
1543 "-> does not match any pending request for %s/%s", src, arch
1544 )
1546 def add_trigger_to_results(
1547 self,
1548 trigger: str,
1549 src: str,
1550 ver: str,
1551 arch: str,
1552 run_id: str,
1553 timestamp: int,
1554 status_to_add: Result,
1555 ) -> None:
1556 # Ensure that we got a new enough version
1557 parts = trigger.split("/")
1558 match len(parts):
1559 case 2: 1559 ↛ 1561line 1559 didn't jump to line 1561 because the pattern on line 1559 always matched
1560 trigsrc, trigver = parts
1561 case 4:
1562 trigsrc, trigarch, trigver, rebuild = parts
1563 case _:
1564 self.logger.info("Ignoring invalid test trigger %s", trigger)
1565 return
1566 if trigsrc == src and apt_pkg.version_compare(ver, trigver) < 0: 1566 ↛ 1567line 1566 didn't jump to line 1567 because the condition on line 1566 was never true
1567 self.logger.debug(
1568 "test trigger %s, but run for older version %s, ignoring", trigger, ver
1569 )
1570 return
1572 stored_result = (
1573 self.test_results.setdefault(trigger, {})
1574 .setdefault(src, {})
1575 .setdefault(arch, [Result.FAIL, None, "", 0])
1576 )
1578 # reruns shouldn't flip the result from PASS or NEUTRAL to
1579 # FAIL, so remember the most recent version of the best result
1580 # we've seen. Except for reference updates, which we always
1581 # want to update with the most recent result. The result data
1582 # may not be ordered by timestamp, so we need to check time.
1583 update = False
1584 if self.options.adt_baseline == "reference" and trigger == REF_TRIG:
1585 if stored_result[3] < timestamp:
1586 update = True
1587 elif status_to_add < stored_result[0]:
1588 update = True
1589 elif status_to_add == stored_result[0] and stored_result[3] < timestamp:
1590 update = True
1592 if update:
1593 stored_result[0] = status_to_add
1594 stored_result[1] = ver
1595 stored_result[2] = run_id
1596 stored_result[3] = timestamp
1598 def send_test_request(
1599 self, src: str, arch: str, triggers: list[str], huge: bool = False
1600 ) -> None:
1601 """Send out AMQP request for testing src/arch for triggers
1603 If huge is true, then the request will be put into the -huge instead of
1604 normal queue.
1605 """
1606 if self.options.dry_run: 1606 ↛ 1607line 1606 didn't jump to line 1607 because the condition on line 1606 was never true
1607 return
1609 params: dict[str, Any] = {"triggers": triggers}
1610 if self.options.adt_ppas:
1611 params["ppas"] = self.options.adt_ppas
1612 qname = f"debci-ppa-{self.options.series}-{arch}"
1613 elif huge:
1614 qname = f"debci-huge-{self.options.series}-{arch}"
1615 else:
1616 qname = f"debci-{self.options.series}-{arch}"
1617 params["submit-time"] = time.strftime("%Y-%m-%d %H:%M:%S%z", time.gmtime())
1619 if self.amqp_channel: 1619 ↛ 1620line 1619 didn't jump to line 1620 because the condition on line 1619 was never true
1620 self.amqp_channel.basic_publish(
1621 amqp.Message(
1622 src + "\n" + json.dumps(params), delivery_mode=2
1623 ), # persistent
1624 routing_key=qname,
1625 )
1626 # we save pending.json with every request, so that if britney
1627 # crashes we don't re-request tests. This is only needed when using
1628 # real amqp, as with file-based submission the pending tests are
1629 # returned by debci along with the results each run.
1630 self.save_pending_json()
1631 else:
1632 # for file-based submission, triggers are space separated
1633 params["triggers"] = [" ".join(params["triggers"])]
1634 assert self.amqp_file_handle
1635 self.amqp_file_handle.write(f"{qname}:{src} {json.dumps(params)}\n")
1637 def pkg_test_request(
1638 self, src: str, arch: str, all_triggers: list[str], huge: bool = False
1639 ) -> None:
1640 """Request one package test for a set of triggers
1642 all_triggers is a list of "pkgname/version". These are the packages
1643 that will be taken from the source suite. The first package in this
1644 list is the package that triggers the testing of src, the rest are
1645 additional packages required for installability of the test deps. If
1646 huge is true, then the request will be put into the -huge instead of
1647 normal queue.
1649 This will only be done if that test wasn't already requested in
1650 a previous run (i. e. if it's not already in self.pending_tests)
1651 or if there is already a fresh or a positive result for it. This
1652 ensures to download current results for this package before
1653 requesting any test."""
1654 trigger = all_triggers[0]
1655 uses_swift = not self.options.adt_swift_url.startswith("file://")
1656 try:
1657 result = self.test_results[trigger][src][arch]
1658 has_result = True
1659 except KeyError:
1660 has_result = False
1662 if has_result:
1663 result_state = result[0]
1664 if result_state in {Result.OLD_PASS, Result.OLD_FAIL, Result.OLD_NEUTRAL}:
1665 pass
1666 elif (
1667 result_state is Result.FAIL
1668 and self.result_in_baseline(src, arch)[0]
1669 in {Result.PASS, Result.NEUTRAL, Result.OLD_PASS, Result.OLD_NEUTRAL}
1670 and self._now - result[3] > self.options.adt_retry_older_than
1671 ):
1672 # We might want to retry this failure, so continue
1673 pass
1674 elif not uses_swift:
1675 # We're done if we don't retrigger and we're not using swift
1676 return
1677 elif result_state in {Result.PASS, Result.NEUTRAL}:
1678 self.logger.debug(
1679 "%s/%s triggered by %s already known", src, arch, trigger
1680 )
1681 return
1683 # Without swift we don't expect new results
1684 if uses_swift:
1685 self.logger.info(
1686 "Checking for new results for failed %s/%s for trigger %s",
1687 src,
1688 arch,
1689 trigger,
1690 )
1691 self.fetch_swift_results(self.options.adt_swift_url, src, arch)
1692 # do we have one now?
1693 try:
1694 self.test_results[trigger][src][arch]
1695 return
1696 except KeyError:
1697 pass
1699 self.request_test_if_not_queued(src, arch, trigger, all_triggers, huge=huge)
1701 def request_test_if_not_queued(
1702 self,
1703 src: str,
1704 arch: str,
1705 trigger: str,
1706 all_triggers: list[str] = [],
1707 huge: bool = False,
1708 ) -> None:
1709 assert self.pending_tests is not None # for type checking
1710 if not all_triggers:
1711 all_triggers = [trigger]
1713 # Don't re-request if it's already pending
1714 arch_dict = self.pending_tests.setdefault(trigger, {}).setdefault(src, {})
1715 if arch in arch_dict:
1716 self.logger.debug(
1717 "Test %s/%s for %s is already pending, not queueing", src, arch, trigger
1718 )
1719 else:
1720 self.logger.debug(
1721 "Requesting %s autopkgtest on %s to verify %s", src, arch, trigger
1722 )
1723 arch_dict[arch] = self._now
1724 self.send_test_request(src, arch, all_triggers, huge=huge)
1726 def result_in_baseline(self, src: str, arch: str) -> list[Any]:
1727 """Get the result for src on arch in the baseline
1729 The baseline is optionally all data or a reference set)
1730 """
1732 # this requires iterating over all cached results and thus is expensive;
1733 # cache the results
1734 try:
1735 return self.result_in_baseline_cache[src][arch]
1736 except KeyError:
1737 pass
1739 result_reference: list[Any] = [Result.NONE, None, "", 0]
1740 if self.options.adt_baseline == "reference":
1741 if src not in self.suite_info.target_suite.sources: 1741 ↛ 1742line 1741 didn't jump to line 1742 because the condition on line 1741 was never true
1742 return result_reference
1744 try:
1745 result_reference = self.test_results[REF_TRIG][src][arch]
1746 self.logger.debug(
1747 "Found result for src %s in reference: %s",
1748 src,
1749 result_reference[0].name,
1750 )
1751 except KeyError:
1752 self.logger.debug(
1753 "Found NO result for src %s in reference: %s",
1754 src,
1755 result_reference[0].name,
1756 )
1757 self.result_in_baseline_cache[src][arch] = deepcopy(result_reference)
1758 return result_reference
1760 result_ever: list[Any] = [Result.FAIL, None, "", 0]
1761 for srcmap in self.test_results.values():
1762 try:
1763 if srcmap[src][arch][0] is not Result.FAIL:
1764 result_ever = srcmap[src][arch]
1765 # If we are not looking at a reference run, We don't really
1766 # care about anything except the status, so we're done
1767 # once we find a PASS.
1768 if result_ever[0] is Result.PASS:
1769 break
1770 except KeyError:
1771 pass
1773 self.result_in_baseline_cache[src][arch] = deepcopy(result_ever)
1774 self.logger.debug("Result for src %s ever: %s", src, result_ever[0].name)
1775 return result_ever
1777 def has_test_in_target(self, src: str) -> bool:
1778 test_in_target = False
1779 try:
1780 srcinfo = self.suite_info.target_suite.sources[src]
1781 if has_autodep8_or_autopkgtest(srcinfo):
1782 test_in_target = True
1783 # AttributeError is only needed for the test suite as
1784 # srcinfo can be a NoneType
1785 except (KeyError, AttributeError):
1786 pass
1788 return test_in_target
1790 def pkg_test_result(
1791 self, src: str, ver: str, arch: str, trigger: str
1792 ) -> tuple[str, str, str | None, str]:
1793 """Get current test status of a particular package
1795 Return (status, real_version, run_id, log_url) tuple; status is a key in
1796 EXCUSES_LABELS. run_id is None if the test is still running.
1797 """
1798 assert self.pending_tests is not None # for type checking
1799 # determine current test result status
1800 run_id = None
1801 try:
1802 r = self.test_results[trigger][src][arch]
1803 ver = r[1]
1804 run_id = r[2]
1806 if r[0] in {Result.FAIL, Result.OLD_FAIL}:
1807 # determine current test result status
1808 baseline_result = self.result_in_baseline(src, arch)[0]
1810 # Special-case triggers from linux-meta*: we cannot compare
1811 # results against different kernels, as e. g. a DKMS module
1812 # might work against the default kernel but fail against a
1813 # different flavor; so for those, ignore the "ever
1814 # passed" check; FIXME: check against trigsrc only
1815 if self.options.adt_baseline != "reference" and (
1816 trigger.startswith("linux-meta") or trigger.startswith("linux/")
1817 ):
1818 baseline_result = Result.FAIL
1820 # Check if the autopkgtest (still) exists in the target suite
1821 test_in_target = self.has_test_in_target(src)
1823 if test_in_target and baseline_result in {
1824 Result.NONE,
1825 Result.OLD_FAIL,
1826 Result.OLD_NEUTRAL,
1827 Result.OLD_PASS,
1828 }:
1829 self.request_test_if_not_queued(src, arch, REF_TRIG)
1831 if self.has_force_badtest(src, ver, arch):
1832 result = "IGNORE-FAIL"
1833 elif not test_in_target:
1834 if self.options.adt_ignore_failure_for_new_tests:
1835 result = "IGNORE-FAIL"
1836 else:
1837 result = r[0].name
1838 elif baseline_result in {Result.FAIL, Result.OLD_FAIL}:
1839 result = "ALWAYSFAIL"
1840 elif baseline_result is Result.NONE: 1840 ↛ 1841line 1840 didn't jump to line 1841 because the condition on line 1840 was never true
1841 result = "RUNNING-REFERENCE"
1842 else:
1843 result = "REGRESSION"
1845 else:
1846 result = r[0].name
1848 url = self.format_log_url(src, arch, run_id)
1849 except KeyError:
1850 # no result for src/arch; still running?
1851 assert arch in self.pending_tests.get(trigger, {}).get(src, {}).keys(), (
1852 "Result for %s/%s/%s (triggered by %s) is neither known nor pending!"
1853 % (src, ver, arch, trigger)
1854 )
1856 if self.has_force_badtest(src, ver, arch):
1857 result = "RUNNING-IGNORE"
1858 else:
1859 if self.has_test_in_target(src):
1860 baseline_result = self.result_in_baseline(src, arch)[0]
1861 if baseline_result is Result.FAIL:
1862 result = "RUNNING-ALWAYSFAIL"
1863 else:
1864 result = "RUNNING"
1865 else:
1866 if self.options.adt_ignore_failure_for_new_tests:
1867 result = "RUNNING-IGNORE"
1868 else:
1869 result = "RUNNING"
1870 url = self.options.adt_ci_url + "status/pending"
1872 return (result, ver, run_id, url)
1874 def has_force_badtest(self, src: str, ver: str, arch: str) -> bool:
1875 """Check if src/ver/arch has a force-badtest hint"""
1877 assert self.hints is not None
1878 for hint in self.hints.search("force-badtest", package=src):
1879 if any(
1880 mi
1881 for mi in hint.packages
1882 if mi.architecture in ("source", arch)
1883 and (
1884 mi.version is None
1885 or mi.version == "all" # Historical unversioned hint
1886 or apt_pkg.version_compare(ver, mi.version) <= 0
1887 )
1888 ):
1889 self.logger.info(
1890 "Checking hints for %s/%s/%s: %s",
1891 src,
1892 arch,
1893 ver,
1894 hint,
1895 )
1896 return True
1898 return False
1900 def has_built_on_this_arch_or_is_arch_all(
1901 self, src_data: SourcePackage, arch: str
1902 ) -> bool:
1903 """When a source builds arch:all binaries, those binaries are
1904 added to all architectures and thus the source 'exists'
1905 everywhere. This function checks if the source has any arch
1906 specific binaries on this architecture and if not, if it
1907 has them on any architecture.
1908 """
1909 packages_s_a = self.suite_info.primary_source_suite.binaries[arch]
1910 has_unknown_binary = False
1911 for binary_s in filter_out_faux_gen(src_data.binaries):
1912 try:
1913 binary_u = packages_s_a[binary_s.package_name]
1914 except KeyError:
1915 # src_data.binaries has all the built binaries, so if
1916 # we get here, we know that at least one architecture
1917 # has architecture specific binaries
1918 has_unknown_binary = True
1919 continue
1920 if binary_u.architecture == arch:
1921 return True
1922 # If we get here, we have only seen arch:all packages for this
1923 # arch.
1924 return not has_unknown_binary