23
23
<https://github.com/pytest-dev/pytest/blob/master/src/_pytest/debugging.py>`_.
24
24
25
25
"""
26
+ from __future__ import annotations
27
+
26
28
import contextlib
27
29
import functools
28
30
import io
31
33
from tempfile import TemporaryFile
32
34
from typing import Any
33
35
from typing import AnyStr
34
- from typing import Dict
35
36
from typing import Generator
36
37
from typing import Generic
37
38
from typing import Iterator
38
- from typing import Optional
39
39
from typing import TextIO
40
- from typing import Tuple
41
40
from typing import TYPE_CHECKING
42
- from typing import Union
43
41
44
42
import click
45
43
from _pytask .config import hookimpl
@@ -92,9 +90,9 @@ def pytask_extend_command_line_interface(cli: click.Group) -> None:
92
90
93
91
@hookimpl
94
92
def pytask_parse_config (
95
- config : Dict [str , Any ],
96
- config_from_cli : Dict [str , Any ],
97
- config_from_file : Dict [str , Any ],
93
+ config : dict [str , Any ],
94
+ config_from_cli : dict [str , Any ],
95
+ config_from_file : dict [str , Any ],
98
96
) -> None :
99
97
"""Parse configuration.
100
98
@@ -122,10 +120,8 @@ def pytask_parse_config(
122
120
123
121
124
122
@hookimpl
125
- def pytask_post_parse (config : Dict [str , Any ]) -> None :
123
+ def pytask_post_parse (config : dict [str , Any ]) -> None :
126
124
"""Initialize the CaptureManager."""
127
- if config ["capture" ] == "fd" :
128
- _py36_windowsconsoleio_workaround (sys .stdout )
129
125
_colorama_workaround ()
130
126
131
127
pluginmanager = config ["pm" ]
@@ -136,7 +132,7 @@ def pytask_post_parse(config: Dict[str, Any]) -> None:
136
132
capman .suspend ()
137
133
138
134
139
- def _capture_callback (x : "Optional[ _CaptureMethod]" ) -> "Optional[ _CaptureMethod]" :
135
+ def _capture_callback (x : _CaptureMethod | None ) -> _CaptureMethod | None :
140
136
"""Validate the passed options for capturing output."""
141
137
if x in [None , "None" , "none" ]:
142
138
x = None
@@ -148,8 +144,8 @@ def _capture_callback(x: "Optional[_CaptureMethod]") -> "Optional[_CaptureMethod
148
144
149
145
150
146
def _show_capture_callback (
151
- x : "Optional[ _CaptureCallback]" ,
152
- ) -> "Optional[ _CaptureCallback]" :
147
+ x : _CaptureCallback | None ,
148
+ ) -> _CaptureCallback | None :
153
149
"""Validate the passed options for showing captured output."""
154
150
if x in [None , "None" , "none" ]:
155
151
x = None
@@ -181,66 +177,6 @@ def _colorama_workaround() -> None:
181
177
pass
182
178
183
179
184
- def _py36_windowsconsoleio_workaround (stream : TextIO ) -> None :
185
- """Workaround for Windows Unicode console handling on Python>=3.6.
186
-
187
- Python 3.6 implemented Unicode console handling for Windows. This works by
188
- reading/writing to the raw console handle using ``{Read,Write}ConsoleW``.
189
-
190
- The problem is that we are going to ``dup2`` over the stdio file descriptors when
191
- doing ``FDCapture`` and this will ``CloseHandle`` the handles used by Python to
192
- write to the console. Though there is still some weirdness and the console handle
193
- seems to only be closed randomly and not on the first call to ``CloseHandle``, or
194
- maybe it gets reopened with the same handle value when we suspend capturing.
195
-
196
- The workaround in this case will reopen stdio with a different fd which also means a
197
- different handle by replicating the logic in
198
- "Py_lifecycle.c:initstdio/create_stdio".
199
-
200
- Parameters
201
- ---------
202
- stream
203
- In practice ``sys.stdout`` or ``sys.stderr``, but given here as parameter for
204
- unit testing purposes.
205
-
206
- See https://github.com/pytest-dev/py/issues/103.
207
-
208
- """
209
- if not sys .platform .startswith ("win32" ) or hasattr (sys , "pypy_version_info" ):
210
- return
211
-
212
- # Bail out if ``stream`` doesn't seem like a proper ``io`` stream (#2666).
213
- if not hasattr (stream , "buffer" ):
214
- return
215
-
216
- buffered = hasattr (stream .buffer , "raw" )
217
- # ``getattr`` hack since ``buffer`` might not have an attribute ``raw``.
218
- raw_stdout = getattr (stream .buffer , "raw" , stream .buffer )
219
-
220
- # ``getattr`` hack since ``_WindowsConsoleIO`` is not defined in stubs.
221
- windowsconsoleio = getattr (io , "_WindowsConsoleIO" , None )
222
- if windowsconsoleio is not None and not isinstance (raw_stdout , windowsconsoleio ):
223
- return
224
-
225
- def _reopen_stdio (f : TextIO , mode : str ) -> TextIO :
226
- if not buffered and mode [0 ] == "w" :
227
- buffering = 0
228
- else :
229
- buffering = - 1
230
-
231
- return io .TextIOWrapper (
232
- open (os .dup (f .fileno ()), mode , buffering ),
233
- f .encoding ,
234
- f .errors ,
235
- f .newlines ,
236
- bool (f .line_buffering ),
237
- )
238
-
239
- sys .stdin = _reopen_stdio (sys .stdin , "rb" )
240
- sys .stdout = _reopen_stdio (sys .stdout , "wb" )
241
- sys .stderr = _reopen_stdio (sys .stderr , "wb" )
242
-
243
-
244
180
# IO Helpers.
245
181
246
182
@@ -292,7 +228,7 @@ def read(self, *_args: Any) -> None: # noqa: U101
292
228
readlines = read
293
229
__next__ = read
294
230
295
- def __iter__ (self ) -> " DontReadFromInput" :
231
+ def __iter__ (self ) -> DontReadFromInput :
296
232
return self
297
233
298
234
def fileno (self ) -> int :
@@ -305,7 +241,7 @@ def close(self) -> None:
305
241
pass
306
242
307
243
@property
308
- def buffer (self ) -> " DontReadFromInput" :
244
+ def buffer (self ) -> DontReadFromInput :
309
245
return self
310
246
311
247
@@ -368,7 +304,7 @@ def __repr__(self) -> str:
368
304
self .tmpfile ,
369
305
)
370
306
371
- def _assert_state (self , op : str , states : Tuple [str , ...]) -> None :
307
+ def _assert_state (self , op : str , states : tuple [str , ...]) -> None :
372
308
assert (
373
309
self ._state in states
374
310
), "cannot {} in state {!r}: expected one of {}" .format (
@@ -463,7 +399,7 @@ def __init__(self, targetfd: int) -> None:
463
399
# Further complications are the need to support suspend() and the
464
400
# possibility of FD reuse (e.g. the tmpfile getting the very same target
465
401
# FD). The following approach is robust, I believe.
466
- self .targetfd_invalid : Optional [ int ] = os .open (os .devnull , os .O_RDWR )
402
+ self .targetfd_invalid : int | None = os .open (os .devnull , os .O_RDWR )
467
403
os .dup2 (self .targetfd_invalid , targetfd )
468
404
else :
469
405
self .targetfd_invalid = None
@@ -496,7 +432,7 @@ def __repr__(self) -> str:
496
432
self .tmpfile ,
497
433
)
498
434
499
- def _assert_state (self , op : str , states : Tuple [str , ...]) -> None :
435
+ def _assert_state (self , op : str , states : tuple [str , ...]) -> None :
500
436
assert (
501
437
self ._state in states
502
438
), "cannot {} in state {!r}: expected one of {}" .format (
@@ -614,8 +550,8 @@ def __getitem__(self, item: int) -> AnyStr:
614
550
return tuple (self )[item ]
615
551
616
552
def _replace (
617
- self , * , out : Optional [ AnyStr ] = None , err : Optional [ AnyStr ] = None
618
- ) -> " CaptureResult[AnyStr]" :
553
+ self , * , out : AnyStr | None = None , err : AnyStr | None = None
554
+ ) -> CaptureResult [AnyStr ]:
619
555
return CaptureResult (
620
556
out = self .out if out is None else out , err = self .err if err is None else err
621
557
)
@@ -657,9 +593,9 @@ class MultiCapture(Generic[AnyStr]):
657
593
658
594
def __init__ (
659
595
self ,
660
- in_ : Optional [ Union [ FDCapture , SysCapture ]] ,
661
- out : Optional [ Union [ FDCapture , SysCapture ]] ,
662
- err : Optional [ Union [ FDCapture , SysCapture ]] ,
596
+ in_ : FDCapture | SysCapture | None ,
597
+ out : FDCapture | SysCapture | None ,
598
+ err : FDCapture | SysCapture | None ,
663
599
) -> None :
664
600
self .in_ = in_
665
601
self .out = out
@@ -686,7 +622,7 @@ def start_capturing(self) -> None:
686
622
if self .err :
687
623
self .err .start ()
688
624
689
- def pop_outerr_to_orig (self ) -> Tuple [AnyStr , AnyStr ]:
625
+ def pop_outerr_to_orig (self ) -> tuple [AnyStr , AnyStr ]:
690
626
"""Pop current snapshot out/err capture and flush to orig streams."""
691
627
out , err = self .readouterr ()
692
628
if out :
@@ -743,7 +679,7 @@ def readouterr(self) -> CaptureResult[AnyStr]:
743
679
return CaptureResult (out , err ) # type: ignore
744
680
745
681
746
- def _get_multicapture (method : " _CaptureMethod" ) -> MultiCapture [str ]:
682
+ def _get_multicapture (method : _CaptureMethod ) -> MultiCapture [str ]:
747
683
"""Set up the MultiCapture class with the passed method.
748
684
749
685
For each valid method, the function instantiates the :class:`MultiCapture` class
@@ -779,9 +715,9 @@ class CaptureManager:
779
715
780
716
"""
781
717
782
- def __init__ (self , method : " _CaptureMethod" ) -> None :
718
+ def __init__ (self , method : _CaptureMethod ) -> None :
783
719
self ._method = method
784
- self ._capturing : Optional [ MultiCapture [str ]] = None
720
+ self ._capturing : MultiCapture [str ] | None = None
785
721
786
722
def __repr__ (self ) -> str :
787
723
return ("<CaptureManager _method={!r} _capturing={!r}>" ).format (
0 commit comments