Coverage for src/debputy/commands/debputy

1import argparse

2import contextlib

3import os

4import re

5import shutil

6import subprocess

7import sys

8import types

9from typing import (

10 IO,

11 Any,

12)

13from collections.abc import Sequence, Iterator, Mapping

15from debputy.util import assume_not_none

17colored: types.ModuleType | None

18try:

19 import colored

21 if ( 21 ↛ 27line 21 didn't jump to line 27 because the condition on line 21 was never true

22 not hasattr(colored, "Style")

23 or not hasattr(colored, "Fore")

24 or not hasattr(colored, "Back")

25 ):

26 # Seen with python3-colored v1 (bookworm)

27 raise ImportError

28except ImportError:

29 colored = None

32def _pager() -> str | None:

33 pager = os.environ.get("DEBPUTY_PAGER")

34 if pager is None:

35 pager = os.environ.get("PAGER")

36 if pager is None and shutil.which("less") is not None:

37 pager = "less"

38 return pager

41URL_START = "\033]8;;"

42URL_END = "\033]8;;\a"

43MAN_URL_REWRITE = re.compile(r"man:(\S+)[(](\d+)[)]")

45_SUPPORTED_COLORS = {

46 "black",

47 "red",

48 "green",

49 "yellow",

50 "blue",

51 "magenta",

52 "cyan",

53 "white",

54}

55_SUPPORTED_STYLES = {"none", "bold"}

58class OutputStyle:

60 def colored(

61 self,

62 text: str,

63 *,

64 fg: str | None = None,

65 bg: str | None = None,

66 style: str | None = None,

67 ) -> str:

68 self._check_color(fg)

69 self._check_color(bg)

70 self._check_text_style(style)

71 return text

73 @property

74 def supports_colors(self) -> bool:

75 return False

77 def _check_color(self, color: str | None) -> None:

78 if color is not None and color not in _SUPPORTED_COLORS:

79 raise ValueError(

80 f"Unsupported color: {color}. Only the following are supported {','.join(_SUPPORTED_COLORS)}"

81 )

83 def _check_text_style(self, style: str | None) -> None:

84 if style is not None and style not in _SUPPORTED_STYLES:

85 raise ValueError(

86 f"Unsupported style: {style}. Only the following are supported {','.join(_SUPPORTED_STYLES)}"

87 )

89 def heading(self, heading: str, level: int) -> str:

90 return heading

92 def render_url(self, link_url: str) -> str:

93 return link_url

95 def bts(self, bugno) -> str:

96 return self.render_url(f"https://bugs.debian.org/{bugno}")

99class MarkdownOutputStyle(OutputStyle):

100 def colored(

101 self,

102 text: str,

103 *,

104 fg: str | None = None,

105 bg: str | None = None,

106 style: str | None = None,

107 ) -> str:

108 result = super().colored(text, fg=fg, bg=bg, style=style)

109 if style == "bold":

110 return f"**{result}**"

111 return result

112

113 def heading(self, heading: str, level: int) -> str:

114 prefix = "#" * level

115 return f"{prefix} {heading}"

116

117 def render_url(self, link_url: str) -> str:

118 if link_url.startswith("man:"):

119 # Rewrite man page to a clickable link for markdown

120 m = MAN_URL_REWRITE.match(link_url)

121 if m:

122 page, section = m.groups()

123 man_page_url = f"https://manpages.debian.org/{page}.{section}"

124 return f"[{link_url}]({man_page_url})"

125 return f"<{link_url}>"

126

127

128class IOBasedOutputStyling(OutputStyle):

129 def __init__(

130 self,

131 stream: IO[str],

132 output_format: str,

133 *,

134 optimize_for_screen_reader: bool = False,

135 ) -> None:

136 self.stream = stream

137 self.output_format = output_format

138 self.optimize_for_screen_reader = optimize_for_screen_reader

139 self._color_support: types.ModuleType | None = None

140

141 def print_list_table(

142 self,

143 headers: Sequence[str | tuple[str, str]],

144 rows: Sequence[Sequence[str]],

145 ) -> None:

146 if rows:

147 if any(len(r) != len(rows[0]) for r in rows):

148 raise ValueError(

149 "Unbalanced table: All rows must have the same column count"

150 )

151 if len(rows[0]) != len(headers):

152 raise ValueError(

153 "Unbalanced table: header list does not agree with row list on number of columns"

154 )

155

156 if not headers:

157 raise ValueError("No headers provided!?")

158

159 cadjust = {}

160 header_names = []

161 for c in headers:

162 if isinstance(c, str):

163 header_names.append(c)

164 else:

165 cname, adjust = c

166 header_names.append(cname)

167 cadjust[cname] = adjust

168

169 if self.output_format == "csv":

170 from csv import writer

171

172 w = writer(self.stream)

173 w.writerow(header_names)

174 w.writerows(rows)

175 return

176

177 column_lengths = [

178 max((len(h), max(len(r[i]) for r in rows)))

179 for i, h in enumerate(header_names)

180 ]

181 # divider => "+---+---+-...-+"

182 divider = "+-" + "-+-".join("-" * x for x in column_lengths) + "-+"

183 # row_format => '| {:<10} | {:<8} | ... |' where the numbers are the column lengths

184 row_format_inner = " | ".join(

185 f"{ CELL_COLOR} { :{cadjust.get(cn, '<')}{x}} { CELL_COLOR_RESET} "

186 for cn, x in zip(header_names, column_lengths)

187 )

188

189 row_format = f"| {row_format_inner} |"

190

191 if self.supports_colors:

192 cs = self._color_support

193 assert cs is not None

194 header_color = cs.Style.bold

195 header_color_reset = cs.Style.reset

196 else:

197 header_color = ""

198 header_color_reset = ""

199

200 self.print_visual_formatting(divider)

201 self.print(

202 row_format.format(

203 *header_names,

204 CELL_COLOR=header_color,

205 CELL_COLOR_RESET=header_color_reset,

206 )

207 )

208 self.print_visual_formatting(divider)

209 for row in rows:

210 self.print(row_format.format(*row, CELL_COLOR="", CELL_COLOR_RESET=""))

211 self.print_visual_formatting(divider)

212

213 def print(self, /, string: str = "", **kwargs) -> None:

214 if "file" in kwargs:

215 raise ValueError("Unsupported kwarg file")

216 print(string, file=self.stream, **kwargs)

217

218 def print_visual_formatting(self, /, format_sequence: str, **kwargs) -> None:

219 if self.optimize_for_screen_reader:

220 return

221 self.print(format_sequence, **kwargs)

222

223 def print_for_screen_reader(self, /, text: str, **kwargs) -> None:

224 if not self.optimize_for_screen_reader:

225 return

226 self.print(text, **kwargs)

227

228 def heading(self, heading: str, level: int) -> str:

229 # Use markdown notation

230 heading_prefix = "#" * level

231 return f"{heading_prefix} {heading}"

232

233

234class ANSIOutputStylingBase(IOBasedOutputStyling):

235 def __init__(

236 self,

237 stream: IO[str],

238 output_format: str,

239 *,

240 support_colors: bool = True,

241 support_clickable_urls: bool = True,

242 **kwargs: Any,

243 ) -> None:

244 super().__init__(stream, output_format, **kwargs)

245 self._stream = stream

246 self._color_support = colored

247 self._support_colors = (

248 support_colors if self._color_support is not None else False

249 )

250 self._support_clickable_urls = support_clickable_urls

251

252 @property

253 def supports_colors(self) -> bool:

254 return self._support_colors

255

256 def colored(

257 self,

258 text: str,

259 *,

260 fg: str | None = None,

261 bg: str | None = None,

262 style: str | None = None,

263 ) -> str:

264 self._check_color(fg)

265 self._check_color(bg)

266 self._check_text_style(style)

267 _colored = self._color_support

268 if not self.supports_colors or _colored is None:

269 return text

270 codes = []

271 if style is not None:

272 code = getattr(_colored.Style, style)

273 assert code is not None

274 codes.append(code)

275 if fg is not None:

276 code = getattr(_colored.Fore, fg)

277 assert code is not None

278 codes.append(code)

279 if bg is not None:

280 code = getattr(_colored.Back, bg)

281 assert code is not None

282 codes.append(code)

283 if not codes:

284 return text

285 return "".join(codes) + text + _colored.Style.reset

286

287 def render_url(self, link_url: str) -> str:

288 if not self._support_clickable_urls:

289 return super().render_url(link_url)

290 link_text = link_url

291 if not self.optimize_for_screen_reader and link_url.startswith("man:"):

292 # Rewrite man page to a clickable link by default. I am not sure how the hyperlink

293 # ANSI code works with screen readers, so lets not rewrite the man page link by

294 # default. My fear is that both the link url and the link text gets read out.

295 m = MAN_URL_REWRITE.match(link_url)

296 if m:

297 page, section = m.groups()

298 link_url = f"https://manpages.debian.org/{page}.{section}"

299 return URL_START + f"{link_url}\a{link_text}" + URL_END

300

301 def heading(self, heading: str, level: int) -> str:

302 return self.colored(super().heading(heading, level), style="bold")

303

304 def bts(self, bugno) -> str:

305 if not self._support_clickable_urls:

306 return super().bts(bugno)

307 return self.render_url(f"https://bugs.debian.org/{bugno}")

308

309

310def no_fancy_output(

311 stream: IO[str] | None = None,

312 output_format: str = "",

313 optimize_for_screen_reader: bool = False,

314) -> IOBasedOutputStyling:

315 if stream is None: 315 ↛ 317line 315 didn't jump to line 317 because the condition on line 315 was always true

316 stream = sys.stdout

317 return IOBasedOutputStyling(

318 stream,

319 output_format,

320 optimize_for_screen_reader=optimize_for_screen_reader,

321 )

322

323

324def _output_styling(

325 parsed_args: argparse.Namespace,

326 stream: IO[str],

327) -> IOBasedOutputStyling:

328 output_format = getattr(parsed_args, "output_format", None)

329 if output_format is None:

330 output_format = "text"

331 optimize_for_screen_reader = os.environ.get("OPTIMIZE_FOR_SCREEN_READER", "") != ""

332 if not stream.isatty():

333 return no_fancy_output(

334 stream,

335 output_format,

336 optimize_for_screen_reader=optimize_for_screen_reader,

337 )

338

339 return ANSIOutputStylingBase(

340 stream, output_format, optimize_for_screen_reader=optimize_for_screen_reader

341 )

342

343

344@contextlib.contextmanager

345def _stream_to_pager(

346 parsed_args: argparse.Namespace,

347) -> Iterator[tuple[IO[str], IOBasedOutputStyling]]:

348 fancy_output = _output_styling(parsed_args, sys.stdout)

349 if (

350 not parsed_args.pager

351 or not sys.stdout.isatty()

352 or fancy_output.output_format != "text"

353 ):

354 yield sys.stdout, fancy_output

355 return

356

357 pager = _pager()

358 if pager is None:

359 yield sys.stdout, fancy_output

360 return

361

362 env: Mapping[str, str] = os.environ

363 if "LESS" not in env:

364 env_copy = dict(os.environ)

365 env_copy["LESS"] = "-FRSXMQ"

366 env = env_copy

367

368 cmd = subprocess.Popen(

369 pager,

370 stdin=subprocess.PIPE,

371 encoding="utf-8",

372 env=env,

373 )

374 stdin = assume_not_none(cmd.stdin)

375 try:

376 fancy_output.stream = stdin

377 yield stdin, fancy_output

378 except Exception:

379 stdin.close()

380 cmd.kill()

381 cmd.wait()

382 raise

383 finally:

384 fancy_output.stream = sys.stdin

385 stdin.close()

386 cmd.wait()

Coverage for src/debputy/commands/debputy_cmd/output.py: 23%

217 statements