Coverage for tests / test_win_console.py: 100%

1""" 

2Tests for the Windows console functionality. 

3Created after vendoring Click to ensure test coverage is back up to 100%. 

4""" 

5 

6import ctypes 7 ctx1cdabefg

7import io 7 ctx1cdabefg

8import sys 7 ctx1cdabefg

9 

10import pytest 7 ctx1cdabefg

11import typer 7 ctx1cdabefg

12from typer.testing import CliRunner 7 ctx1cdabefg

13 

14from .utils import needs_windows 7 ctx1cdabefg

15 

16pytestmark = needs_windows 7 ctx1cdabefg

17 

18 

19if sys.platform == "win32": 7 ctx1cdabefg

20 from typer._click import _compat, _winconsole 2 ctx1ab

21 

22 

23def _identity_buffer(obj, writable=False): # noqa: ARG001 7 ctx1cdabefg

24 return obj 2 ctx1ab

25 

26 

27def _route_console_stream(target_name, wrapper, state=None): 7 ctx1cdabefg

28 def patched_windows_stream(stream, encoding, errors): # noqa: ARG001 2 ctx1ab

29 current_target = getattr(sys, target_name) 2 ctx1ab

30 if stream is current_target: 2 ctx1ab

31 if state is not None and target_name == "stderr": 2 ctx1ab

32 state["stderr_wrap_calls"] += 1 2 ctx1ab

33 buffer = getattr(stream, "buffer", None) 2 ctx1ab

34 return wrapper(buffer) if buffer else None 2 ctx1ab

35 return None 2 ctx1ab

36 

37 return patched_windows_stream 2 ctx1ab

38 

39 

40def _capture_write_console(state): 7 ctx1cdabefg

41 def fake_write_console(handle, buffer, units_to_write, units_written_ptr, reserved): # noqa: ARG001 2 ctx1ab

42 state["write_calls"] += 1 2 ctx1ab

43 bytes_to_write = units_to_write * 2 2 ctx1ab

44 state["written"].extend(buffer[:bytes_to_write]) 2 ctx1ab

45 units_written_ptr._obj.value = units_to_write 2 ctx1ab

46 return 1 2 ctx1ab

47 

48 return fake_write_console 2 ctx1ab

49 

50 

51def test_winconsole_stdin(monkeypatch): 7 ctx1cdabefg

52 runner = CliRunner() 2 ctx1ab

53 app = typer.Typer() 2 ctx1ab

54 

55 @app.command() 2 ctx1ab

56 def read_name(config: typer.FileText = typer.Option(...)) -> None: 2 ctx1ab

57 name = config.readline().strip() 2 ctx1ab

58 typer.echo(f"Hello {name}") 2 ctx1ab

59 

60 utf16_data = bytearray("Rick\r\n".encode("utf-16-le")) 2 ctx1ab

61 state = {"pos": 0, "read_calls": 0} 2 ctx1ab

62 

63 def fake_read_console(handle, buffer, units_to_read, units_read_ptr, reserved): # noqa: ARG001 2 ctx1ab

64 state["read_calls"] += 1 2 ctx1ab

65 max_bytes = units_to_read * 2 2 ctx1ab

66 chunk = utf16_data[state["pos"] : state["pos"] + max_bytes] 2 ctx1ab

67 if chunk: 2 ctx1ab

68 buffer[0 : len(chunk)] = chunk 2 ctx1ab

69 state["pos"] += len(chunk) 2 ctx1ab

70 units_read_ptr._obj.value = len(chunk) // 2 2 ctx1ab

71 return 1 2 ctx1ab

72 

73 return 1 # pragma: no cover 

74 

75 monkeypatch.setattr(_winconsole, "get_buffer", _identity_buffer) 2 ctx1ab

76 monkeypatch.setattr(_winconsole, "ReadConsoleW", fake_read_console) 2 ctx1ab

77 monkeypatch.setattr(_winconsole, "GetLastError", lambda: 0) 2 ctx1ab

78 monkeypatch.setattr( 2 ctx1ab

79 _compat, 

80 "_get_windows_console_stream", 

81 _route_console_stream("stdin", _winconsole._get_text_stdin), 

82 ) 

83 

84 result = runner.invoke(app, ["--config", "-"]) 2 ctx1ab

85 assert result.exit_code == 0, result.output 2 ctx1ab

86 assert "Hello Rick" in result.stdout 2 ctx1ab

87 assert state["read_calls"] > 0 2 ctx1ab

88 

89 

90def test_winconsole_stdout(monkeypatch): 7 ctx1cdabefg

91 runner = CliRunner() 2 ctx1ab

92 app = typer.Typer() 2 ctx1ab

93 state = {"write_calls": 0, "written": bytearray()} 2 ctx1ab

94 

95 @app.command() 2 ctx1ab

96 def write_message(out: typer.FileTextWrite = typer.Option(...)) -> None: 2 ctx1ab

97 out.write("Hello Summer\n") 2 ctx1ab

98 

99 monkeypatch.setattr(_winconsole, "get_buffer", _identity_buffer) 2 ctx1ab

100 monkeypatch.setattr(_winconsole, "WriteConsoleW", _capture_write_console(state)) 2 ctx1ab

101 monkeypatch.setattr(_winconsole, "GetLastError", lambda: 0) 2 ctx1ab

102 monkeypatch.setattr( 2 ctx1ab

103 _compat, 

104 "_get_windows_console_stream", 

105 _route_console_stream("stdout", _winconsole._get_text_stdout), 

106 ) 

107 

108 result = runner.invoke(app, ["--out", "-"]) 2 ctx1ab

109 assert result.exit_code == 0, result.output 2 ctx1ab

110 assert state["write_calls"] > 0 2 ctx1ab

111 assert _winconsole._WindowsConsoleWriter(1).isatty() is True 2 ctx1ab

112 decoded = state["written"].decode("utf-16-le", errors="ignore") 2 ctx1ab

113 assert "Hello Summer\r\n" in decoded 2 ctx1ab

114 

115 

116def test_winconsole_stderr(monkeypatch): 7 ctx1cdabefg

117 runner = CliRunner() 2 ctx1ab

118 app = typer.Typer() 2 ctx1ab

119 state = {"write_calls": 0, "written": bytearray(), "stderr_wrap_calls": 0} 2 ctx1ab

120 

121 @app.command() 2 ctx1ab

122 def main() -> None: 2 ctx1ab

123 typer.echo("Ran out of adventure time!", err=True) 2 ctx1ab

124 

125 monkeypatch.setattr(_winconsole, "get_buffer", _identity_buffer) 2 ctx1ab

126 monkeypatch.setattr(_winconsole, "WriteConsoleW", _capture_write_console(state)) 2 ctx1ab

127 monkeypatch.setattr(_winconsole, "GetLastError", lambda: 0) 2 ctx1ab

128 monkeypatch.setattr( 2 ctx1ab

129 _compat, 

130 "_get_windows_console_stream", 

131 _route_console_stream("stderr", _winconsole._get_text_stderr, state), 

132 ) 

133 

134 result = runner.invoke(app) 2 ctx1ab

135 assert result.exit_code == 0, result.output 2 ctx1ab

136 assert state["stderr_wrap_calls"] > 0 2 ctx1ab

137 assert state["write_calls"] > 0 2 ctx1ab

138 decoded = state["written"].decode("utf-16-le", errors="ignore") 2 ctx1ab

139 assert "Ran out of adventure time!\r\n" in decoded 2 ctx1ab

140 

141 

142@pytest.mark.parametrize( 7 ctx1cdabefg

143 ("writable", "source", "expected_flags"), 

144 [ 

145 (True, bytearray(b"python"), 1), # PyBUF_WRITABLE 

146 (False, b"python", 0), # PyBUF_SIMPLE 

147 ], 

148) 

149def test_get_buffer(monkeypatch, writable, source, expected_flags): 7 ctx1cdabefg

150 state = {"flags": None, "released": 0} 2 ctx1ab

151 if writable: 2 ctx1ab

152 backing = (_winconsole.c_char * len(source)).from_buffer(source) 2 ctx1ab

153 else: 

154 backing = (_winconsole.c_char * len(source)).from_buffer_copy(source) 2 ctx1ab

155 backing_ptr = _winconsole.c_void_p(ctypes.addressof(backing)) 2 ctx1ab

156 

157 def fake_object_get_buffer(obj, buf_ref, flags): # noqa: ARG001 2 ctx1ab

158 state["flags"] = flags 2 ctx1ab

159 buf = buf_ref._obj 2 ctx1ab

160 buf.buf = backing_ptr 2 ctx1ab

161 buf.len = len(source) 2 ctx1ab

162 

163 def fake_buffer_release(buf_ref): # noqa: ARG001 2 ctx1ab

164 state["released"] += 1 2 ctx1ab

165 

166 monkeypatch.setattr(_winconsole, "PyObject_GetBuffer", fake_object_get_buffer) 2 ctx1ab

167 monkeypatch.setattr(_winconsole, "PyBuffer_Release", fake_buffer_release) 2 ctx1ab

168 

169 probe = source if writable else b"x" 2 ctx1ab

170 out = _winconsole.get_buffer(probe, writable=writable) 2 ctx1ab

171 if writable: 2 ctx1ab

172 # mutate the first byte of "python" to obtain another beloved programming language 

173 out[0] = b"c" 2 ctx1ab

174 assert source == bytearray(b"cython") 2 ctx1ab

175 else: 

176 assert bytes(out[: len(source)]) == source 2 ctx1ab

177 assert state["flags"] == expected_flags 2 ctx1ab

178 assert state["released"] == 1 2 ctx1ab

179 

180 

181def test_isatty(): 7 ctx1cdabefg

182 assert _winconsole._WindowsConsoleRawIOBase(None).isatty() is True 2 ctx1ab

183 assert _winconsole._WindowsConsoleReader(0).isatty() is True 2 ctx1ab

184 assert _winconsole._WindowsConsoleReader(1).isatty() is True 2 ctx1ab

185 

186 

187def test_console_stream(): 7 ctx1cdabefg

188 class NamedBytesIO(io.BytesIO): 2 ctx1ab

189 name = "fake-buffer" 2 ctx1ab

190 

191 def isatty(self): 2 ctx1ab

192 return False 2 ctx1ab

193 

194 stream = _winconsole.ConsoleStream( 2 ctx1ab

195 io.TextIOWrapper(io.BytesIO(), encoding="utf-8"), NamedBytesIO() 

196 ) 

197 assert stream.isatty() is False 2 ctx1ab

198 assert stream.name == "fake-buffer" 2 ctx1ab

199 assert "fake-buffer" in repr(stream) 2 ctx1ab

200 assert "utf-8" in repr(stream) 2 ctx1ab

201 

202 # test writelines 

203 stream.writelines(["hello", " ", "world"]) 2 ctx1ab

204 stream._text_stream.flush() 2 ctx1ab

205 assert stream._text_stream.buffer.getvalue().decode("utf-8") == "hello world" 2 ctx1ab

206 

207 # Cover bytes write path. 

208 assert stream.write(b"!") == 1 2 ctx1ab

209 assert stream.buffer.getvalue().endswith(b"!") 2 ctx1ab

210 

211 

212@pytest.mark.parametrize( 7 ctx1cdabefg

213 ("error", "msg"), 

214 [ 

215 (0, "ERROR_SUCCESS"), # ERROR_SUCCESS 

216 (8, "ERROR_NOT_ENOUGH_MEMORY"), # ERROR_NOT_ENOUGH_MEMORY 

217 (342, "Windows error 342"), 

218 ], 

219) 

220def test_error_message(error, msg): 7 ctx1cdabefg

221 writer = _winconsole._WindowsConsoleWriter 2 ctx1ab

222 assert writer._get_error_message(error) == msg 2 ctx1ab

223 

224 

225def test_is_console(): 7 ctx1cdabefg

226 assert _winconsole._is_console(object()) is False 2 ctx1ab

227 

228 

229def test_get_windows_console_stream_factory_and_buffer_paths(monkeypatch): 7 ctx1cdabefg

230 monkeypatch.setattr(_winconsole, "_is_console", lambda f: True) 2 ctx1ab

231 monkeypatch.setattr(_winconsole, "get_buffer", object()) 2 ctx1ab

232 

233 class FakeStream: 2 ctx1ab

234 def __init__(self, fd, buffer=None): 2 ctx1ab

235 self._fd = fd 2 ctx1ab

236 self.buffer = buffer 2 ctx1ab

237 

238 def fileno(self): 2 ctx1ab

239 return self._fd 2 ctx1ab

240 

241 wrapped = {"called": False, "buffer": None} 2 ctx1ab

242 

243 def fake_factory(buffer): 2 ctx1ab

244 wrapped["called"] = True 2 ctx1ab

245 wrapped["buffer"] = buffer 2 ctx1ab

246 return "wrapped", buffer 2 ctx1ab

247 

248 monkeypatch.setattr(_winconsole, "_stream_factories", {7: fake_factory}) 2 ctx1ab

249 

250 # Known console stream preconditions pass, but no stream factory for this fd. 

251 get_stream = _winconsole._get_windows_console_stream 2 ctx1ab

252 assert get_stream(FakeStream(99, object()), "utf-16-le", "strict") is None 2 ctx1ab

253 

254 # Factory exists, but stream has no usable .buffer. 

255 assert get_stream(FakeStream(7, None), "utf-16-le", "strict") is None 2 ctx1ab

256 

257 # Factory exists and buffer is present, so wrapper result is returned. 

258 raw_buffer = object() 2 ctx1ab

259 out = get_stream(FakeStream(7, raw_buffer), "utf-16-le", "strict") 2 ctx1ab

260 assert out == ("wrapped", raw_buffer) 2 ctx1ab

261 assert wrapped["called"] is True 2 ctx1ab

262 assert wrapped["buffer"] is raw_buffer 2 ctx1ab

263 

264 

265def test_windows_console_reader(monkeypatch): 7 ctx1cdabefg

266 reader = _winconsole._WindowsConsoleReader(42) 2 ctx1ab

267 

268 # Empty input buffer returns early 

269 assert reader.readinto(bytearray()) == 0 2 ctx1ab

270 

271 # Require an even number of bytes 

272 with pytest.raises(ValueError): 2 ctx1ab

273 reader.readinto(bytearray(3)) 2 ctx1ab

274 

275 def writable_buffer(obj, writable=False): # noqa: ARG001 2 ctx1ab

276 return (ctypes.c_char * len(obj)).from_buffer(obj) 2 ctx1ab

277 

278 monkeypatch.setattr(_winconsole, "get_buffer", writable_buffer) 2 ctx1ab

279 

280 def patch_console(read_console, error): 2 ctx1ab

281 monkeypatch.setattr(_winconsole, "ReadConsoleW", read_console) 2 ctx1ab

282 monkeypatch.setattr(_winconsole, "GetLastError", lambda: error) 2 ctx1ab

283 

284 def make_read(payload=b"", rv=1, units_read=None): 2 ctx1ab

285 def read_console(handle, buffer, units_to_read, units_read_ptr, reserved): # noqa: ARG001 2 ctx1ab

286 bytes_to_copy = min(len(payload), units_to_read * 2) 2 ctx1ab

287 if bytes_to_copy: 2 ctx1ab

288 buffer[0:bytes_to_copy] = payload[:bytes_to_copy] 2 ctx1ab

289 read_units = units_read if units_read is not None else bytes_to_copy // 2 2 ctx1ab

290 units_read_ptr._obj.value = read_units 2 ctx1ab

291 return rv 2 ctx1ab

292 

293 return read_console 2 ctx1ab

294 

295 # Normal successful read returns the number of bytes read 

296 patch_console(make_read(payload=b"A\x00B\x00"), _winconsole.ERROR_SUCCESS) 2 ctx1ab

297 assert reader.readinto(bytearray(4)) == 4 2 ctx1ab

298 

299 # CTRL+Z (EOF) should be translated into an empty read 

300 patch_console( 2 ctx1ab

301 make_read(payload=_winconsole.EOF + b"\x00", units_read=1), 

302 _winconsole.ERROR_SUCCESS, 

303 ) 

304 assert reader.readinto(bytearray(2)) == 0 2 ctx1ab

305 

306 # An aborted read should sleep briefly while waiting for KeyboardInterrupt 

307 sleep_state = {"calls": 0} 2 ctx1ab

308 

309 def fake_sleep(seconds): 2 ctx1ab

310 sleep_state["calls"] += 1 2 ctx1ab

311 assert seconds == 0.1 2 ctx1ab

312 

313 monkeypatch.setattr( 2 ctx1ab

314 _winconsole, "time", type("FakeTime", (), {"sleep": fake_sleep}) 

315 ) 

316 patch_console( 2 ctx1ab

317 make_read(payload=b"Z\x00", units_read=1), 

318 _winconsole.ERROR_OPERATION_ABORTED, 

319 ) 

320 assert reader.readinto(bytearray(2)) == 2 2 ctx1ab

321 assert sleep_state["calls"] == 1 2 ctx1ab

322 

323 # Failed reads propagate a Windows error 

324 patch_console(make_read(rv=0), _winconsole.ERROR_NOT_ENOUGH_MEMORY) 2 ctx1ab

325 with pytest.raises(OSError, match="Windows error"): 2 ctx1ab

326 reader.readinto(bytearray(2)) 2 ctx1ab