Coverage for dak/init_dirs.py: 78%

1"""Initial setup of an archive.""" 

2 

3# Copyright (C) 2002, 2004, 2006 James Troup <james@nocrew.org> 

4 

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

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

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

8# (at your option) any later version. 

9 

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

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

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

13# GNU General Public License for more details. 

14 

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

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

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

18 

19################################################################################ 

20 

21import os 

22import sys 

23 

24import apt_pkg 

25from sqlalchemy import select 

26 

27from daklib import utils 

28from daklib.dbconn import Component, DBConn, Keyring, PolicyQueue, Suite 

29 

30################################################################################ 

31 

32Cnf: apt_pkg.Configuration 

33 

34################################################################################ 

35 

36 

37def usage(exit_code=0): 

38 """Print a usage message and exit with 'exit_code'.""" 

39 

40 print( 

41 """Usage: dak init-dirs 

42Creates directories for an archive based on dak.conf configuration file. 

43 

44 -h, --help show this help and exit.""" 

45 ) 

46 sys.exit(exit_code) 

47 

48 

49################################################################################ 

50 

51 

52def do_dir(target, config_name): 

53 """If 'target' exists, make sure it is a directory. If it doesn't, create 

54 it.""" 

55 

56 if os.path.exists(target): 

57 if not os.path.isdir(target): 57 ↛ 58line 57 didn't jump to line 58 because the condition on line 57 was never true

58 utils.fubar("%s (%s) is not a directory." % (target, config_name)) 

59 else: 

60 print("Creating {} ...".format(target)) 

61 os.makedirs(target) 

62 

63 

64def process_file(config, config_name): 

65 """Create directories for a config entry that's a filename.""" 

66 

67 if config_name in config: 

68 target = os.path.dirname(config[config_name]) 

69 do_dir(target, config_name) 

70 

71 

72def process_tree(config, tree): 

73 """Create directories for a config tree.""" 

74 

75 for entry in config.subtree(tree).list(): 

76 entry = entry.lower() 

77 config_name = "%s::%s" % (tree, entry) 

78 target = config[config_name] 

79 do_dir(target, config_name) 

80 

81 

82def process_morguesubdir(subdir): 

83 """Create directories for morgue sub directories.""" 

84 

85 config_name = "%s::MorgueSubDir" % (subdir) 

86 if config_name in Cnf: 86 ↛ 87line 86 didn't jump to line 87 because the condition on line 86 was never true

87 target = os.path.join(Cnf["Dir::Morgue"], Cnf[config_name]) 

88 do_dir(target, config_name) 

89 

90 

91def process_keyring(fullpath, secret=False): 

92 """Create empty keyring if necessary.""" 

93 

94 if os.path.exists(fullpath): 94 ↛ 95line 94 didn't jump to line 95 because the condition on line 94 was never true

95 return 

96 

97 keydir = os.path.dirname(fullpath) 

98 

99 if not os.path.isdir(keydir): 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true

100 print("Creating {} ...".format(keydir)) 

101 os.makedirs(keydir) 

102 if secret: 

103 # Make sure secret keyring directories are 0700 

104 os.chmod(keydir, 0o700) 

105 

106 # Touch the file 

107 print("Creating {} ...".format(fullpath)) 

108 with open(fullpath, "w") as fh: 

109 if secret: 109 ↛ 110line 109 didn't jump to line 110 because the condition on line 109 was never true

110 os.fchmod(fh.fileno(), 0o600) 

111 else: 

112 os.fchmod(fh.fileno(), 0o644) 

113 

114 

115###################################################################### 

116 

117 

118def create_directories(): 

119 """Create directories referenced in dak.conf.""" 

120 

121 session = DBConn().session() 

122 

123 # Process directories from dak.conf 

124 process_tree(Cnf, "Dir") 

125 

126 # Hardcode creation of the unchecked directory 

127 if "Dir::Base" in Cnf: 127 ↛ 133line 127 didn't jump to line 133 because the condition on line 127 was always true

128 do_dir( 

129 os.path.join(Cnf["Dir::Base"], "queue", "unchecked"), "unchecked directory" 

130 ) 

131 

132 # Process queue directories 

133 for queue in session.scalars(select(PolicyQueue)): 

134 do_dir(queue.path, "%s queue" % queue.queue_name) 

135 # If we're doing the NEW queue, make sure it has a COMMENTS directory 

136 if queue.queue_name == "new": 

137 do_dir( 

138 os.path.join(queue.path, "COMMENTS"), 

139 "%s queue comments" % queue.queue_name, 

140 ) 

141 

142 for config_name in ["Rm::LogFile", "Import-Archive::ExportDir"]: 

143 process_file(Cnf, config_name) 

144 

145 for subdir in ["Clean-Queues", "Clean-Suites"]: 

146 process_morguesubdir(subdir) 

147 

148 suite_suffix = "%s" % (Cnf.find("Dinstall::SuiteSuffix")) 

149 

150 if "Dinstall::SigningPubKeyring" in Cnf: 150 ↛ 151line 150 didn't jump to line 151 because the condition on line 150 was never true

151 process_keyring(Cnf["Dinstall::SigningPubKeyring"], secret=True) 

152 

153 # Process public keyrings 

154 for keyring_name in session.scalars( 

155 select(Keyring.keyring_name).filter_by(active=True) 

156 ): 

157 process_keyring(keyring_name) 

158 

159 # Process dists directories 

160 # TODO: Store location of each suite in database 

161 for suite in session.scalars(select(Suite)): 

162 suite_dir = os.path.join( 

163 suite.archive.path, "dists", suite.suite_name, suite_suffix 

164 ) 

165 

166 # TODO: Store valid suite/component mappings in database 

167 for component_name in session.scalars(select(Component.component_name)): 

168 

169 sc_dir = os.path.join(suite_dir, component_name) 

170 

171 do_dir(sc_dir, "%s/%s" % (suite.suite_name, component_name)) 

172 

173 for arch in suite.architectures: 173 ↛ 174line 173 didn't jump to line 174 because the loop on line 173 never started

174 if arch.arch_string == "source": 

175 arch_string = "source" 

176 else: 

177 arch_string = "binary-%s" % arch.arch_string 

178 

179 suite_arch_dir = os.path.join(sc_dir, arch_string) 

180 

181 do_dir( 

182 suite_arch_dir, 

183 "%s/%s/%s" % (suite.suite_name, component_name, arch_string), 

184 ) 

185 

186 

187################################################################################ 

188 

189 

190def main(): 

191 """Initial setup of an archive.""" 

192 

193 global Cnf 

194 

195 Cnf = utils.get_conf() 

196 arguments = [("h", "help", "Init-Dirs::Options::Help")] 

197 for i in ["help"]: 

198 key = "Init-Dirs::Options::%s" % i 

199 if key not in Cnf: 199 ↛ 197line 199 didn't jump to line 197 because the condition on line 199 was always true

200 Cnf[key] = "" # type: ignore[index] 

201 

202 arguments = apt_pkg.parse_commandline(Cnf, arguments, sys.argv) # type: ignore[attr-defined] 

203 

204 options = Cnf.subtree("Init-Dirs::Options") # type: ignore[attr-defined] 

205 if options["Help"]: 

206 usage() 

207 elif arguments: 207 ↛ 208line 207 didn't jump to line 208 because the condition on line 207 was never true

208 utils.warn("dak init-dirs takes no arguments.") 

209 usage(exit_code=1) 

210 

211 create_directories()