Coverage for typer / testing.py: 100%

1import contextlib 7 ctx1abcdefg

2import io 7 ctx1abcdefg

3import os 7 ctx1abcdefg

4import shlex 7 ctx1abcdefg

5import sys 7 ctx1abcdefg

6from collections.abc import Generator, Mapping, Sequence 7 ctx1abcdefg

7from types import TracebackType 7 ctx1abcdefg

8from typing import IO, TYPE_CHECKING, Any, BinaryIO, cast 7 ctx1abcdefg

9 

10from typer.main import Typer 7 ctx1abcdefg

11from typer.main import get_command as _get_command 7 ctx1abcdefg

12 

13from . import _click 7 ctx1abcdefg

14from ._click import _compat, formatting, termui, utils 7 ctx1abcdefg

15 

16if TYPE_CHECKING: 7 ctx1abcdefg

17 from _typeshed import ReadableBuffer 

18 

19 

20def make_input_stream(input: str | bytes | None, charset: str) -> BinaryIO: 7 ctx1abcdefg

21 if input is None: 7 ctx1abcdefg

22 input = b"" 7 ctx1abcdefg

23 elif isinstance(input, str): 7 ctx1abcdefg

24 input = input.encode(charset) 7 ctx1abcdefg

25 

26 return io.BytesIO(input) 7 ctx1abcdefg

27 

28 

29class BytesIOCopy(io.BytesIO): 7 ctx1abcdefg

30 """Patch ``io.BytesIO`` to let the written stream be copied to another.""" 

31 

32 def __init__(self, copy_to: io.BytesIO) -> None: 7 ctx1abcdefg

33 super().__init__() 7 ctx1abcdefg

34 self.copy_to = copy_to 7 ctx1abcdefg

35 

36 def flush(self) -> None: 7 ctx1abcdefg

37 super().flush() 7 ctx1abcdefg

38 self.copy_to.flush() 7 ctx1abcdefg

39 

40 def write(self, b: "ReadableBuffer") -> int: 7 ctx1abcdefg

41 self.copy_to.write(b) 7 ctx1abcdefg

42 return super().write(b) 7 ctx1abcdefg

43 

44 

45class StreamMixer: 7 ctx1abcdefg

46 """Mixes `<stdout>` and `<stderr>` streams. 

47 

48 The result is available in the ``output`` attribute. 

49 """ 

50 

51 def __init__(self) -> None: 7 ctx1abcdefg

52 self.output: io.BytesIO = io.BytesIO() 7 ctx1abcdefg

53 self.stdout: io.BytesIO = BytesIOCopy(copy_to=self.output) 7 ctx1abcdefg

54 self.stderr: io.BytesIO = BytesIOCopy(copy_to=self.output) 7 ctx1abcdefg

55 

56 def __del__(self) -> None: 7 ctx1abcdefg

57 """Guarantee that file-like objects are closed in a predictable order""" 

58 self.stderr.close() 7 ctx1abcdefg

59 self.stdout.close() 7 ctx1abcdefg

60 self.output.close() 7 ctx1abcdefg

61 

62 

63class _NamedTextIOWrapper(io.TextIOWrapper): 7 ctx1abcdefg

64 def __init__(self, buffer: BinaryIO, name: str, mode: str, **kwargs: Any) -> None: 7 ctx1abcdefg

65 super().__init__(buffer, **kwargs) 7 ctx1abcdefg

66 self._name = name 7 ctx1abcdefg

67 self._mode = mode 7 ctx1abcdefg

68 

69 @property 7 ctx1abcdefg

70 def name(self) -> str: 7 ctx1abcdefg

71 return self._name # pragma: no cover 

72 

73 @property 7 ctx1abcdefg

74 def mode(self) -> str: 7 ctx1abcdefg

75 return self._mode # pragma: no cover 

76 

77 

78class Result: 7 ctx1abcdefg

79 """Holds the captured result of an invoked CLI script.""" 

80 

81 def __init__( 7 ctx1abcdefg

82 self, 

83 runner: "CliRunner", 

84 stdout_bytes: bytes, 

85 stderr_bytes: bytes, 

86 output_bytes: bytes, 

87 return_value: Any, 

88 exit_code: int, 

89 exception: BaseException | None, 

90 exc_info: tuple[type[BaseException], BaseException, TracebackType] 

91 | None = None, 

92 ): 

93 self.runner = runner 7 ctx1abcdefg

94 self.stdout_bytes = stdout_bytes 7 ctx1abcdefg

95 self.stderr_bytes = stderr_bytes 7 ctx1abcdefg

96 self.output_bytes = output_bytes 7 ctx1abcdefg

97 self.return_value = return_value 7 ctx1abcdefg

98 self.exit_code = exit_code 7 ctx1abcdefg

99 self.exception = exception 7 ctx1abcdefg

100 self.exc_info = exc_info 7 ctx1abcdefg

101 

102 @property 7 ctx1abcdefg

103 def output(self) -> str: 7 ctx1abcdefg

104 """The terminal output as unicode string, as the user would see it.""" 

105 return self.output_bytes.decode(self.runner.charset, "replace").replace( 7 ctx1abcdefg

106 "\r\n", "\n" 

107 ) 

108 

109 @property 7 ctx1abcdefg

110 def stdout(self) -> str: 7 ctx1abcdefg

111 """The standard output as unicode string.""" 

112 return self.stdout_bytes.decode(self.runner.charset, "replace").replace( 7 ctx1abcdefg

113 "\r\n", "\n" 

114 ) 

115 

116 @property 7 ctx1abcdefg

117 def stderr(self) -> str: 7 ctx1abcdefg

118 """The standard error as unicode string.""" 

119 return self.stderr_bytes.decode(self.runner.charset, "replace").replace( 7 ctx1abcdefg

120 "\r\n", "\n" 

121 ) 

122 

123 def __repr__(self) -> str: 7 ctx1abcdefg

124 exc_str = repr(self.exception) if self.exception else "okay" 7 ctx1abcdefg

125 return f"<{type(self).__name__} {exc_str}>" 7 ctx1abcdefg

126 

127 

128class CliRunner: 7 ctx1abcdefg

129 """The CLI runner provides functionality to invoke a command line 

130 script for unittesting purposes in an isolated environment. This only 

131 works in single-threaded systems without any concurrency as it changes the 

132 global interpreter state. Based on functionality from Click. 

133 """ 

134 

135 def __init__( 7 ctx1abcdefg

136 self, 

137 charset: str = "utf-8", 

138 env: Mapping[str, str | None] | None = None, 

139 ) -> None: 

140 self.charset = charset 7 ctx1abcdefg

141 self.env: Mapping[str, str | None] = env or {} 7 ctx1abcdefg

142 

143 def get_default_prog_name(self, cli: _click.Command) -> str: 7 ctx1abcdefg

144 """Return the default program name for a command. 

145 The default is the `name` attribute or ``"root"`` if not set. 

146 """ 

147 return cli.name or "root" 7 ctx1abcdefg

148 

149 def make_env( 7 ctx1abcdefg

150 self, overrides: Mapping[str, str | None] | None = None 

151 ) -> Mapping[str, str | None]: 

152 """Returns the environment overrides for invoking a script.""" 

153 rv = dict(self.env) 7 ctx1abcdefg

154 if overrides: 7 ctx1abcdefg

155 rv.update(overrides) 7 ctx1abcdefg

156 return rv 7 ctx1abcdefg

157 

158 @contextlib.contextmanager 7 ctx1abcdefg

159 def isolation( 7 ctx1abcdefg

160 self, 

161 input: str | bytes | None = None, 

162 env: Mapping[str, str | None] | None = None, 

163 color: bool = False, 

164 ) -> Generator[tuple[io.BytesIO, io.BytesIO, io.BytesIO]]: 

165 """A context manager that sets up the isolation for invoking of a 

166 command line tool. This sets up `<stdin>` with the given input data 

167 and `os.environ` with the overrides from the given dictionary. 

168 """ 

169 bytes_input = make_input_stream(input, self.charset) 7 ctx1abcdefg

170 

171 old_stdin = sys.stdin 7 ctx1abcdefg

172 old_stdout = sys.stdout 7 ctx1abcdefg

173 old_stderr = sys.stderr 7 ctx1abcdefg

174 old_forced_width = formatting.FORCED_WIDTH 7 ctx1abcdefg

175 formatting.FORCED_WIDTH = 80 7 ctx1abcdefg

176 

177 env = self.make_env(env) 7 ctx1abcdefg

178 

179 stream_mixer = StreamMixer() 7 ctx1abcdefg

180 

181 sys.stdin = text_input = _NamedTextIOWrapper( 7 ctx1abcdefg

182 bytes_input, encoding=self.charset, name="<stdin>", mode="r" 

183 ) 

184 

185 sys.stdout = _NamedTextIOWrapper( 7 ctx1abcdefg

186 stream_mixer.stdout, encoding=self.charset, name="<stdout>", mode="w" 

187 ) 

188 

189 sys.stderr = _NamedTextIOWrapper( 7 ctx1abcdefg

190 stream_mixer.stderr, 

191 encoding=self.charset, 

192 name="<stderr>", 

193 mode="w", 

194 errors="backslashreplace", 

195 ) 

196 

197 def visible_input(prompt: str | None = None) -> str: 7 ctx1abcdefg

198 sys.stdout.write(prompt or "") 7 ctx1abcdefg

199 try: 7 ctx1abcdefg

200 val = next(text_input).rstrip("\r\n") 7 ctx1abcdefg

201 except StopIteration as e: # pragma: no cover 

202 raise EOFError() from e 

203 sys.stdout.write(f"{val}\n") 7 ctx1abcdefg

204 sys.stdout.flush() 7 ctx1abcdefg

205 return val 7 ctx1abcdefg

206 

207 def hidden_input(prompt: str | None = None) -> str: 7 ctx1abcdefg

208 sys.stdout.write(f"{prompt or ''}\n") 7 ctx1abcdefg

209 sys.stdout.flush() 7 ctx1abcdefg

210 try: 7 ctx1abcdefg

211 return next(text_input).rstrip("\r\n") 7 ctx1abcdefg

212 except StopIteration as e: # pragma: no cover 

213 raise EOFError() from e 

214 

215 def _getchar(echo: bool) -> str: 7 ctx1abcdefg

216 char = sys.stdin.read(1) 7 ctx1abcdefg

217 

218 if echo: 7 ctx1abcdefg

219 sys.stdout.write(char) 7 ctx1abcdefg

220 

221 sys.stdout.flush() 7 ctx1abcdefg

222 return char 7 ctx1abcdefg

223 

224 default_color = color 7 ctx1abcdefg

225 

226 def should_strip_ansi( 7 ctx1abcdefg

227 stream: IO[Any] | None = None, color: bool | None = None 

228 ) -> bool: 

229 if color is None: 7 ctx1abcdefg

230 return not default_color 7 ctx1abcdefg

231 return not color 7 ctx1abcdefg

232 

233 old_visible_prompt_func = termui.visible_prompt_func 7 ctx1abcdefg

234 old_hidden_prompt_func = termui.hidden_prompt_func 7 ctx1abcdefg

235 old__getchar_func = termui._getchar 7 ctx1abcdefg

236 old_should_strip_ansi = utils.should_strip_ansi # type: ignore[attr-defined] 7 ctx1abcdefg

237 old__compat_should_strip_ansi = _compat.should_strip_ansi 7 ctx1abcdefg

238 termui.visible_prompt_func = visible_input 7 ctx1abcdefg

239 termui.hidden_prompt_func = hidden_input # ty: ignore[invalid-assignment] 7 ctx1abcdefg

240 termui._getchar = _getchar 7 ctx1abcdefg

241 utils.should_strip_ansi = should_strip_ansi # type: ignore 7 ctx1abcdefg

242 _compat.should_strip_ansi = should_strip_ansi # ty: ignore[invalid-assignment] 7 ctx1abcdefg

243 

244 old_env = {} 7 ctx1abcdefg

245 try: 7 ctx1abcdefg

246 for key, value in env.items(): 7 ctx1abcdefg

247 old_env[key] = os.environ.get(key) 7 ctx1abcdefg

248 if value is None: 7 ctx1abcdefg

249 try: 7 ctx1abcdefg

250 del os.environ[key] 7 ctx1abcdefg

251 except Exception: # pragma: no cover 

252 pass 

253 else: 

254 os.environ[key] = value 7 ctx1abcdefg

255 yield (stream_mixer.stdout, stream_mixer.stderr, stream_mixer.output) 7 ctx1abcdefg

256 finally: 

257 for key, value in old_env.items(): 7 ctx1abcdefg

258 if value is None: 7 ctx1abcdefg

259 try: 7 ctx1abcdefg

260 del os.environ[key] 7 ctx1abcdefg

261 except Exception: # pragma: no cover 

262 pass 

263 else: 

264 os.environ[key] = value 7 ctx1abcdefg

265 sys.stdout = old_stdout 7 ctx1abcdefg

266 sys.stderr = old_stderr 7 ctx1abcdefg

267 sys.stdin = old_stdin 7 ctx1abcdefg

268 termui.visible_prompt_func = old_visible_prompt_func 7 ctx1abcdefg

269 termui.hidden_prompt_func = old_hidden_prompt_func 7 ctx1abcdefg

270 termui._getchar = old__getchar_func 7 ctx1abcdefg

271 utils.should_strip_ansi = old_should_strip_ansi # type: ignore[attr-defined] 7 ctx1abcdefg

272 _compat.should_strip_ansi = old__compat_should_strip_ansi 7 ctx1abcdefg

273 formatting.FORCED_WIDTH = old_forced_width 7 ctx1abcdefg

274 

275 def invoke( 7 ctx1abcdefg

276 self, 

277 app: Typer, 

278 args: str | Sequence[str] | None = None, 

279 input: bytes | str | None = None, 

280 env: Mapping[str, str | None] | None = None, 

281 catch_exceptions: bool = True, 

282 color: bool = False, 

283 **extra: Any, 

284 ) -> Result: 

285 cli = _get_command(app) 7 ctx1abcdefg

286 exc_info = None 7 ctx1abcdefg

287 

288 with self.isolation(input=input, env=env, color=color) as outstreams: 7 ctx1abcdefg

289 return_value = None 7 ctx1abcdefg

290 exception: BaseException | None = None 7 ctx1abcdefg

291 exit_code = 0 7 ctx1abcdefg

292 

293 if isinstance(args, str): 7 ctx1abcdefg

294 args = shlex.split(args) 7 ctx1abcdefg

295 

296 try: 7 ctx1abcdefg

297 prog_name = extra.pop("prog_name") 7 ctx1abcdefg

298 except KeyError: 7 ctx1abcdefg

299 prog_name = self.get_default_prog_name(cli) 7 ctx1abcdefg

300 

301 try: 7 ctx1abcdefg

302 return_value = cli.main(args=args or (), prog_name=prog_name, **extra) 7 ctx1abcdefg

303 except SystemExit as e: 7 ctx1abcdefg

304 exc_info = sys.exc_info() 7 ctx1abcdefg

305 e_code = cast("int | Any | None", e.code) 7 ctx1abcdefg

306 

307 if e_code is None: 7 ctx1abcdefg

308 e_code = 0 7 ctx1abcdefg

309 

310 if e_code != 0: 7 ctx1abcdefg

311 exception = e 7 ctx1abcdefg

312 

313 if not isinstance(e_code, int): 7 ctx1abcdefg

314 sys.stdout.write(str(e_code)) 7 ctx1abcdefg

315 sys.stdout.write("\n") 7 ctx1abcdefg

316 e_code = 1 7 ctx1abcdefg

317 

318 exit_code = e_code 7 ctx1abcdefg

319 

320 except Exception as e: 7 ctx1abcdefg

321 if not catch_exceptions: 7 ctx1abcdefg

322 raise 7 ctx1abcdefg

323 exception = e 7 ctx1abcdefg

324 exit_code = 1 7 ctx1abcdefg

325 exc_info = sys.exc_info() 7 ctx1abcdefg

326 finally: 

327 sys.stdout.flush() 7 ctx1abcdefg

328 sys.stderr.flush() 7 ctx1abcdefg

329 stdout = outstreams[0].getvalue() 7 ctx1abcdefg

330 stderr = outstreams[1].getvalue() 7 ctx1abcdefg

331 output = outstreams[2].getvalue() 7 ctx1abcdefg

332 

333 return Result( 7 ctx1abcdefg

334 runner=self, 

335 stdout_bytes=stdout, 

336 stderr_bytes=stderr, 

337 output_bytes=output, 

338 return_value=return_value, 

339 exit_code=exit_code, 

340 exception=exception, 

341 exc_info=exc_info, # type: ignore 

342 )