refactor: wrap_stdio

Author: saygox

commitizen/wrap_stdio/linux.py

@@ -1,61 +1,62 @@
+import os
 import sys
 
-if sys.platform == "linux":  # pragma: no cover
-    import os
 
-    # from io import IOBase
+class WrapStdinLinux:
+    def __init__(self):
+        fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY)
+        tty = open(fd, "wb+", buffering=0)
+        self.tty = tty
 
-    class WrapStdinLinux:
-        def __init__(self):
-            fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY)
-            tty = open(fd, "wb+", buffering=0)
-            self.tty = tty
+    def __getattr__(self, key):
+        if key == "encoding":
+            return "UTF-8"
+        return getattr(self.tty, key)
 
-        def __getattr__(self, key):
-            if key == "encoding":
-                return "UTF-8"
-            return getattr(self.tty, key)
+    def __del__(self):
+        self.tty.close()
 
-        def __del__(self):
-            self.tty.close()
 
-    class WrapStdoutLinux:
-        def __init__(self):
-            tty = open("/dev/tty", "w")
-            self.tty = tty
+class WrapStdoutLinux:
+    def __init__(self):
+        tty = open("/dev/tty", "w")
+        self.tty = tty
 
-        def __getattr__(self, key):
-            return getattr(self.tty, key)
+    def __getattr__(self, key):
+        return getattr(self.tty, key)
 
-        def __del__(self):
-            self.tty.close()
+    def __del__(self):
+        self.tty.close()
 
-    backup_stdin = None
-    backup_stdout = None
-    backup_stderr = None
 
-    def _wrap_stdio():
-        global backup_stdin
-        backup_stdin = sys.stdin
-        sys.stdin = WrapStdinLinux()
+backup_stdin = None
+backup_stdout = None
+backup_stderr = None
 
-        global backup_stdout
-        backup_stdout = sys.stdout
-        sys.stdout = WrapStdoutLinux()
 
-        global backup_stderr
-        backup_stderr = sys.stderr
-        sys.stderr = WrapStdoutLinux()
+def _wrap_stdio():
+    global backup_stdin
+    backup_stdin = sys.stdin
+    sys.stdin = WrapStdinLinux()
 
-    def _unwrap_stdio():
-        global backup_stdin
-        sys.stdin.close()
-        sys.stdin = backup_stdin
+    global backup_stdout
+    backup_stdout = sys.stdout
+    sys.stdout = WrapStdoutLinux()
 
-        global backup_stdout
-        sys.stdout.close()
-        sys.stdout = backup_stdout
+    global backup_stderr
+    backup_stderr = sys.stderr
+    sys.stderr = WrapStdoutLinux()
 
-        global backup_stderr
-        sys.stderr.close()
-        sys.stderr = backup_stderr
+
+def _unwrap_stdio():
+    global backup_stdin
+    sys.stdin.close()
+    sys.stdin = backup_stdin
+
+    global backup_stdout
+    sys.stdout.close()
+    sys.stdout = backup_stdout
+
+    global backup_stderr
+    sys.stderr.close()
+    sys.stderr = backup_stderr

commitizen/wrap_stdio/unix.py

@@ -1,72 +1,73 @@
+import os
+import selectors
 import sys
+from asyncio import (
+    DefaultEventLoopPolicy,
+    SelectorEventLoop,
+    get_event_loop_policy,
+    set_event_loop_policy,
+)
+from io import IOBase
 
-if sys.platform != "win32" and sys.platform != "linux":  # pragma: no cover
-    import os
-    import selectors
-    from asyncio import (
-        DefaultEventLoopPolicy,
-        SelectorEventLoop,
-        get_event_loop_policy,
-        set_event_loop_policy,
-    )
-    from io import IOBase
-
-    class WrapStdioUnix:
-        def __init__(self, stdx: IOBase):
-            self._fileno = stdx.fileno()
-            fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY)
-            if self._fileno == 0:
-                tty = open(fd, "wb+", buffering=0)
-            else:
-                tty = open(fd, "rb+", buffering=0)
-            self.tty = tty
-
-        def __getattr__(self, key):
-            if key == "encoding":
-                return "UTF-8"
-            return getattr(self.tty, key)
-
-        def __del__(self):
-            self.tty.close()
-
-    # backup_event_loop = None
-    backup_event_loop_policy = None
-    backup_stdin = None
-    backup_stdout = None
-    backup_stderr = None
-
-    def _wrap_stdio():
-        global backup_event_loop_policy
-        backup_event_loop_policy = get_event_loop_policy()
-
-        event_loop = DefaultEventLoopPolicy()
-        event_loop.set_event_loop(SelectorEventLoop(selectors.SelectSelector()))
-        set_event_loop_policy(event_loop)
-
-        global backup_stdin
-        backup_stdin = sys.stdin
-        sys.stdin = WrapStdioUnix(sys.stdin)
-
-        global backup_stdout
-        backup_stdout = sys.stdout
-        sys.stdout = WrapStdioUnix(sys.stdout)
-
-        global backup_stderr
-        backup_stdout = sys.stderr
-        sys.stderr = WrapStdioUnix(sys.stderr)
-
-    def _unwrap_stdio():
-        global backup_event_loop_policy
-        set_event_loop_policy(backup_event_loop_policy)
-
-        global backup_stdin
-        sys.stdin.close()
-        sys.stdin = backup_stdin
-
-        global backup_stdout
-        sys.stdout.close()
-        sys.stdout = backup_stdout
-
-        global backup_stderr
-        sys.stderr.close()
-        sys.stderr = backup_stderr
+
+class WrapStdioUnix:
+    def __init__(self, stdx: IOBase):
+        self._fileno = stdx.fileno()
+        fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY)
+        if self._fileno == 0:
+            tty = open(fd, "wb+", buffering=0)
+        else:
+            tty = open(fd, "rb+", buffering=0)
+        self.tty = tty
+
+    def __getattr__(self, key):
+        if key == "encoding":
+            return "UTF-8"
+        return getattr(self.tty, key)
+
+    def __del__(self):
+        self.tty.close()
+
+
+backup_event_loop_policy = None
+backup_stdin = None
+backup_stdout = None
+backup_stderr = None
+
+
+def _wrap_stdio():
+    global backup_event_loop_policy
+    backup_event_loop_policy = get_event_loop_policy()
+
+    event_loop = DefaultEventLoopPolicy()
+    event_loop.set_event_loop(SelectorEventLoop(selectors.SelectSelector()))
+    set_event_loop_policy(event_loop)
+
+    global backup_stdin
+    backup_stdin = sys.stdin
+    sys.stdin = WrapStdioUnix(sys.stdin)
+
+    global backup_stdout
+    backup_stdout = sys.stdout
+    sys.stdout = WrapStdioUnix(sys.stdout)
+
+    global backup_stderr
+    backup_stdout = sys.stderr
+    sys.stderr = WrapStdioUnix(sys.stderr)
+
+
+def _unwrap_stdio():
+    global backup_event_loop_policy
+    set_event_loop_policy(backup_event_loop_policy)
+
+    global backup_stdin
+    sys.stdin.close()
+    sys.stdin = backup_stdin
+
+    global backup_stdout
+    sys.stdout.close()
+    sys.stdout = backup_stdout
+
+    global backup_stderr
+    sys.stderr.close()
+    sys.stderr = backup_stderr