Coverage for britney2/policies/lintian.py: 92%
106 statements
« prev ^ index » next coverage.py v7.6.0, created at 2026-04-19 18:02 +0000
« prev ^ index » next coverage.py v7.6.0, created at 2026-04-19 18:02 +0000
1import optparse
2import os
3from typing import TYPE_CHECKING, Any, Optional
4from urllib.parse import quote
6import yaml
8from britney2 import PackageId, SuiteClass
9from britney2.hints import HintAnnotate, HintType
10from britney2.migrationitem import MigrationItem
11from britney2.policies import PolicyVerdict
12from britney2.policies.policy import AbstractBasePolicy
13from britney2.utils import binaries_from_source_version, filter_out_faux, parse_option
15if TYPE_CHECKING: 15 ↛ 16line 15 didn't jump to line 16 because the condition on line 15 was never true
16 from .. import SourcePackage, Suites
17 from ..britney import Britney
18 from ..excuse import Excuse
19 from ..hints import HintParser
22class LintianPolicy(AbstractBasePolicy):
23 def __init__(self, options: optparse.Values, suite_info: "Suites") -> None:
24 super().__init__(
25 "lintian",
26 options,
27 suite_info,
28 {SuiteClass.PRIMARY_SOURCE_SUITE},
29 )
30 self._lintian: dict[PackageId, tuple[str, set[str]]] = {}
32 # Default values for this policy's options
33 parse_option(options, "lintian_url")
35 def register_hints(self, hint_parser: "HintParser") -> None:
36 hint_parser.register_hint_type(
37 HintType(
38 "ignore-lintian",
39 versioned=HintAnnotate.OPTIONAL,
40 )
41 )
43 def initialise(self, britney: "Britney") -> None:
44 super().initialise(britney)
45 try:
46 filename = os.path.join(self.state_dir, "lintian.yaml")
47 except AttributeError as e: # pragma: no cover
48 raise RuntimeError(
49 "Please set STATE_DIR in the britney configuration"
50 ) from e
52 self._read_lintian_status(filename)
54 def apply_src_policy_impl(
55 self,
56 lintian_info: dict[str, Any],
57 source_data_tdist: Optional["SourcePackage"],
58 source_data_srcdist: "SourcePackage",
59 excuse: "Excuse",
60 ) -> PolicyVerdict:
61 verdict = PolicyVerdict.PASS
63 item = excuse.item
64 source_name = item.package
65 src_suite_bins = self.suite_info.primary_source_suite.all_binaries_in_suite
66 src_archs = set()
67 for pkg_id in filter_out_faux(source_data_srcdist.binaries):
68 src_archs.add(src_suite_bins[pkg_id].architecture)
70 if self.options.lintian_url:
71 url = self.options.lintian_url.format(package=quote(source_name))
72 url_html = f' - <a href="{url}">info</a>'
73 else:
74 url = None
75 url_html = ""
77 # skip until a new package is built somewhere
78 if not binaries_from_source_version(source_data_srcdist, self.suite_info):
79 verdict = PolicyVerdict.REJECTED_TEMPORARILY
80 self.logger.debug(
81 "%s hasn't been built anywhere, skipping lintian policy",
82 source_name,
83 )
84 excuse.add_verdict_info(verdict, "Lintian check deferred: missing builds")
85 lintian_info["result"] = "not built"
86 else:
87 assert item.version # for type checking
88 src_pkg_id = PackageId(item.package, item.version, "source")
89 try:
90 results = self._lintian[src_pkg_id]
91 except KeyError:
92 verdict = PolicyVerdict.REJECTED_TEMPORARILY
93 self.logger.debug(
94 "%s doesn't have lintian results yet",
95 source_name,
96 )
97 excuse.add_verdict_info(
98 verdict,
99 f"Lintian check waiting for test results{url_html}",
100 )
101 lintian_info["result"] = "no data available"
103 if not verdict.is_rejected:
104 if results[0] == "FAILED":
105 verdict = PolicyVerdict.REJECTED_PERMANENTLY
106 self.logger.debug(
107 "lintian failed on %s",
108 source_name,
109 )
110 excuse.add_verdict_info(
111 verdict,
112 f"Lintian crashed and produced no output, please contact "
113 f"{self.options.distribution}-release for a hint{url_html}",
114 )
115 lintian_info["result"] = "lintian failed"
116 elif results[0] == "TAGS":
117 verdict = PolicyVerdict.REJECTED_PERMANENTLY
118 lintian_info["result"] = "triggered tags"
119 lintian_info["tags"] = sorted(results[1])
120 self.logger.debug(
121 "%s triggered lintian tags",
122 source_name,
123 )
124 excuse.add_verdict_info(
125 verdict,
126 f"Lintian triggered tags: {', '.join(results[1])}{url_html}",
127 )
128 elif src_archs - results[1]:
129 # britney has more architecture
130 verdict = PolicyVerdict.REJECTED_TEMPORARILY
131 archs_missing = src_archs - results[1]
132 lintian_info["result"] = "lintian misses archs"
133 lintian_info["archs"] = ", ".join(sorted(archs_missing))
134 self.logger.debug(
135 "%s lintian misses architectures",
136 source_name,
137 )
138 excuse.add_verdict_info(
139 verdict,
140 f"Lintian check waiting for test results on {', '.join(archs_missing)}{url_html}",
141 )
142 elif results[1] - src_archs: 142 ↛ 144line 142 didn't jump to line 144 because the condition on line 142 was never true
143 # lintian had more architecutes than britney, e.g. because of new/old architecture
144 lintian_info["result"] = "lintian saw more archs"
145 lintian_info["archs"] = ", ".join(results[1] - src_archs)
146 else:
147 pass
149 if verdict.is_rejected:
150 assert self.hints is not None
151 hints = self.hints.search(
152 "ignore-lintian",
153 package=source_name,
154 version=source_data_srcdist.version,
155 )
156 if hints:
157 verdict = PolicyVerdict.PASS_HINTED
158 lintian_info.setdefault("ignored-lintian", {}).setdefault(
159 "issued-by", hints[0].user
160 )
161 excuse.addinfo(
162 f"Lintian issues ignored as requested by {hints[0].user}"
163 )
165 if verdict != PolicyVerdict.PASS:
166 lintian_info["url"] = url
168 return verdict
170 def _read_lintian_status(
171 self, filename: str
172 ) -> dict[PackageId, tuple[str, set[str]]]:
173 summary = self._lintian
174 self.logger.debug("Loading lintian status from %s", filename)
175 with open(filename) as fd: 175 ↛ exitline 175 didn't return from function '_read_lintian_status' because the return on line 177 wasn't executed
176 if os.fstat(fd.fileno()).st_size < 1: 176 ↛ 177line 176 didn't jump to line 177 because the condition on line 176 was never true
177 return summary
178 data = yaml.safe_load(fd)
180 for pkgver, info in data.items():
181 pkg, ver = pkgver.split("/")
182 pkg_id = PackageId(pkg, ver, "source")
183 # TODO: might want to enumerate potential values
184 if "status" in info.keys() and info["status"] == "lintian failed":
185 summary[pkg_id] = ("FAILED", set())
186 elif info["tags"]:
187 summary[pkg_id] = ("TAGS", set(info["tags"].keys()))
188 else:
189 archs = info["architectures"]
190 assert "source" in archs, (
191 "LintianPolicy expects at least source as architecure in lintian.yaml, "
192 f"missing for {pkgver}"
193 )
194 summary[pkg_id] = ("ARCHS", set(archs.split()) - {"source"})
196 return summary