Coverage for dak/external_overrides.py: 21%

1""" 

2Modify external overrides. 

3 

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

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

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

7""" 

8 

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

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

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

12# (at your option) any later version. 

13 

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

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

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

17# GNU General Public License for more details. 

18 

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

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

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

22 

23import sys 

24from collections.abc import Iterable, Iterator 

25from typing import NoReturn 

26 

27import apt_pkg 

28from sqlalchemy import delete, sql 

29 

30from daklib.config import Config 

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

32 

33 

34def usage() -> NoReturn: 

35 print( 

36 """Usage: dak external-overrides COMMAND 

37Modify external overrides. 

38 

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

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

41 

42Commands can use a long or abbreviated form: 

43 

44 import SUITE COMPONENT KEY import external overrides for KEY 

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

46 

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

48 NOTE: Needs --force for untouchable TO 

49 

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

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

52""" 

53 ) 

54 sys.exit() 

55 

56 

57############################################################################# 

58 

59 

60class ExternalOverrideReader: 

61 """ 

62 Parses an external override file 

63 """ 

64 

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

66 self.fh = fh 

67 self.package: str = "" 

68 self.key: str = "" 

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

70 

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

72 """ 

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

74 """ 

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

76 self.package = self.key = "" 

77 self.value = [] 

78 return res 

79 

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

81 """ 

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

83 override file 

84 """ 

85 for line in self.fh: 

86 if not line: 

87 continue 

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

89 # Continuation line 

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

91 else: 

92 if self.package is not None: 

93 yield self._flush() 

94 

95 # New line 

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

97 self.value = [value] 

98 

99 if self.package: 

100 yield self._flush() 

101 

102 

103############################################################################# 

104 

105 

106def external_overrides_copy( 

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

108) -> None: 

109 session = DBConn().session() 

110 

111 from_suite = get_suite(from_suite_name, session) 

112 to_suite = get_suite(to_suite_name, session) 

113 

114 if from_suite is None: 

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

116 session.rollback() 

117 return 

118 if to_suite is None: 

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

120 session.rollback() 

121 return 

122 

123 if not force and to_suite.untouchable: 

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

125 session.rollback() 

126 return 

127 

128 session.execute(delete(ExternalOverride).filter_by(suite=to_suite)) 

129 session.execute( 

130 sql.text( 

131 """ 

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

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

134 """ 

135 ), 

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

137 ) 

138 

139 session.commit() 

140 

141 

142def external_overrides_import( 

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

144) -> None: 

145 session = DBConn().session() 

146 

147 suite = get_suite(suite_name, session) 

148 assert suite is not None 

149 component = get_component(component_name, session) 

150 assert component is not None 

151 

152 if not force and suite.untouchable: 

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

154 session.rollback() 

155 return 

156 

157 session.execute( 

158 delete(ExternalOverride).filter_by(suite=suite, component=component, key=key) 

159 ) 

160 

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

162 eo = ExternalOverride() 

163 eo.suite = suite 

164 eo.component = component 

165 eo.package = package 

166 eo.key = key 

167 eo.value = value 

168 session.add(eo) 

169 

170 session.commit() 

171 

172 

173############################################################################# 

174 

175 

176def main() -> None: 

177 cnf = Config() 

178 

179 Arguments = [ 

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

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

182 ] 

183 

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

185 try: 

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

187 except KeyError: 

188 Options = {} 

189 

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

191 usage() 

192 

193 force = False 

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

195 force = True 

196 

197 command = args[0] 

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

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

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

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

202 else: 

203 print("E: Unknown commands.")