1
1
import os
2
- import signal
3
2
import ssl
4
3
import threading
5
4
from base64 import b64encode
11
10
from werkzeug .serving import BaseWSGIServer , WSGIRequestHandler
12
11
from werkzeug .serving import make_server as _make_server
13
12
14
- from .compat import nullcontext
13
+ from .compat import blocked_signals
15
14
16
15
if TYPE_CHECKING :
17
16
from _typeshed .wsgi import StartResponse , WSGIApplication , WSGIEnvironment
@@ -23,37 +22,6 @@ class MockServer(BaseWSGIServer):
23
22
mock : Mock = Mock ()
24
23
25
24
26
- # Applies on Windows.
27
- if not hasattr (signal , "pthread_sigmask" ):
28
- # We're not relying on this behavior anywhere currently, it's just best
29
- # practice.
30
- blocked_signals = nullcontext
31
- else :
32
-
33
- @contextmanager
34
- def blocked_signals () -> Iterator [None ]:
35
- """Block all signals for e.g. starting a worker thread."""
36
- # valid_signals() was added in Python 3.8 (and not using it results
37
- # in a warning on pthread_sigmask() call)
38
- mask : Iterable [int ]
39
- try :
40
- mask = signal .valid_signals ()
41
- except AttributeError :
42
- mask = set (range (1 , signal .NSIG ))
43
-
44
- old_mask = signal .pthread_sigmask ( # type: ignore[attr-defined]
45
- signal .SIG_SETMASK , # type: ignore[attr-defined]
46
- mask ,
47
- )
48
- try :
49
- yield
50
- finally :
51
- signal .pthread_sigmask ( # type: ignore[attr-defined]
52
- signal .SIG_SETMASK , # type: ignore[attr-defined]
53
- old_mask ,
54
- )
55
-
56
-
57
25
class _RequestHandler (WSGIRequestHandler ):
58
26
def make_environ (self ) -> Dict [str , Any ]:
59
27
environ = super ().make_environ ()
0 commit comments