Skip to content

Commit a371fc7

Browse files
committed
test(wrap_stdio): does not load other platform
1 parent 6677754 commit a371fc7

File tree

2 files changed

+121
-122
lines changed

2 files changed

+121
-122
lines changed

commitizen/wrap_stdio_linux.py

+51-53
Original file line numberDiff line numberDiff line change
@@ -1,55 +1,53 @@
1-
import os
21
import sys
3-
from io import IOBase
42

5-
6-
class WrapStdioLinux:
7-
def __init__(self, stdx: IOBase):
8-
self._fileno = stdx.fileno()
9-
if self._fileno == 0:
10-
fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY)
11-
tty = open(fd, "wb+", buffering=0)
12-
else:
13-
tty = open("/dev/tty", "w") # type: ignore
14-
self.tty = tty
15-
16-
def __getattr__(self, key):
17-
if key == "encoding" and self._fileno == 0:
18-
return "UTF-8"
19-
return getattr(self.tty, key)
20-
21-
def __del__(self):
22-
self.tty.close()
23-
24-
25-
backup_stdin = None
26-
backup_stdout = None
27-
backup_stderr = None
28-
29-
30-
def _wrap_stdio():
31-
global backup_stdin
32-
backup_stdin = sys.stdin
33-
sys.stdin = WrapStdioLinux(sys.stdin)
34-
35-
global backup_stdout
36-
backup_stdout = sys.stdout
37-
sys.stdout = WrapStdioLinux(sys.stdout)
38-
39-
global backup_stderr
40-
backup_stderr = sys.stderr
41-
sys.stderr = WrapStdioLinux(sys.stderr)
42-
43-
44-
def _unwrap_stdio():
45-
global backup_stdin
46-
sys.stdin.close()
47-
sys.stdin = backup_stdin
48-
49-
global backup_stdout
50-
sys.stdout.close()
51-
sys.stdout = backup_stdout
52-
53-
global backup_stderr
54-
sys.stderr.close()
55-
sys.stderr = backup_stderr
3+
if sys.platform == "linux": # pragma: no cover
4+
import os
5+
from io import IOBase
6+
7+
class WrapStdioLinux:
8+
def __init__(self, stdx: IOBase):
9+
self._fileno = stdx.fileno()
10+
if self._fileno == 0:
11+
fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY)
12+
tty = open(fd, "wb+", buffering=0)
13+
else:
14+
tty = open("/dev/tty", "w") # type: ignore
15+
self.tty = tty
16+
17+
def __getattr__(self, key):
18+
if key == "encoding" and self._fileno == 0:
19+
return "UTF-8"
20+
return getattr(self.tty, key)
21+
22+
def __del__(self):
23+
self.tty.close()
24+
25+
backup_stdin = None
26+
backup_stdout = None
27+
backup_stderr = None
28+
29+
def _wrap_stdio():
30+
global backup_stdin
31+
backup_stdin = sys.stdin
32+
sys.stdin = WrapStdioLinux(sys.stdin)
33+
34+
global backup_stdout
35+
backup_stdout = sys.stdout
36+
sys.stdout = WrapStdioLinux(sys.stdout)
37+
38+
global backup_stderr
39+
backup_stderr = sys.stderr
40+
sys.stderr = WrapStdioLinux(sys.stderr)
41+
42+
def _unwrap_stdio():
43+
global backup_stdin
44+
sys.stdin.close()
45+
sys.stdin = backup_stdin
46+
47+
global backup_stdout
48+
sys.stdout.close()
49+
sys.stdout = backup_stdout
50+
51+
global backup_stderr
52+
sys.stderr.close()
53+
sys.stderr = backup_stderr

commitizen/wrap_stdio_unix.py

+70-69
Original file line numberDiff line numberDiff line change
@@ -1,71 +1,72 @@
1-
import os
2-
import selectors
31
import sys
4-
from asyncio import DefaultEventLoopPolicy, get_event_loop_policy, set_event_loop_policy
5-
from io import IOBase
62

7-
8-
class CZEventLoopPolicy(DefaultEventLoopPolicy): # type: ignore
9-
def get_event_loop(self):
10-
self.set_event_loop(self._loop_factory(selectors.SelectSelector()))
11-
return self._local._loop
12-
13-
14-
class WrapStdioUnix:
15-
def __init__(self, stdx: IOBase):
16-
self._fileno = stdx.fileno()
17-
fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY)
18-
if self._fileno == 0:
19-
tty = open(fd, "wb+", buffering=0)
20-
else:
21-
tty = open(fd, "rb+", buffering=0)
22-
self.tty = tty
23-
24-
def __getattr__(self, key):
25-
if key == "encoding":
26-
return "UTF-8"
27-
return getattr(self.tty, key)
28-
29-
def __del__(self):
30-
self.tty.close()
31-
32-
33-
backup_event_loop_policy = None
34-
backup_stdin = None
35-
backup_stdout = None
36-
backup_stderr = None
37-
38-
39-
def _wrap_stdio():
40-
global backup_event_loop_policy
41-
backup_event_loop_policy = get_event_loop_policy()
42-
set_event_loop_policy(CZEventLoopPolicy())
43-
44-
global backup_stdin
45-
backup_stdin = sys.stdin
46-
sys.stdin = WrapStdioUnix(sys.stdin)
47-
48-
global backup_stdout
49-
backup_stdout = sys.stdout
50-
sys.stdout = WrapStdioUnix(sys.stdout)
51-
52-
global backup_stderr
53-
backup_stdout = sys.stderr
54-
sys.stderr = WrapStdioUnix(sys.stderr)
55-
56-
57-
def _unwrap_stdio():
58-
global backup_event_loop_policy
59-
set_event_loop_policy(backup_event_loop_policy)
60-
61-
global backup_stdin
62-
sys.stdin.close()
63-
sys.stdin = backup_stdin
64-
65-
global backup_stdout
66-
sys.stdout.close()
67-
sys.stdout = backup_stdout
68-
69-
global backup_stderr
70-
sys.stderr.close()
71-
sys.stderr = backup_stderr
3+
if sys.platform != "win32" and sys.platform != "linux": # pragma: no cover
4+
import os
5+
import selectors
6+
from asyncio import (
7+
DefaultEventLoopPolicy,
8+
get_event_loop_policy,
9+
set_event_loop_policy,
10+
)
11+
from io import IOBase
12+
13+
class CZEventLoopPolicy(DefaultEventLoopPolicy): # pragma: no cover
14+
def get_event_loop(self):
15+
self.set_event_loop(self._loop_factory(selectors.SelectSelector()))
16+
return self._local._loop
17+
18+
class WrapStdioUnix:
19+
def __init__(self, stdx: IOBase):
20+
self._fileno = stdx.fileno()
21+
fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY)
22+
if self._fileno == 0:
23+
tty = open(fd, "wb+", buffering=0)
24+
else:
25+
tty = open(fd, "rb+", buffering=0)
26+
self.tty = tty
27+
28+
def __getattr__(self, key):
29+
if key == "encoding":
30+
return "UTF-8"
31+
return getattr(self.tty, key)
32+
33+
def __del__(self):
34+
self.tty.close()
35+
36+
backup_event_loop_policy = None
37+
backup_stdin = None
38+
backup_stdout = None
39+
backup_stderr = None
40+
41+
def _wrap_stdio():
42+
global backup_event_loop_policy
43+
backup_event_loop_policy = get_event_loop_policy()
44+
set_event_loop_policy(CZEventLoopPolicy())
45+
46+
global backup_stdin
47+
backup_stdin = sys.stdin
48+
sys.stdin = WrapStdioUnix(sys.stdin)
49+
50+
global backup_stdout
51+
backup_stdout = sys.stdout
52+
sys.stdout = WrapStdioUnix(sys.stdout)
53+
54+
global backup_stderr
55+
backup_stdout = sys.stderr
56+
sys.stderr = WrapStdioUnix(sys.stderr)
57+
58+
def _unwrap_stdio():
59+
global backup_event_loop_policy
60+
set_event_loop_policy(backup_event_loop_policy)
61+
62+
global backup_stdin
63+
sys.stdin.close()
64+
sys.stdin = backup_stdin
65+
66+
global backup_stdout
67+
sys.stdout.close()
68+
sys.stdout = backup_stdout
69+
70+
global backup_stderr
71+
sys.stderr.close()
72+
sys.stderr = backup_stderr

0 commit comments

Comments
 (0)