Coverage for dak/show

1#! /usr/bin/env python3

3"""Output html for packages in NEW"""

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

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

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

9# (at your option) any later version.

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

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

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

14# GNU General Public License for more details.

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

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

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

20################################################################################

22# <elmo> I'm James Troup, long term source of all evil in Debian. you may

23# know me from such debian-devel-announce gems as "Serious

24# Problems With ...."

26################################################################################

28import os

29import sys

30import time

31from multiprocessing import Manager

32from typing import TYPE_CHECKING, NoReturn

34import apt_pkg

36import dak.examine_package

37from daklib import policy

38from daklib.config import Config

39from daklib.dakmultiprocessing import (

40 PROC_STATUS_EXCEPTION,

41 PROC_STATUS_SUCCESS,

42 DakProcessPool,

43)

44from daklib.dbconn import DBChange, DBConn, PolicyQueue, PolicyQueueUpload

46if TYPE_CHECKING:

47 from collections.abc import Iterable

49 from sqlalchemy.orm import Session

51# Globals

52cnf: Config

53Options: apt_pkg.Configuration

54manager = Manager()

55sources = manager.list()

56htmlfiles_to_process = manager.list()

57timeout_str = "Error or timeout while processing"

60################################################################################

61################################################################################

62################################################################################

65def html_header(name: str, missing: list[tuple[str, str]]) -> str:

66 if name.endswith(".changes"): 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 name = " ".join(name.split("_")[:2])

68 result = """<!DOCTYPE html>

69<html lang="en">

70 <head>

71 <meta charset="utf-8">

72 <title>%(name)s - Debian NEW package overview</title>

73 <link rel="stylesheet" href="/style.css">

74 <link rel="shortcut icon" href="https://www.debian.org/favicon.ico">

75 <script>

76 function toggle(id, initial, display) {

77 var o = document.getElementById(id);

78 toggleObj(o, initial, display);

79 }

80 function show(id, display) {

81 var o = document.getElementById(id);

82 o.style.display = 'table-row-group';

83 }

84 function toggleObj(o, initial, display) {

85 if(! o.style.display)

86 o.style.display = initial;

87 if(o.style.display == display) {

88 o.style.display = "none";

89 } else {

90 o.style.display = display;

91 }

92 }

93 </script>

94 </head>

95 <body id="NEW-details-page">

96 <div id="logo">

97 <a href="https://www.debian.org/">

98 <img src="https://www.debian.org/logos/openlogo-nd-50.png"

99 alt=""></a>

100 <a href="https://www.debian.org/">

101 <img src="https://www.debian.org/Pics/debian.png"

102 alt="Debian Project"></a>

103 </div>

104 <div id="titleblock">

105 <img src="https://www.debian.org/Pics/red-upperleft.png"

106 id="red-upperleft" alt="">

107 <img src="https://www.debian.org/Pics/red-lowerleft.png"

108 id="red-lowerleft" alt="">

109 <img src="https://www.debian.org/Pics/red-upperright.png"

110 id="red-upperright" alt="">

111 <img src="https://www.debian.org/Pics/red-lowerright.png"

112 id="red-lowerright" alt="">

113 <span class="title">

114 Debian NEW package overview for %(name)s

115 </span>

116 </div>

117

118 """ % {

119 "name": name

120 }

121

122 # we assume only one source (.dsc) per changes here

123 result += """

124 <div id="menu">

125 <p class="title">Navigation</p>

126 <p><a href="#changes" onclick="show('changes-body')">.changes</a></p>

127 <p><a href="#dsc" onclick="show('dsc-body')">.dsc</a></p>

128 <p><a href="#source-lintian" onclick="show('source-lintian-body')">source lintian</a></p>

129

130"""

131 for binarytype, packagename in [m for m in missing if m[0] in ("deb", "udeb")]:

132 result += """

133 <p class="subtitle">%(pkg)s</p>

134 <p><a href="#binary-%(pkg)s-control" onclick="show('binary-%(pkg)s-control-body')">control file</a></p>

135 <p><a href="#binary-%(pkg)s-lintian" onclick="show('binary-%(pkg)s-lintian-body')">binary lintian</a></p>

136 <p><a href="#binary-%(pkg)s-contents" onclick="show('binary-%(pkg)s-contents-body')">.deb contents</a></p>

137 <p><a href="#binary-%(pkg)s-copyright" onclick="show('binary-%(pkg)s-copyright-body')">copyright</a></p>

138 <p><a href="#binary-%(pkg)s-file-listing" onclick="show('binary-%(pkg)s-file-listing-body')">file listing</a></p>

139

140""" % {

141 "pkg": packagename

142 }

143 result += " </div>"

144 return result

145

146

147def html_footer() -> str:

148 result = """ <p class="validate">Timestamp: %s (UTC)</p>

149""" % (

150 time.strftime("%d.%m.%Y / %H:%M:%S", time.gmtime())

151 )

152 result += "</body></html>"

153 return result

154

155

156################################################################################

157

158

159def do_pkg(upload_id: int) -> tuple[int, str]:

160 cnf = Config()

161

162 session = DBConn().session()

163 upload = session.query(PolicyQueueUpload).filter_by(id=upload_id).one()

164

165 queue = upload.policy_queue

166 changes = upload.changes

167

168 origchanges = os.path.join(queue.path, changes.changesname)

169 print(origchanges)

170

171 htmlname = "{0}_{1}.html".format(changes.source, changes.version)

172 htmlfile = os.path.join(cnf["Show-New::HTMLPath"], htmlname)

173

174 # Have we already processed this?

175 if os.path.exists(htmlfile) and os.stat(htmlfile).st_mtime > time.mktime( 175 ↛ 178line 175 didn't jump to line 178 because the condition on line 175 was never true

176 changes.created.timetuple()

177 ):

178 with open(htmlfile, "r") as fd:

179 if fd.read() != timeout_str:

180 sources.append(htmlname)

181 return (PROC_STATUS_SUCCESS, "%s already up-to-date" % htmlfile)

182

183 # Go, process it... Now!

184 htmlfiles_to_process.append(htmlfile)

185 sources.append(htmlname)

186

187 group = cnf.get("Dinstall::UnprivGroup") or None

188

189 with (

190 open(htmlfile, "w") as outfile,

191 policy.UploadCopy(upload, group=group) as upload_copy,

192 ):

193 handler = policy.PolicyQueueUploadHandler(upload, session)

194 missing = [(o["type"], o["package"]) for o in handler.missing_overrides()]

195 distribution = changes.distribution

196

197 outfile.write(html_header(changes.source, missing))

198 outfile.write(dak.examine_package.display_changes(distribution, origchanges))

199

200 if upload.source is not None and ("dsc", upload.source.source) in missing: 200 ↛ 203line 200 didn't jump to line 203 because the condition on line 200 was always true

201 fn = os.path.join(upload_copy.directory, upload.source.poolfile.basename)

202 outfile.write(dak.examine_package.check_dsc(distribution, fn, session))

203 for binary in upload.binaries:

204 if (binary.binarytype, binary.package) not in missing: 204 ↛ 205line 204 didn't jump to line 205 because the condition on line 204 was never true

205 continue

206 fn = os.path.join(upload_copy.directory, binary.poolfile.basename)

207 outfile.write(dak.examine_package.check_deb(distribution, fn, session))

208

209 outfile.write(html_footer())

210

211 session.close()

212 htmlfiles_to_process.remove(htmlfile)

213 return (PROC_STATUS_SUCCESS, "{0} already updated".format(htmlfile))

214

215

216################################################################################

217

218

219def usage(exit_code=0) -> NoReturn:

220 print(

221 """Usage: dak show-new [OPTION]... [CHANGES]...

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

223 -p, --html-path [path] override output directory.

224 """

225 )

226 sys.exit(exit_code)

227

228

229################################################################################

230

231

232def init(session: "Session") -> "Iterable[PolicyQueueUpload]":

233 global cnf, Options

234

235 cnf = Config()

236

237 Arguments = [

238 ("h", "help", "Show-New::Options::Help"),

239 ("p", "html-path", "Show-New::HTMLPath", "HasArg"),

240 ("q", "queue", "Show-New::Options::Queue", "HasArg"),

241 ]

242

243 for i in ["help"]:

244 key = "Show-New::Options::%s" % i

245 if key not in cnf: 245 ↛ 243line 245 didn't jump to line 243 because the condition on line 245 was always true

246 cnf[key] = ""

247

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

249 Options = cnf.subtree("Show-New::Options")

250

251 if Options["help"]:

252 usage()

253

254 queue_names = Options.find("Queue", "new").split(",")

255 uploads = (

256 session.query(PolicyQueueUpload)

257 .join(PolicyQueueUpload.policy_queue)

258 .filter(PolicyQueue.queue_name.in_(queue_names))

259 .join(PolicyQueueUpload.changes)

260 .order_by(DBChange.source)

261 )

262

263 if len(changesnames) > 0: 263 ↛ 264line 263 didn't jump to line 264 because the condition on line 263 was never true

264 uploads = uploads.filter(DBChange.changesname.in_(changesnames))

265

266 return uploads

267

268

269def result_callback(r: tuple[int, str]) -> None:

270 code, msg = r

271 if code == PROC_STATUS_EXCEPTION:

272 print("Job raised exception: %s" % (msg))

273 elif code != PROC_STATUS_SUCCESS:

274 print("Job failed: %s" % (msg))

275

276

277################################################################################

278################################################################################

279

280

281def main() -> None:

282 dak.examine_package.use_html = True

283 pool = DakProcessPool(processes=5)

284

285 session = DBConn().session()

286 upload_ids = [u.id for u in init(session)]

287 session.close()

288

289 for upload_id in upload_ids:

290 pool.apply_async(do_pkg, [upload_id], callback=result_callback)

291 pool.close()

292

293 pool.join()

294

295 for htmlfile in htmlfiles_to_process: 295 ↛ 296line 295 didn't jump to line 296 because the loop on line 295 never started

296 with open(htmlfile, "w") as fd:

297 fd.write(timeout_str)

298

299 if pool.overall_status() != PROC_STATUS_SUCCESS: 299 ↛ 300line 299 didn't jump to line 300 because the condition on line 299 was never true

300 raise Exception("Processing failed (code %s)" % (pool.overall_status()))

301

302 files = set(os.listdir(cnf["Show-New::HTMLPath"]))

303 to_delete = [x for x in files.difference(set(sources)) if x.endswith(".html")]

304 for f in to_delete:

305 os.remove(os.path.join(cnf["Show-New::HTMLPath"], f))

306

307

308################################################################################

309

310

311if __name__ == "__main__":

312 main()

Coverage for dak/show_new.py: 77%

110 statements