Coverage for dak/external_overrides.py: 21%

1#! /usr/bin/env python3 

2 

3""" 

4Modify external overrides. 

5 

6@contact: Debian FTP Master <ftpmaster@debian.org> 

7@copyright: 2011 Ansgar Burchardt <ansgar@debian.org> 

8@license: GNU General Public License version 2 or later 

9""" 

10 

11# This program is free software; you can redistribute it and/or modify 

12# it under the terms of the GNU General Public License as published by 

13# the Free Software Foundation; either version 2 of the License, or 

14# (at your option) any later version. 

15 

16# This program is distributed in the hope that it will be useful, 

17# but WITHOUT ANY WARRANTY; without even the implied warranty of 

18# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

19# GNU General Public License for more details. 

20 

21# You should have received a copy of the GNU General Public License 

22# along with this program; if not, write to the Free Software 

23# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 

24 

25import sys 

26from collections.abc import Iterable, Iterator 

27from typing import NoReturn 

28 

29import apt_pkg 

30from sqlalchemy import sql 

31 

32from daklib.config import Config 

33from daklib.dbconn import DBConn, ExternalOverride, get_component, get_suite 

34 

35 

36def usage() -> NoReturn: 

37 print( 

38 """Usage: dak external-overrides COMMAND 

39Modify external overrides. 

40 

41 -h, --help show this help and exit. 

42 -f, --force allow processing of untouchable suites. 

43 

44Commands can use a long or abbreviated form: 

45 

46 import SUITE COMPONENT KEY import external overrides for KEY 

47 i SUITE COMPONENT KEY NOTE: This will replace existing overrides. 

48 

49 copy FROM TO copy external overrides from suite FROM to TO 

50 NOTE: Needs --force for untouchable TO 

51 

52For the 'import' command, external overrides are read from standard input and 

53should be given as lines of the form 'PACKAGE KEY VALUE'. 

54""" 

55 ) 

56 sys.exit() 

57 

58 

59############################################################################# 

60 

61 

62class ExternalOverrideReader: 

63 """ 

64 Parses an external override file 

65 """ 

66 

67 def __init__(self, fh: Iterable[str]): 

68 self.fh = fh 

69 self.package: str = "" 

70 self.key: str = "" 

71 self.value: list[str] = [] 

72 

73 def _flush(self) -> tuple[str, str, str]: 

74 """ 

75 Return the parsed line that is being built and start parsing a new line 

76 """ 

77 res = self.package, self.key, "\n".join(self.value) 

78 self.package = self.key = "" 

79 self.value = [] 

80 return res 

81 

82 def __iter__(self) -> Iterator[tuple[str, str, str]]: 

83 """ 

84 returns a (package, key, value) tuple for every entry in the external 

85 override file 

86 """ 

87 for line in self.fh: 

88 if not line: 

89 continue 

90 if line[0] in (" ", "\t"): 

91 # Continuation line 

92 self.value.append(line.rstrip()) 

93 else: 

94 if self.package is not None: 

95 yield self._flush() 

96 

97 # New line 

98 (self.package, self.key, value) = line.rstrip().split(None, 2) 

99 self.value = [value] 

100 

101 if self.package: 

102 yield self._flush() 

103 

104 

105############################################################################# 

106 

107 

108def external_overrides_copy( 

109 from_suite_name: str, to_suite_name: str, force=False 

110) -> None: 

111 session = DBConn().session() 

112 

113 from_suite = get_suite(from_suite_name, session) 

114 to_suite = get_suite(to_suite_name, session) 

115 

116 if from_suite is None: 

117 print("E: source %s not found." % from_suite_name) 

118 session.rollback() 

119 return 

120 if to_suite is None: 

121 print("E: target %s not found." % to_suite_name) 

122 session.rollback() 

123 return 

124 

125 if not force and to_suite.untouchable: 

126 print("E: refusing to touch untouchable suite %s (not forced)." % to_suite_name) 

127 session.rollback() 

128 return 

129 

130 session.query(ExternalOverride).filter_by(suite=to_suite).delete() 

131 session.execute( 

132 sql.text( 

133 """ 

134 INSERT INTO external_overrides (suite, component, package, key, value) 

135 SELECT :to_suite, component, package, key, value FROM external_overrides WHERE suite = :from_suite 

136 """ 

137 ), 

138 {"from_suite": from_suite.suite_id, "to_suite": to_suite.suite_id}, 

139 ) 

140 

141 session.commit() 

142 

143 

144def external_overrides_import( 

145 suite_name: str, component_name: str, key: str, file: Iterable[str], force=False 

146) -> None: 

147 session = DBConn().session() 

148 

149 suite = get_suite(suite_name, session) 

150 assert suite is not None 

151 component = get_component(component_name, session) 

152 assert component is not None 

153 

154 if not force and suite.untouchable: 

155 print("E: refusing to touch untouchable suite %s (not forced)." % suite_name) 

156 session.rollback() 

157 return 

158 

159 session.query(ExternalOverride).filter_by( 

160 suite=suite, component=component, key=key 

161 ).delete() 

162 

163 for package, key, value in ExternalOverrideReader(file): 

164 eo = ExternalOverride() 

165 eo.suite = suite 

166 eo.component = component 

167 eo.package = package 

168 eo.key = key 

169 eo.value = value 

170 session.add(eo) 

171 

172 session.commit() 

173 

174 

175############################################################################# 

176 

177 

178def main() -> None: 

179 cnf = Config() 

180 

181 Arguments = [ 

182 ("h", "help", "External-Overrides::Options::Help"), 

183 ("f", "force", "External-Overrides::Options::Force"), 

184 ] 

185 

186 args = apt_pkg.parse_commandline(cnf.Cnf, Arguments, sys.argv) # type: ignore[attr-defined] 

187 try: 

188 Options = cnf.subtree("External-Overrides::Options") 

189 except KeyError: 

190 Options = {} 

191 

192 if "Help" in Options: 192 ↛ 195line 192 didn't jump to line 195 because the condition on line 192 was always true

193 usage() 

194 

195 force = False 

196 if "Force" in Options and Options["Force"]: 

197 force = True 

198 

199 command = args[0] 

200 if command in ("import", "i"): 

201 external_overrides_import(args[1], args[2], args[3], sys.stdin, force) 

202 elif command in ("copy", "c"): 

203 external_overrides_copy(args[1], args[2], force) 

204 else: 

205 print("E: Unknown commands.") 

206 

207 

208if __name__ == "__main__": 

209 main()