feat(wrap_stdio): separate wrap stdio module

Author: saygox

commitizen/commands/commit.py

@@ -2,13 +2,9 @@
 
 import contextlib
 import os
-import selectors
 import shutil
 import subprocess
-import sys
 import tempfile
-from asyncio import DefaultEventLoopPolicy, get_event_loop_policy, set_event_loop_policy
-from io import IOBase
 
 import questionary
 
@@ -28,38 +24,7 @@
     NothingToCommitError,
 )
 from commitizen.git import smart_open
-
-
-class CZEventLoopPolicy(DefaultEventLoopPolicy):  # type: ignore
-    def get_event_loop(self):
-        self.set_event_loop(self._loop_factory(selectors.SelectSelector()))
-        return self._local._loop
-
-
-class WrapStdx:
-    def __init__(self, stdx: IOBase):
-        self._fileno = stdx.fileno()
-        if sys.platform == "linux":
-            if self._fileno == 0:
-                fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY)
-                tty = open(fd, "wb+", buffering=0)
-            else:
-                tty = open("/dev/tty", "w")  # type: ignore
-        else:
-            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" and (sys.platform != "linux" or self._fileno == 0):
-            return "UTF-8"
-        return getattr(self.tty, key)
-
-    def __del__(self):
-        self.tty.close()
+from commitizen.wrap_stdio import unwrap_stdio, wrap_stdio
 
 
 class Commit:
@@ -143,14 +108,7 @@ def __call__(self):
 
         commit_msg_file: str = self.arguments.get("commit_msg_file")
         if commit_msg_file:
-            old_stdin = sys.stdin
-            old_stdout = sys.stdout
-            old_stderr = sys.stderr
-            old_event_loop_policy = get_event_loop_policy()
-            set_event_loop_policy(CZEventLoopPolicy())
-            sys.stdin = WrapStdx(sys.stdin)
-            sys.stdout = WrapStdx(sys.stdout)
-            sys.stderr = WrapStdx(sys.stderr)
+            wrap_stdio()
 
         if git.is_staging_clean() and not (dry_run or allow_empty):
             raise NothingToCommitError("No files added to staging!")
@@ -174,13 +132,7 @@ def __call__(self):
             m = self.prompt_commit_questions()
 
         if commit_msg_file:
-            sys.stdin.close()
-            sys.stdout.close()
-            sys.stderr.close()
-            set_event_loop_policy(old_event_loop_policy)
-            sys.stdin = old_stdin
-            sys.stdout = old_stdout
-            sys.stderr = old_stderr
+            unwrap_stdio()
 
         if manual_edit:
             m = self.manual_edit(m)

commitizen/wrap_stdio.py

@@ -0,0 +1,18 @@
+import sys
+
+if sys.platform == "win32":  # pragma: no cover
+    from .wrap_stdio_windows import _unwrap_stdio, _wrap_stdio
+elif sys.platform == "linux":
+    from .wrap_stdio_linux import _unwrap_stdio, _wrap_stdio  # pragma: no cover
+else:
+    from .wrap_stdio_unix import _unwrap_stdio, _wrap_stdio  # pragma: no cover
+
+
+def wrap_stdio():
+    _wrap_stdio()
+    return None
+
+
+def unwrap_stdio():
+    _unwrap_stdio()
+    return None

commitizen/wrap_stdio_linux.py

@@ -0,0 +1,55 @@
+import os
+import sys
+from io import IOBase
+
+
+class WrapStdioLinux:
+    def __init__(self, stdx: IOBase):
+        self._fileno = stdx.fileno()
+        if self._fileno == 0:
+            fd = os.open("/dev/tty", os.O_RDWR | os.O_NOCTTY)
+            tty = open(fd, "wb+", buffering=0)
+        else:
+            tty = open("/dev/tty", "w")  # type: ignore
+        self.tty = tty
+
+    def __getattr__(self, key):
+        if key == "encoding" and self._fileno == 0:
+            return "UTF-8"
+        return getattr(self.tty, key)
+
+    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 = WrapStdioLinux(sys.stdin)
+
+    global backup_stdout
+    backup_stdout = sys.stdout
+    sys.stdout = WrapStdioLinux(sys.stdout)
+
+    global backup_stderr
+    backup_stderr = sys.stderr
+    sys.stderr = WrapStdioLinux(sys.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

@@ -0,0 +1,71 @@
+import os
+import selectors
+import sys
+from asyncio import DefaultEventLoopPolicy, get_event_loop_policy, set_event_loop_policy
+from io import IOBase
+
+
+class CZEventLoopPolicy(DefaultEventLoopPolicy):  # type: ignore
+    def get_event_loop(self):
+        self.set_event_loop(self._loop_factory(selectors.SelectSelector()))
+        return self._local._loop
+
+
+class WrapStdioLinux:
+    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()
+    set_event_loop_policy(CZEventLoopPolicy())
+
+    global backup_stdin
+    backup_stdin = sys.stdin
+    sys.stdin = WrapStdioLinux(sys.stdin)
+
+    global backup_stdout
+    backup_stdout = sys.stdout
+    sys.stdout = WrapStdioLinux(sys.stdout)
+
+    global backup_stderr
+    backup_stdout = sys.stderr
+    sys.stderr = WrapStdioLinux(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

commitizen/wrap_stdio_windows.py

@@ -0,0 +1,6 @@
+def _wrap_stdio():
+    pass
+
+
+def _unwrap_stdio():
+    pass

tests/commands/test_commit_command.py

@@ -543,8 +543,9 @@ def test_commit_from_pre_commit_msg_hook(config, mocker, capsys):
 
     commit_mock = mocker.patch("commitizen.git.commit")
     commit_mock.return_value = cmd.Command("success", "", "", "", 0)
-    mocker.patch("commitizen.commands.commit.WrapStdx")
-    mocker.patch("os.open")
+
+    mocker.patch("commitizen.commands.commit.wrap_stdio")
+    mocker.patch("commitizen.commands.commit.unwrap_stdio")
     reader_mock = mocker.mock_open(read_data="\n\n#test\n")
     mocker.patch("builtins.open", reader_mock, create=True)
 
@@ -553,43 +554,3 @@ def test_commit_from_pre_commit_msg_hook(config, mocker, capsys):
     out, _ = capsys.readouterr()
     assert "Commit message is successful!" in out
     commit_mock.assert_not_called()
-
-
-def test_WrapStdx(mocker):
-    mocker.patch("os.open")
-    reader_mock = mocker.mock_open(read_data="data")
-    mocker.patch("builtins.open", reader_mock, create=True)
-
-    stdin_mock_fileno = mocker.patch.object(sys.stdin, "fileno")
-    stdin_mock_fileno.return_value = 0
-    wrap_stdin = commands.commit.WrapStdx(sys.stdin)
-
-    assert wrap_stdin.encoding == "UTF-8"
-    assert wrap_stdin.read() == "data"
-
-    writer_mock = mocker.mock_open(read_data="data")
-    mocker.patch("builtins.open", writer_mock, create=True)
-    stdout_mock_fileno = mocker.patch.object(sys.stdout, "fileno")
-    stdout_mock_fileno.return_value = 1
-    wrap_stout = commands.commit.WrapStdx(sys.stdout)
-    wrap_stout.write("data")
-
-    if sys.platform == "linux":
-        writer_mock.assert_called_once_with("/dev/tty", "w")
-    else:
-        pass
-    writer_mock().write.assert_called_once_with("data")
-
-    writer_mock = mocker.mock_open(read_data="data")
-    mocker.patch("builtins.open", writer_mock, create=True)
-    stderr_mock_fileno = mocker.patch.object(sys.stdout, "fileno")
-    stderr_mock_fileno.return_value = 2
-    wrap_sterr = commands.commit.WrapStdx(sys.stderr)
-
-    wrap_sterr.write("data")
-
-    if sys.platform == "linux":
-        writer_mock.assert_called_once_with("/dev/tty", "w")
-    else:
-        pass
-    writer_mock().write.assert_called_once_with("data")