Coverage for britney2/policies/lintian.py: 93%

1import optparse 

2import os 

3from enum import Enum, StrEnum, auto 

4from typing import TYPE_CHECKING, Any, Optional 

5from urllib.parse import quote 

6 

7import yaml 

8 

9from britney2 import PackageId, SuiteClass 

10from britney2.hints import HintAnnotate, HintType 

11from britney2.migrationitem import MigrationItem 

12from britney2.policies import PolicyVerdict 

13from britney2.policies.policy import AbstractBasePolicy 

14from britney2.utils import ( 

15 binaries_from_source_version, 

16 filter_out_faux_gen, 

17 parse_option, 

18) 

19 

20if TYPE_CHECKING: 20 ↛ 21line 20 didn't jump to line 21 because the condition on line 20 was never true

21 from .. import SourcePackage, Suites 

22 from ..britney import Britney 

23 from ..excuse import Excuse 

24 from ..hints import HintParser 

25 

26 

27class LintianResult(Enum): 

28 FAILED = auto() 

29 TAGS = auto() 

30 ARCH = auto() 

31 

32 

33class Result(StrEnum): 

34 NOT_BUILT = "not built" 

35 NO_DATA = "no data available" 

36 TRIGGERED_TAGS = "triggered tags" 

37 MISSES_ARCHS = "lintian misses archs" 

38 SAW_MORE_ARCHS = "lintian saw more archs" 

39 

40 

41class LintianPolicy(AbstractBasePolicy): 

42 def __init__(self, options: optparse.Values, suite_info: "Suites") -> None: 

43 super().__init__( 

44 "lintian", 

45 options, 

46 suite_info, 

47 {SuiteClass.PRIMARY_SOURCE_SUITE}, 

48 ) 

49 self._lintian: dict[PackageId, tuple[LintianResult, set[str]]] = {} 

50 

51 # Default values for this policy's options 

52 parse_option(options, "lintian_url") 

53 

54 def register_hints(self, hint_parser: "HintParser") -> None: 

55 hint_parser.register_hint_type( 

56 HintType( 

57 "ignore-lintian", 

58 versioned=HintAnnotate.OPTIONAL, 

59 ) 

60 ) 

61 

62 def initialise(self, britney: "Britney") -> None: 

63 super().initialise(britney) 

64 try: 

65 filename = os.path.join(self.state_dir, "lintian.yaml") 

66 except AttributeError as e: # pragma: no cover 

67 raise RuntimeError( 

68 "Please set STATE_DIR in the britney configuration" 

69 ) from e 

70 

71 self._read_lintian_status(filename) 

72 

73 def apply_src_policy_impl( 

74 self, 

75 lintian_info: dict[str, Any], 

76 source_data_tdist: Optional["SourcePackage"], 

77 source_data_srcdist: "SourcePackage", 

78 excuse: "Excuse", 

79 ) -> PolicyVerdict: 

80 verdict = PolicyVerdict.PASS 

81 

82 item = excuse.item 

83 source_name = item.package 

84 src_suite_bins = self.suite_info.primary_source_suite.all_binaries_in_suite 

85 src_archs = set() 

86 for pkg_id in filter_out_faux_gen(source_data_srcdist.binaries): 

87 src_archs.add(src_suite_bins[pkg_id].architecture) 

88 

89 if self.options.lintian_url: 

90 url = self.options.lintian_url.format(package=quote(source_name)) 

91 url_html = f' - <a href="{url}">info</a>' 

92 else: 

93 url = None 

94 url_html = "" 

95 

96 # skip until a new package is built somewhere 

97 if not binaries_from_source_version(source_data_srcdist, self.suite_info)[0]: 

98 verdict = PolicyVerdict.REJECTED_TEMPORARILY 

99 self.logger.debug( 

100 "%s hasn't been built anywhere, skipping lintian policy", 

101 source_name, 

102 ) 

103 excuse.add_verdict_info(verdict, "Lintian check deferred: missing builds") 

104 lintian_info["result"] = Result.NOT_BUILT 

105 else: 

106 assert item.version # for type checking 

107 src_pkg_id = PackageId(item.package, item.version, "source") 

108 try: 

109 results = self._lintian[src_pkg_id] 

110 except KeyError: 

111 verdict = PolicyVerdict.REJECTED_TEMPORARILY 

112 self.logger.debug( 

113 "%s doesn't have lintian results yet", 

114 source_name, 

115 ) 

116 excuse.add_verdict_info( 

117 verdict, 

118 f"Lintian check waiting for test results{url_html}", 

119 ) 

120 lintian_info["result"] = Result.NO_DATA 

121 

122 if not verdict.is_rejected: 

123 if results[0] is LintianResult.FAILED: 

124 verdict = PolicyVerdict.REJECTED_PERMANENTLY 

125 self.logger.debug( 

126 "lintian failed on %s", 

127 source_name, 

128 ) 

129 excuse.add_verdict_info( 

130 verdict, 

131 f"Lintian crashed and produced no output, please contact " 

132 f"{self.options.distribution}-release for a hint{url_html}", 

133 ) 

134 lintian_info["result"] = "lintian failed" 

135 elif results[0] is LintianResult.TAGS: 

136 verdict = PolicyVerdict.REJECTED_PERMANENTLY 

137 lintian_info["result"] = Result.TRIGGERED_TAGS 

138 lintian_info["tags"] = sorted(results[1]) 

139 self.logger.debug( 

140 "%s triggered lintian tags", 

141 source_name, 

142 ) 

143 excuse.add_verdict_info( 

144 verdict, 

145 f"Lintian triggered tags: {', '.join(results[1])}{url_html}", 

146 ) 

147 elif src_archs - results[1]: 

148 # britney has more architecture 

149 verdict = PolicyVerdict.REJECTED_TEMPORARILY 

150 archs_missing = src_archs - results[1] 

151 lintian_info["result"] = Result.MISSES_ARCHS 

152 lintian_info["archs"] = ", ".join(sorted(archs_missing)) 

153 self.logger.debug( 

154 "%s lintian misses architectures", 

155 source_name, 

156 ) 

157 excuse.add_verdict_info( 

158 verdict, 

159 f"Lintian check waiting for test results on {', '.join(archs_missing)}{url_html}", 

160 ) 

161 elif results[1] - src_archs: 161 ↛ 163line 161 didn't jump to line 163 because the condition on line 161 was never true

162 # lintian had more architecutes than britney, e.g. because of new/old architecture 

163 lintian_info["result"] = Result.SAW_MORE_ARCHS 

164 lintian_info["archs"] = ", ".join(results[1] - src_archs) 

165 else: 

166 pass 

167 

168 if verdict.is_rejected: 

169 assert self.hints is not None 

170 if ( 

171 hint := self.hints.search_first( 

172 "ignore-lintian", 

173 package=source_name, 

174 version=source_data_srcdist.version, 

175 ) 

176 ) is not None: 

177 verdict = PolicyVerdict.PASS_HINTED 

178 lintian_info.setdefault("ignored-lintian", {}).setdefault( 

179 "issued-by", hint.user 

180 ) 

181 excuse.addinfo(f"Lintian issues ignored as requested by {hint.user}") 

182 

183 if verdict is not PolicyVerdict.PASS: 

184 lintian_info["url"] = url 

185 

186 return verdict 

187 

188 def _read_lintian_status( 

189 self, filename: str 

190 ) -> dict[PackageId, tuple[LintianResult, set[str]]]: 

191 summary = self._lintian 

192 self.logger.debug("Loading lintian status from %s", filename) 

193 with open(filename) as fd: 193 ↛ exitline 193 didn't return from function '_read_lintian_status' because the return on line 195 wasn't executed

194 if os.fstat(fd.fileno()).st_size < 1: 194 ↛ 195line 194 didn't jump to line 195 because the condition on line 194 was never true

195 return summary 

196 data = yaml.safe_load(fd) 

197 

198 empty_set: set[str] = set() 

199 for pkgver, info in data.items(): 

200 pkg, ver = pkgver.split("/") 

201 pkg_id = PackageId(pkg, ver, "source") 

202 # TODO: might want to enumerate potential values 

203 if "status" in info.keys() and info["status"] == "lintian failed": 

204 summary[pkg_id] = (LintianResult.FAILED, empty_set) 

205 elif info["tags"]: 

206 summary[pkg_id] = (LintianResult.TAGS, set(info["tags"].keys())) 

207 else: 

208 archs = info["architectures"] 

209 assert "source" in archs, ( 

210 "LintianPolicy expects at least source as architecure in lintian.yaml, " 

211 f"missing for {pkgver}" 

212 ) 

213 summary[pkg_id] = ( 

214 LintianResult.ARCH, 

215 {arch for arch in archs.split() if arch != "source"}, 

216 ) 

217 

218 return summary