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

1import optparse 

2import os 

3from typing import TYPE_CHECKING, Any, Optional 

4from urllib.parse import quote 

5 

6import yaml 

7 

8from britney2 import PackageId, SuiteClass 

9from britney2.hints import HintAnnotate, HintType 

10from britney2.migrationitem import MigrationItem 

11from britney2.policies import ApplySrcPolicy, PolicyVerdict 

12from britney2.policies.policy import AbstractBasePolicy 

13from britney2.utils import iter_except, parse_option 

14 

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 

20 

21 

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]]] = {} 

31 

32 # Default values for this policy's options 

33 parse_option(options, "lintian_url") 

34 

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 ) 

42 

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

44 super().initialise(britney) 

45 source_suite = self.suite_info.primary_source_suite 

46 try: 

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

48 except AttributeError as e: # pragma: no cover 

49 raise RuntimeError( 

50 "Please set STATE_DIR in the britney configuration" 

51 ) from e 

52 

53 self._read_lintian_status(filename) 

54 

55 def apply_src_policy_impl( 

56 self, 

57 lintian_info: dict[str, Any], 

58 item: MigrationItem, 

59 source_data_tdist: Optional["SourcePackage"], 

60 source_data_srcdist: "SourcePackage", 

61 excuse: "Excuse", 

62 ) -> PolicyVerdict: 

63 verdict = PolicyVerdict.PASS 

64 

65 source_name = item.package 

66 src_suite_bins = self.suite_info.primary_source_suite.all_binaries_in_suite 

67 src_archs = set() 

68 for pkg_id in source_data_srcdist.binaries: 

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

70 

71 if self.options.lintian_url: 71 ↛ 75line 71 didn't jump to line 75 because the condition on line 71 was always true

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

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

74 else: 

75 url = None 

76 url_html = "" 

77 

78 # skip until a new package is built somewhere 

79 if not source_data_srcdist.binaries: 

80 verdict = PolicyVerdict.REJECTED_TEMPORARILY 

81 self.logger.debug( 

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

83 source_name, 

84 ) 

85 excuse.add_verdict_info(verdict, "nothing built yet, lintian delayed") 

86 lintian_info["result"] = "not built" 

87 else: 

88 assert item.version # for type checking 

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

90 try: 

91 results = self._lintian[src_pkg_id] 

92 except KeyError: 

93 verdict = PolicyVerdict.REJECTED_TEMPORARILY 

94 self.logger.debug( 

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

96 source_name, 

97 ) 

98 excuse.add_verdict_info( 

99 verdict, 

100 f"Waiting for lintian test results (stalls migration){url_html}", 

101 ) 

102 lintian_info["result"] = "no data available" 

103 

104 if not verdict.is_rejected: 

105 if results[0] == "FAILED": 

106 verdict = PolicyVerdict.REJECTED_PERMANENTLY 

107 self.logger.debug( 

108 "lintian failed on %s", 

109 source_name, 

110 ) 

111 excuse.add_verdict_info( 

112 verdict, 

113 f"lintian crashed and produced no output, please contact " 

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

115 ) 

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

117 elif results[0] == "TAGS": 

118 verdict = PolicyVerdict.REJECTED_PERMANENTLY 

119 lintian_info["result"] = "triggered tags" 

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

121 self.logger.debug( 

122 "%s triggered lintian tags", 

123 source_name, 

124 ) 

125 excuse.add_verdict_info( 

126 verdict, 

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

128 ) 

129 elif src_archs - results[1]: 

130 # britney has more architecture 

131 verdict = PolicyVerdict.REJECTED_TEMPORARILY 

132 archs_missing = src_archs - results[1] 

133 lintian_info["result"] = "lintian misses archs" 

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

135 self.logger.debug( 

136 "%s lintian misses architectures", 

137 source_name, 

138 ) 

139 excuse.add_verdict_info( 

140 verdict, 

141 f"Waiting for lintian test results on {', '.join(archs_missing)}{url_html}", 

142 ) 

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

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

145 lintian_info["result"] = "lintian saw more archs" 

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

147 else: 

148 pass 

149 

150 if verdict.is_rejected: 

151 assert self.hints is not None 

152 hints = self.hints.search( 

153 "ignore-lintian", 

154 package=source_name, 

155 version=source_data_srcdist.version, 

156 ) 

157 if hints: 

158 verdict = PolicyVerdict.PASS_HINTED 

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

160 "issued-by", hints[0].user 

161 ) 

162 excuse.addinfo( 

163 f"Ignoring lintian issues as requested by {hints[0].user}" 

164 ) 

165 

166 if verdict != PolicyVerdict.PASS: 

167 lintian_info["url"] = url 

168 

169 return verdict 

170 

171 def _read_lintian_status( 

172 self, filename: str 

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

174 summary = self._lintian 

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

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

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

178 return summary 

179 data = yaml.safe_load(fd) 

180 

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

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

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

184 # TODO: might want to enumerate potential values 

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

186 summary[pkg_id] = ("FAILED", set()) 

187 elif info["tags"]: 

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

189 else: 

190 archs = info["architectures"] 

191 assert "source" in archs, ( 

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

193 f"missing for {pkgver}" 

194 ) 

195 summary[pkg_id] = ("ARCHS", set(archs.split()) - {"source"}) 

196 

197 return summary