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

1import optparse 

2import os 

3import yaml 

4from typing import TYPE_CHECKING, Any, Optional 

5from urllib.parse import quote 

6 

7from britney2 import PackageId, SuiteClass 

8from britney2.hints import split_into_one_hint_per_package 

9from britney2.migrationitem import MigrationItem 

10from britney2.policies import ApplySrcPolicy, PolicyVerdict 

11from britney2.policies.policy import AbstractBasePolicy 

12from britney2.utils import iter_except, parse_option 

13 

14if TYPE_CHECKING: 14 ↛ 15line 14 didn't jump to line 15, because the condition on line 14 was never true

15 from ..britney import Britney 

16 from ..excuse import Excuse 

17 from ..hints import HintParser 

18 from .. import SourcePackage, Suites 

19 

20 

21class LintianPolicy(AbstractBasePolicy): 

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

23 super().__init__( 

24 "lintian", 

25 options, 

26 suite_info, 

27 {SuiteClass.PRIMARY_SOURCE_SUITE}, 

28 ) 

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

30 

31 # Default values for this policy's options 

32 parse_option(options, "lintian_url") 

33 

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

35 hint_parser.register_hint_type( 

36 "ignore-lintian", split_into_one_hint_per_package 

37 ) 

38 

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

40 super().initialise(britney) 

41 source_suite = self.suite_info.primary_source_suite 

42 try: 

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

44 except AttributeError as e: # pragma: no cover 

45 raise RuntimeError( 

46 "Please set STATE_DIR in the britney configuration" 

47 ) from e 

48 

49 self._read_lintian_status(filename) 

50 

51 def apply_src_policy_impl( 

52 self, 

53 lintian_info: dict[str, Any], 

54 item: MigrationItem, 

55 source_data_tdist: Optional["SourcePackage"], 

56 source_data_srcdist: "SourcePackage", 

57 excuse: "Excuse", 

58 ) -> PolicyVerdict: 

59 verdict = PolicyVerdict.PASS 

60 

61 source_name = item.package 

62 src_suite_bins = self.suite_info.primary_source_suite.all_binaries_in_suite 

63 src_archs = set() 

64 for pkg_id in source_data_srcdist.binaries: 

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

66 

67 if self.options.lintian_url: 67 ↛ 71line 67 didn't jump to line 71, because the condition on line 67 was never false

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

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

70 else: 

71 url = None 

72 url_html = "" 

73 

74 # skip until a new package is built somewhere 

75 if not source_data_srcdist.binaries: 

76 verdict = PolicyVerdict.REJECTED_TEMPORARILY 

77 self.logger.debug( 

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

79 source_name, 

80 ) 

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

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

83 else: 

84 assert item.version # for type checking 

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

86 try: 

87 results = self._lintian[src_pkg_id] 

88 except KeyError: 

89 verdict = PolicyVerdict.REJECTED_TEMPORARILY 

90 self.logger.debug( 

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

92 source_name, 

93 ) 

94 excuse.add_verdict_info( 

95 verdict, 

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

97 ) 

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

99 

100 if not verdict.is_rejected: 

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

102 verdict = PolicyVerdict.REJECTED_PERMANENTLY 

103 self.logger.debug( 

104 "lintian failed on %s", 

105 source_name, 

106 ) 

107 excuse.add_verdict_info( 

108 verdict, 

109 f"lintian failed, please contact {self.options.distribution}-release " 

110 f"for a hint{url_html}", 

111 ) 

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

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

114 verdict = PolicyVerdict.REJECTED_PERMANENTLY 

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

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

117 self.logger.debug( 

118 "%s triggered lintian tags", 

119 source_name, 

120 ) 

121 excuse.add_verdict_info( 

122 verdict, 

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

124 ) 

125 elif src_archs - results[1]: 

126 # britney has more architecture 

127 verdict = PolicyVerdict.REJECTED_TEMPORARILY 

128 archs_missing = src_archs - results[1] 

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

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

131 self.logger.debug( 

132 "%s lintian misses architectures", 

133 source_name, 

134 ) 

135 excuse.add_verdict_info( 

136 verdict, 

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

138 ) 

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

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

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

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

143 else: 

144 pass 

145 

146 if verdict.is_rejected: 

147 assert self.hints is not None 

148 hints = self.hints.search( 

149 "ignore-lintian", 

150 package=source_name, 

151 version=source_data_srcdist.version, 

152 ) 

153 if hints: 

154 verdict = PolicyVerdict.PASS_HINTED 

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

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

157 ) 

158 excuse.addinfo( 

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

160 ) 

161 

162 if verdict != PolicyVerdict.PASS: 

163 lintian_info["url"] = url 

164 

165 return verdict 

166 

167 def _read_lintian_status( 

168 self, filename: str 

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

170 summary = self._lintian 

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

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

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

174 return summary 

175 data = yaml.safe_load(fd) 

176 

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

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

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

180 # TODO: might want to enumerate potential values 

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

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

183 elif info["tags"]: 

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

185 else: 

186 archs = info["architectures"] 

187 assert "source" in archs, ( 

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

189 f"missing for {pkgver}" 

190 ) 

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

192 

193 return summary