Coverage for tests / test_types_file.py: 100%

1import subprocess 7 ctx1abcdefg

2import sys 7 ctx1abcdefg

3from io import BytesIO, StringIO 7 ctx1abcdefg

4from pathlib import Path 7 ctx1abcdefg

5 

6import pytest 7 ctx1abcdefg

7import typer 7 ctx1abcdefg

8from typer._click._compat import get_best_encoding, should_strip_ansi 7 ctx1abcdefg

9from typer._click.utils import PacifyFlushWrapper 7 ctx1abcdefg

10from typer.testing import CliRunner 7 ctx1abcdefg

11 

12from tests.utils import needs_linux, needs_macos, needs_windows 7 ctx1abcdefg

13 

14app = typer.Typer() 7 ctx1abcdefg

15 

16 

17@app.command() 7 ctx1abcdefg

18def read_text(file_in: typer.FileText = typer.Option(..., lazy=True)): 7 ctx1abcdefg

19 data = file_in.read() 7 ctx1abcdefg

20 typer.echo(f"text-len={len(data)}") 7 ctx1abcdefg

21 

22 

23@app.command() 7 ctx1abcdefg

24def write_text(file_out: typer.FileTextWrite = typer.Option(..., lazy=None)): 7 ctx1abcdefg

25 file_out.write("This is a single line\n") 7 ctx1abcdefg

26 typer.echo("1 line written") 7 ctx1abcdefg

27 

28 

29@app.command() 7 ctx1abcdefg

30def write_lazy(file_out: typer.FileTextWrite = typer.Option(..., lazy=True)): 7 ctx1abcdefg

31 file_out.write("This is a single lazy line\n") 7 ctx1abcdefg

32 typer.echo("1 line written") 7 ctx1abcdefg

33 

34 

35@app.command() 7 ctx1abcdefg

36def probe_lazy_file_behaviors( 7 ctx1abcdefg

37 file_in: typer.FileText = typer.Option(..., lazy=True), 

38 file_out: typer.FileTextWrite = typer.Option(..., lazy=True), 

39): 

40 typer.echo(f"repr-before={repr(file_out)}") 7 ctx1abcdefg

41 file_out.write("repr-opened\n") 7 ctx1abcdefg

42 typer.echo(f"repr-after={repr(file_out)}") 7 ctx1abcdefg

43 with file_in as stream: 7 ctx1abcdefg

44 typer.echo(f"context-len={len(stream.read())}") 7 ctx1abcdefg

45 stream.seek(0) 7 ctx1abcdefg

46 first_line = next(iter(stream), "") 7 ctx1abcdefg

47 typer.echo(f"first-line={first_line.rstrip()}") 7 ctx1abcdefg

48 

49 

50@app.command() 7 ctx1abcdefg

51def write_binary(file_out: typer.FileBinaryWrite = typer.Option(...)): 7 ctx1abcdefg

52 file_out.write(b"binary-written\n") 7 ctx1abcdefg

53 

54 

55@app.command() 7 ctx1abcdefg

56def write_binary_stderr(): 7 ctx1abcdefg

57 stream = typer.get_binary_stream("stderr") 7 ctx1abcdefg

58 stream.write(b"binary-stderr\n") 7 ctx1abcdefg

59 stream.flush() 7 ctx1abcdefg

60 

61 

62@app.command() 7 ctx1abcdefg

63def read_binary(file_in: typer.FileBinaryRead = typer.Option(...)): 7 ctx1abcdefg

64 data = file_in.read() 7 ctx1abcdefg

65 typer.echo(f"binary-len={len(data)}") 7 ctx1abcdefg

66 

67 

68runner = CliRunner() 7 ctx1abcdefg

69 

70 

71def test_text_stdin_dash() -> None: 7 ctx1abcdefg

72 result = runner.invoke(app, ["read-text", "--file-in=-"], input="hello\n") 7 ctx1abcdefg

73 assert result.exit_code == 0 7 ctx1abcdefg

74 assert "text-len=6" in result.output 7 ctx1abcdefg

75 

76 

77def test_lazy_file(tmp_path: Path) -> None: 7 ctx1abcdefg

78 # dash: written to stdout 

79 result = runner.invoke(app, ["write-text", "--file-out=-"]) 7 ctx1abcdefg

80 assert result.exit_code == 0 7 ctx1abcdefg

81 assert "This is a single line" in result.output 7 ctx1abcdefg

82 assert "1 line written" in result.output 7 ctx1abcdefg

83 

84 # lazy + file 

85 file_path = tmp_path / "example.txt" 7 ctx1abcdefg

86 result = runner.invoke(app, ["write-lazy", f"--file-out={file_path}"]) 7 ctx1abcdefg

87 assert result.exit_code == 0 7 ctx1abcdefg

88 assert "This is a single lazy line" not in result.output 7 ctx1abcdefg

89 assert "1 line written" in result.output 7 ctx1abcdefg

90 assert file_path.exists() 7 ctx1abcdefg

91 assert file_path.read_text() == "This is a single lazy line\n" 7 ctx1abcdefg

92 

93 # lazy probe: unopened/opened repr, context manager, and iteration. 

94 result = runner.invoke( 7 ctx1abcdefg

95 app, 

96 [ 

97 "probe-lazy-file-behaviors", 

98 f"--file-in={file_path}", 

99 f"--file-out={tmp_path / 'repr-opened.txt'}", 

100 ], 

101 ) 

102 assert result.exit_code == 0 7 ctx1abcdefg

103 assert "repr-before=<unopened file" in result.output 7 ctx1abcdefg

104 assert "repr-after=<unopened file" not in result.output 7 ctx1abcdefg

105 assert "repr-opened.txt" in result.output 7 ctx1abcdefg

106 assert "context-len=27" in result.output 7 ctx1abcdefg

107 assert "first-line=This is a single lazy line" in result.output 7 ctx1abcdefg

108 

109 # lazy + dash: written to stdout (lazy setting is pretty much ignored) 

110 result = runner.invoke(app, ["write-lazy", "--file-out=-"]) 7 ctx1abcdefg

111 assert result.exit_code == 0 7 ctx1abcdefg

112 assert "This is a single lazy line" in result.output 7 ctx1abcdefg

113 assert "1 line written" in result.output 7 ctx1abcdefg

114 

115 

116def test_filelike_conversion() -> None: 7 ctx1abcdefg

117 stream = StringIO() 7 ctx1abcdefg

118 result = runner.invoke( 7 ctx1abcdefg

119 app, ["write-text"], default_map={"write-text": {"file_out": stream}} 

120 ) 

121 assert result.exit_code == 0 7 ctx1abcdefg

122 assert "1 line written" in result.output 7 ctx1abcdefg

123 assert stream.getvalue() == "This is a single line\n" 7 ctx1abcdefg

124 

125 

126def test_binary_dash() -> None: 7 ctx1abcdefg

127 result = runner.invoke(app, ["write-binary", "--file-out=-"]) 7 ctx1abcdefg

128 assert result.exit_code == 0 7 ctx1abcdefg

129 assert result.stdout_bytes == b"binary-written\n" 7 ctx1abcdefg

130 

131 result = runner.invoke( 7 ctx1abcdefg

132 app, ["read-binary", "--file-in=-"], input=b"\x00\x01\x02abc" 

133 ) 

134 assert result.exit_code == 0 7 ctx1abcdefg

135 assert "binary-len=6" in result.output 7 ctx1abcdefg

136 

137 

138def test_binary_stderr() -> None: 7 ctx1abcdefg

139 result = subprocess.run( 7 ctx1abcdefg

140 [ 

141 sys.executable, 

142 "-m", 

143 "coverage", 

144 "run", 

145 __file__, 

146 "write-binary-stderr", 

147 ], 

148 capture_output=True, 

149 ) 

150 assert result.returncode == 0 7 ctx1abcdefg

151 assert result.stderr == b"binary-stderr\n" 7 ctx1abcdefg

152 

153 

154@pytest.mark.parametrize( 7 ctx1abcdefg

155 ("errors_arg", "expected_errors"), 

156 [ 

157 (None, "replace"), 

158 ("strict", "strict"), 

159 ], 

160) 

161def test_get_text_stream_errors( 7 ctx1abcdefg

162 monkeypatch, 

163 errors_arg: str | None, 

164 expected_errors: str, 

165) -> None: 

166 class BinaryStdout(BytesIO): 7 ctx1abcdefg

167 pass 7 ctx1abcdefg

168 

169 binary_stdout = BinaryStdout() 7 ctx1abcdefg

170 monkeypatch.setattr(sys, "stdout", binary_stdout) 7 ctx1abcdefg

171 

172 text_stream = typer.get_text_stream("stdout", encoding=None, errors=errors_arg) 7 ctx1abcdefg

173 text_stream.write("stream-text") 7 ctx1abcdefg

174 text_stream.flush() 7 ctx1abcdefg

175 

176 assert text_stream.errors == expected_errors 7 ctx1abcdefg

177 assert text_stream.writable() is True 7 ctx1abcdefg

178 assert binary_stdout.getvalue() == b"stream-text" 7 ctx1abcdefg

179 

180 

181def test_get_best_encoding() -> None: 7 ctx1abcdefg

182 """Test that ASCII is being transformed into UTF-8""" 

183 

184 class AsciiStream: 7 ctx1abcdefg

185 encoding = "ascii" 7 ctx1abcdefg

186 

187 class Utf8Stream: 7 ctx1abcdefg

188 encoding = "utf-8" 7 ctx1abcdefg

189 

190 class UnknownStream: 7 ctx1abcdefg

191 encoding = "unknown" 7 ctx1abcdefg

192 

193 assert get_best_encoding(AsciiStream()) == "utf-8" 7 ctx1abcdefg

194 assert get_best_encoding(Utf8Stream()) == "utf-8" 7 ctx1abcdefg

195 assert get_best_encoding(UnknownStream()) == "unknown" 7 ctx1abcdefg

196 

197 

198def test_pacify_flush_wrapper() -> None: 7 ctx1abcdefg

199 class Wrapped: 7 ctx1abcdefg

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

201 self.name = "wrapped-stream" 7 ctx1abcdefg

202 

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

204 return None # pragma: no cover 

205 

206 wrapped = PacifyFlushWrapper(Wrapped()) 7 ctx1abcdefg

207 assert wrapped.name == "wrapped-stream" 7 ctx1abcdefg

208 

209 

210def test_text_stream_isatty(monkeypatch) -> None: 7 ctx1abcdefg

211 class BinaryStdout(BytesIO): 7 ctx1abcdefg

212 def isatty(self) -> bool: 7 ctx1abcdefg

213 return True 7 ctx1abcdefg

214 

215 binary_stdout = BinaryStdout() 7 ctx1abcdefg

216 monkeypatch.setattr(sys, "stdout", binary_stdout) 7 ctx1abcdefg

217 text_stream = typer.get_text_stream("stdout", encoding="utf-8", errors=None) 7 ctx1abcdefg

218 assert text_stream.isatty() is True 7 ctx1abcdefg

219 

220 

221def test_text_stream_buffer_read1(monkeypatch) -> None: 7 ctx1abcdefg

222 class BinaryStdinNoRead1: 7 ctx1abcdefg

223 def __init__(self, data: bytes) -> None: 7 ctx1abcdefg

224 self._data = data 7 ctx1abcdefg

225 self._pos = 0 7 ctx1abcdefg

226 

227 def read(self, size: int = -1) -> bytes: 7 ctx1abcdefg

228 if size < 0: 7 ctx1abcdefg

229 size = len(self._data) - self._pos # pragma: no cover 

230 chunk = self._data[self._pos : self._pos + size] 7 ctx1abcdefg

231 self._pos += len(chunk) 7 ctx1abcdefg

232 return chunk 7 ctx1abcdefg

233 

234 binary_stdin = BinaryStdinNoRead1(b"hello") 7 ctx1abcdefg

235 monkeypatch.setattr(sys, "stdin", binary_stdin) 7 ctx1abcdefg

236 text_stream = typer.get_text_stream("stdin", encoding="utf-8", errors=None) 7 ctx1abcdefg

237 assert text_stream._stream.read1(4) == b"hell" 7 ctx1abcdefg

238 

239 

240def test_binary_stream(monkeypatch) -> None: 7 ctx1abcdefg

241 binary_stdin = BytesIO(b"hello") 7 ctx1abcdefg

242 binary_stdout = BytesIO() 7 ctx1abcdefg

243 monkeypatch.setattr(sys, "stdin", binary_stdin) 7 ctx1abcdefg

244 monkeypatch.setattr(sys, "stdout", binary_stdout) 7 ctx1abcdefg

245 

246 assert typer.get_binary_stream("stdin") is binary_stdin 7 ctx1abcdefg

247 assert typer.get_binary_stream("stdout") is binary_stdout 7 ctx1abcdefg

248 

249 

250def test_binary_stream_raises(monkeypatch) -> None: 7 ctx1abcdefg

251 class TextOnlyStdin: 7 ctx1abcdefg

252 def read(self, n: int = -1) -> str: 7 ctx1abcdefg

253 return "hello" 7 ctx1abcdefg

254 

255 monkeypatch.setattr(sys, "stdin", TextOnlyStdin()) 7 ctx1abcdefg

256 with pytest.raises(RuntimeError, match="Was not able to determine binary stream"): 7 ctx1abcdefg

257 typer.get_binary_stream("stdin") 7 ctx1abcdefg

258 

259 

260def test_stream_unknown() -> None: 7 ctx1abcdefg

261 with pytest.raises(TypeError, match="Unknown standard stream 'Plumbus'"): 7 ctx1abcdefg

262 typer.get_binary_stream("Plumbus") # type: ignore[arg-type] 7 ctx1abcdefg

263 

264 with pytest.raises(TypeError, match="Unknown standard stream 'Fleeb'"): 7 ctx1abcdefg

265 typer.get_text_stream("Fleeb") # type: ignore[arg-type] 7 ctx1abcdefg

266 

267 

268def test_format_filename() -> None: 7 ctx1abcdefg

269 filename = b"folder/subdir/demo.txt" 7 ctx1abcdefg

270 assert typer.format_filename(filename, shorten=True) == "demo.txt" 7 ctx1abcdefg

271 

272 

273def test_file_error(monkeypatch, tmp_path: Path) -> None: 7 ctx1abcdefg

274 file_path = tmp_path / "cannot-open.txt" 7 ctx1abcdefg

275 

276 def fake_open(path, *args, **kwargs): 7 ctx1abcdefg

277 if Path(path) == file_path: 7 ctx1abcdefg

278 raise OSError() 7 ctx1abcdefg

279 

280 monkeypatch.setattr("builtins.open", fake_open) 7 ctx1abcdefg

281 result = runner.invoke(app, ["write-text", f"--file-out={file_path}"]) 7 ctx1abcdefg

282 assert result.exit_code == 1 7 ctx1abcdefg

283 assert "Could not open file" in result.output 7 ctx1abcdefg

284 assert "cannot-open.txt" in result.output 7 ctx1abcdefg

285 assert "unknown error" in result.output 7 ctx1abcdefg

286 

287 

288@needs_windows 7 ctx1abcdefg

289def test_app_dir_windows_fallback( 7 ctx1abcdefg

290 monkeypatch: pytest.MonkeyPatch, 

291) -> None: 

292 monkeypatch.delenv("APPDATA", raising=False) 2 ctx1cd

293 monkeypatch.setattr("os.path.expanduser", lambda _path: r"C:\Users\Tester") 2 ctx1cd

294 

295 assert typer.get_app_dir("My App", roaming=True) == r"C:\Users\Tester\My App" 2 ctx1cd

296 

297 

298@needs_linux 7 ctx1abcdefg

299def test_app_dir_force_posix(monkeypatch: pytest.MonkeyPatch) -> None: 7 ctx1abcdefg

300 monkeypatch.setattr("os.path.expanduser", lambda _path: "/home/tester/.my-app") 2 ctx1ab

301 

302 assert typer.get_app_dir("My App", force_posix=True) == "/home/tester/.my-app" 2 ctx1ab

303 

304 

305@needs_linux 7 ctx1abcdefg

306def test_app_dir_linux_xdg_config_home(monkeypatch: pytest.MonkeyPatch) -> None: 7 ctx1abcdefg

307 monkeypatch.setenv("XDG_CONFIG_HOME", "/home/tester/config") 2 ctx1ab

308 

309 assert typer.get_app_dir("My App") == "/home/tester/config/my-app" 2 ctx1ab

310 

311 

312@needs_linux 7 ctx1abcdefg

313def test_app_dir_linux_default(monkeypatch: pytest.MonkeyPatch) -> None: 7 ctx1abcdefg

314 monkeypatch.delenv("XDG_CONFIG_HOME", raising=False) 2 ctx1ab

315 monkeypatch.setattr("os.path.expanduser", lambda _path: "/home/tester/.config") 2 ctx1ab

316 

317 assert typer.get_app_dir("My App") == "/home/tester/.config/my-app" 2 ctx1ab

318 

319 

320@needs_macos 7 ctx1abcdefg

321def test_app_dir_macos(monkeypatch: pytest.MonkeyPatch) -> None: 7 ctx1abcdefg

322 monkeypatch.setattr( 3 ctx1efg

323 "os.path.expanduser", lambda _path: "/Users/tester/Library/Application Support" 

324 ) 

325 

326 assert ( 3 ctx1efg

327 typer.get_app_dir("My App") 

328 == "/Users/tester/Library/Application Support/My App" 

329 ) 

330 

331 

332def test_text_stream_binary_buffer(monkeypatch) -> None: 7 ctx1abcdefg

333 class TextStdinWithBinaryBuffer: 7 ctx1abcdefg

334 def __init__(self, data: bytes) -> None: 7 ctx1abcdefg

335 self.buffer = BytesIO(data) 7 ctx1abcdefg

336 self.encoding = "latin-1" 7 ctx1abcdefg

337 

338 def read(self, n: int = -1) -> str: 7 ctx1abcdefg

339 raise OSError("text stream is not readable directly") 7 ctx1abcdefg

340 

341 class TextStdoutWithBinaryBuffer: 7 ctx1abcdefg

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

343 self.buffer = BytesIO() 7 ctx1abcdefg

344 self.encoding = "latin-1" 7 ctx1abcdefg

345 

346 def write(self, s: str) -> int: 7 ctx1abcdefg

347 raise OSError("text stream is not writable directly") 7 ctx1abcdefg

348 

349 stdin = TextStdinWithBinaryBuffer(b"hello") 7 ctx1abcdefg

350 stdout = TextStdoutWithBinaryBuffer() 7 ctx1abcdefg

351 

352 monkeypatch.setattr(sys, "stdin", stdin) 7 ctx1abcdefg

353 monkeypatch.setattr(sys, "stdout", stdout) 7 ctx1abcdefg

354 

355 text_stdin = typer.get_text_stream("stdin", encoding="utf-8", errors=None) 7 ctx1abcdefg

356 text_stdout = typer.get_text_stream("stdout", encoding="utf-8", errors=None) 7 ctx1abcdefg

357 

358 assert text_stdin.read() == "hello" 7 ctx1abcdefg

359 text_stdout.write("ok") 7 ctx1abcdefg

360 text_stdout.flush() 7 ctx1abcdefg

361 assert stdout.buffer.getvalue() == b"ok" 7 ctx1abcdefg

362 

363 

364def test_text_stream_binary_stream(monkeypatch) -> None: 7 ctx1abcdefg

365 binary_stdout = BytesIO() 7 ctx1abcdefg

366 monkeypatch.setattr(sys, "stdout", binary_stdout) 7 ctx1abcdefg

367 text_stream = typer.get_text_stream("stdout", encoding="utf-8", errors=None) 7 ctx1abcdefg

368 text_stream.write("ok") 7 ctx1abcdefg

369 text_stream.flush() 7 ctx1abcdefg

370 assert binary_stdout.getvalue() == b"ok" 7 ctx1abcdefg

371 

372 

373def test_text_stream_stdout_no_binary( 7 ctx1abcdefg

374 monkeypatch, 

375) -> None: 

376 class TextStdoutNoBinaryFallback: 7 ctx1abcdefg

377 encoding = "utf-8" 7 ctx1abcdefg

378 errors = "strict" 7 ctx1abcdefg

379 

380 def write(self, s: str) -> int: 7 ctx1abcdefg

381 if isinstance(s, bytes): 7 ctx1abcdefg

382 raise TypeError("bytes not supported") 7 ctx1abcdefg

383 return len(s) 7 ctx1abcdefg

384 

385 stdout = TextStdoutNoBinaryFallback() 7 ctx1abcdefg

386 monkeypatch.setattr(sys, "stdout", stdout) 7 ctx1abcdefg

387 text_stream = typer.get_text_stream("stdout", encoding="utf-8", errors="replace") 7 ctx1abcdefg

388 assert text_stream is stdout 7 ctx1abcdefg

389 

390 

391def test_jupyter_wrapped_stream(monkeypatch) -> None: 7 ctx1abcdefg

392 class JupyterLikeStdout(BytesIO): 7 ctx1abcdefg

393 __module__ = "ipykernel.iostream" 7 ctx1abcdefg

394 

395 def isatty(self) -> bool: 7 ctx1abcdefg

396 return False 7 ctx1abcdefg

397 

398 binary_stdout = JupyterLikeStdout() 7 ctx1abcdefg

399 monkeypatch.setattr(sys, "stdout", binary_stdout) 7 ctx1abcdefg

400 text_stream = typer.get_text_stream("stdout", encoding="utf-8", errors=None) 7 ctx1abcdefg

401 assert should_strip_ansi(text_stream, color=None) is False 7 ctx1abcdefg

402 

403 

404def test_should_strip_ansi(monkeypatch) -> None: 7 ctx1abcdefg

405 class NonTtyStdin(BytesIO): 7 ctx1abcdefg

406 def isatty(self) -> bool: 7 ctx1abcdefg

407 return False 7 ctx1abcdefg

408 

409 stdin = NonTtyStdin() 7 ctx1abcdefg

410 monkeypatch.setattr(sys, "stdin", stdin) 7 ctx1abcdefg

411 assert should_strip_ansi(stream=None, color=None) is True 7 ctx1abcdefg

412 assert should_strip_ansi(stream=None, color=True) is False 7 ctx1abcdefg

413 assert should_strip_ansi(stream=None, color=False) is True 7 ctx1abcdefg

414 

415 

416if __name__ == "__main__": 7 ctx1abcdefg

417 app() 7 ctx1abcdefg