(feat): merge run_cmd to exec_cmd. close #22

pigit/common/utils.py

@@ -1,6 +1,6 @@
 # -*- coding:utf-8 -*-
 
-from typing import Callable, Dict, Iterable, List, Optional, Tuple
+from typing import Callable, Dict, Iterable, List, Optional, Tuple, ByteString, Union
 import sys
 import subprocess
 import asyncio
@@ -35,56 +35,40 @@ def traceback_info(extra_msg: str = "null") -> str:
     )
 
 
-def run_cmd(*args, cwd: Optional[str] = None) -> bool:
-    """Run system command.
+def exec_cmd(
+    *args, cwd: Optional[str] = None, reply: bool = True, decoding: bool = True
+) -> Tuple[Union[str, ByteString, None], ...]:
+    """Run system shell command.
 
-    Returns:
-        (bool): Whether run successful.
-
-    Docs test
-        >>> run_cmd('pwd')
-        True
-        >>> run_cmd('which', 'python')
-        True
+    Args:
+        reply (bool): whether return execute result. Default is True.
+        decoding (bool): whether decode the return. Default is True.
     """
+    _stderr: Optional[int] = subprocess.PIPE if reply else None
+    _stdout: Optional[int] = subprocess.PIPE if reply else None
 
-    try:
-        # ? In python2, `subprocess` not support `with` sentence.
-        proc = subprocess.Popen(" ".join(args), shell=True, cwd=cwd)
-        proc.wait()
-    except Exception:
-        Log.error(traceback_info())
-        return False
-    else:
-        return True
-
-
-def exec_cmd(*args, cwd: Optional[str] = None) -> Tuple[str, str]:
-    """Run system command and get result.
-
-    Returns:
-        (str, str): Error string and result string.
-    """
+    _error: Union[str, ByteString, None] = None
+    _result: Union[str, ByteString, None] = None
 
     try:
         # Take over the input stream and get the return information.
         with subprocess.Popen(
-            " ".join(args),
-            stderr=subprocess.PIPE,
-            stdout=subprocess.PIPE,
-            shell=True,
-            cwd=cwd,
+            " ".join(args), stderr=_stderr, stdout=_stdout, shell=True, cwd=cwd
         ) as proc:
-
-            output = proc.communicate()
-            # Get normal output and error output.
-            res = output[0].decode()
-            err = output[1].decode()
+            outres, errres = proc.communicate()
     except Exception as e:
         Log.error(traceback_info())
-        return str(e), ""
+        _error = str(e).encode()
+    else:
+        _error, _result = errres, outres
+
+    if decoding:
+        return (
+            _error is not None and _error.decode(),
+            _result is not None and _result.decode(),
+        )
     else:
-        return err, res
+        return _error, _result
 
 
 async def async_run_cmd(

pigit/gitlib/_cmd_func.py

@@ -14,7 +14,7 @@
 
 from typing import List, Tuple, Union
 
-from ..common.utils import run_cmd
+from ..common.utils import exec_cmd
 from ..render import echo
 
 
@@ -31,7 +31,7 @@ def add(args: Union[List, Tuple]) -> None:
             "all" if args_str.strip() == "." else args_str
         )
     )
-    run_cmd(f"git add {args_str}")
+    exec_cmd(f"git add {args_str}", reply=False)
 
 
 def fetch_remote_branch(args: Union[List, Tuple]) -> None:
@@ -40,7 +40,7 @@ def fetch_remote_branch(args: Union[List, Tuple]) -> None:
     branch = args[0] if len(args) > 1 else None
 
     if branch:
-        run_cmd("git fetch origin {0}:{0} ".format(branch))
+        exec_cmd("git fetch origin {0}:{0} ".format(branch), reply=False)
     else:
         echo("`This option need a branch name.`<error>")
 
@@ -82,8 +82,8 @@ def set_email_and_username(args: Union[List, Tuple]) -> None:
         else:
             echo("`Bad mailbox format.`<error>")
 
-    if run_cmd(f"git config user.name {name} {is_global}") and run_cmd(
-        f"git config user.email {email} {is_global}"
+    if exec_cmd(f"git config user.name {name} {is_global}", reply=False) and exec_cmd(
+        f"git config user.email {email} {is_global}", reply=False
     ):
         echo("`Successfully set.`<ok>")
     else:

pigit/gitlib/processor.py

@@ -7,7 +7,7 @@
 import random
 import logging
 
-from ..common.utils import run_cmd, confirm, similar_command, traceback_info
+from ..common.utils import exec_cmd, confirm, similar_command, traceback_info
 from ..common.singleton import Singleton
 from ..render import get_console
 from ..render.str_utils import shorten
@@ -150,7 +150,7 @@ def process_command(
                 command = " ".join([command, args_str])
             if self.show_original:
                 get_console().echo(f":rainbow:  {self.color_command(command)}")
-            run_cmd(command)
+            exec_cmd(command, reply=False)
 
     # ============================
     # Print command help message.

pigit/tui/widgets.py

@@ -5,7 +5,7 @@
 from math import ceil
 
 from .console import Term
-from ..common.utils import run_cmd, confirm
+from ..common.utils import exec_cmd, confirm
 from ..render.str_utils import get_width
 from ..render.style import Fx
 
@@ -273,6 +273,6 @@ def __init__(self, cmd: str, auto_run: bool = True, path: str = ".") -> None:
 
     def run(self):
         print(Term.normal_screen)
-        res_code = run_cmd(self.cmd, cwd=self.run_path)
+        res_code = exec_cmd(self.cmd, cwd=self.run_path, reply=False)
         print(Term.alt_screen)
         return res_code

tests/test_common.py

@@ -7,6 +7,7 @@
 from pigit.common.utils import (
     traceback_info,
     confirm,
+    exec_cmd,
     async_run_cmd,
     exec_async_tasks,
 )
@@ -45,6 +46,22 @@ def test_confirm(mock_input, input_value: str, return_bool: bool):
     assert confirm("confirm:") == return_bool
 
 
+def test_exec_cmd():
+    # has reply
+    print(exec_cmd("pwd"))
+    print(exec_cmd("which", "python3"))
+
+    # output to shell
+    print(exec_cmd("pwd", reply=False))
+    print(exec_cmd("which", "python3", reply=False))
+
+    # don't decoding
+    print(exec_cmd("which", "python3", decoding=False))
+
+    # execute error
+    print(exec_cmd("xxxxxxxxx"))
+
+
 def test_async_cmd_func():
     print()
 

tests/test_gitlib.py

@@ -121,11 +121,11 @@ def test_load_extra_cmds(self):
 
 
 class TestCmdFunc:
-    @patch("pigit.gitlib._cmd_func.run_cmd", return_value=None)
+    @patch("pigit.gitlib._cmd_func.exec_cmd", return_value=None)
     def test_add(self, _):
         add([])
 
-    @patch("pigit.gitlib._cmd_func.run_cmd", return_value=None)
+    @patch("pigit.gitlib._cmd_func.exec_cmd", return_value=None)
     def test_fetch_remote(self, _):
         fetch_remote_branch([])
 
@@ -139,6 +139,6 @@ def test_fetch_remote(self, _):
         ],
     )
     @patch("builtins.input", return_value="abc@gmail.com")
-    @patch("pigit.gitlib._cmd_func.run_cmd", return_value=False)
+    @patch("pigit.gitlib._cmd_func.exec_cmd", return_value=False)
     def test_set_ua(self, _a, _b, args):
         set_email_and_username(args)