feat(hooks): support interactive hook scripts

commitizen/cmd.py

@@ -36,6 +36,15 @@ def _try_decode(bytes_: bytes) -> str:
 
 
 def run(cmd: str, env: Mapping[str, str] | None = None) -> Command:
+    """Run a command in a subprocess and capture stdout and stderr
+
+    Args:
+        cmd: The command to run
+        env: Extra environment variables to define in the subprocess. Defaults to None.
+
+    Returns:
+        Command: _description_
+    """
     if env is not None:
         env = {**os.environ, **env}
     process = subprocess.Popen(
@@ -55,3 +64,18 @@ def run(cmd: str, env: Mapping[str, str] | None = None) -> Command:
         stderr,
         return_code,
     )
+
+
+def run_interactive(cmd: str, env: Mapping[str, str] | None = None) -> int:
+    """Run a command in a subprocess without redirecting stdin, stdout, or stderr
+
+    Args:
+        cmd: The command to run
+        env: Extra environment variables to define in the subprocess. Defaults to None.
+
+    Returns:
+        subprocess returncode
+    """
+    if env is not None:
+        env = {**os.environ, **env}
+    return subprocess.run(cmd, shell=True, env=env).returncode

commitizen/hooks.py

@@ -17,14 +17,9 @@ def run(hooks: str | list[str], _env_prefix: str = "CZ_", **env: object) -> None
     for hook in hooks:
         out.info(f"Running hook '{hook}'")
 
-        c = cmd.run(hook, env=_format_env(_env_prefix, env))
+        return_code = cmd.run_interactive(hook, env=_format_env(_env_prefix, env))
 
-        if c.out:
-            out.write(c.out)
-        if c.err:
-            out.error(c.err)
-
-        if c.return_code != 0:
+        if return_code != 0:
             raise RunHookError(f"Running hook '{hook}' failed")
 
 

tests/test_bump_hooks.py

@@ -12,8 +12,8 @@ def test_run(mocker: MockFixture):
     bump_hooks = ["pre_bump_hook", "pre_bump_hook_1"]
 
     cmd_run_mock = mocker.Mock()
-    cmd_run_mock.return_value.return_code = 0
-    mocker.patch.object(cmd, "run", cmd_run_mock)
+    cmd_run_mock.return_value = 0
+    mocker.patch.object(cmd, "run_interactive", cmd_run_mock)
 
     hooks.run(bump_hooks)
 
@@ -29,8 +29,8 @@ def test_run_error(mocker: MockFixture):
     bump_hooks = ["pre_bump_hook", "pre_bump_hook_1"]
 
     cmd_run_mock = mocker.Mock()
-    cmd_run_mock.return_value.return_code = 1
-    mocker.patch.object(cmd, "run", cmd_run_mock)
+    cmd_run_mock.return_value = 1
+    mocker.patch.object(cmd, "run_interactive", cmd_run_mock)
 
     with pytest.raises(RunHookError):
         hooks.run(bump_hooks)

tests/test_cmd.py

@@ -55,3 +55,78 @@ def decode(self, encoding="utf-8", errors="strict"):
 
     with pytest.raises(CharacterSetDecodeError):
         cmd._try_decode(_bytes())
+
+
+class TestRun:
+    def test_stdout_captured(self):
+        result = cmd.run("python -c \"print('hello')\"")
+        assert "hello" in result.out
+        assert isinstance(result.stdout, bytes)
+        assert b"hello" in result.stdout
+
+    def test_stderr_captured(self):
+        result = cmd.run("python -c \"import sys; print('err msg', file=sys.stderr)\"")
+        assert "err msg" in result.err
+        assert isinstance(result.stderr, bytes)
+        assert b"err msg" in result.stderr
+
+    def test_zero_return_code_on_success(self):
+        result = cmd.run('python -c "import sys; sys.exit(0)"')
+        assert result.return_code == 0
+
+    def test_nonzero_return_code_on_failure(self):
+        result = cmd.run('python -c "import sys; sys.exit(42)"')
+        assert result.return_code == 42
+
+    def test_env_passed_to_subprocess(self):
+        result = cmd.run(
+            "python -c \"import os; print(os.environ['CZ_TEST_VAR'])\"",
+            env={"CZ_TEST_VAR": "sentinelvalue"},
+        )
+        assert "sentinelvalue" in result.out
+        assert result.return_code == 0
+
+    def test_env_merged_with_os_environ(self, monkeypatch):
+        monkeypatch.setenv("CZ_EXISTING_VAR", "fromenv")
+        result = cmd.run(
+            "python -c \"import os; print(os.environ['CZ_EXISTING_VAR'])\"",
+            env={"CZ_EXTRA_VAR": "extra"},
+        )
+        assert "fromenv" in result.out
+
+    def test_empty_stdout_and_stderr(self):
+        result = cmd.run('python -c "pass"')
+        assert result.out == ""
+        assert result.err == ""
+        assert result.stdout == b""
+        assert result.stderr == b""
+
+    def test_no_env_uses_os_environ(self, monkeypatch):
+        monkeypatch.setenv("CZ_NO_ENV_TEST", "inherited")
+        result = cmd.run("python -c \"import os; print(os.environ['CZ_NO_ENV_TEST'])\"")
+        assert "inherited" in result.out
+
+
+class TestRunInteractive:
+    def test_zero_return_code_on_success(self):
+        return_code = cmd.run_interactive('python -c "import sys; sys.exit(0)"')
+        assert return_code == 0
+
+    def test_nonzero_return_code_on_failure(self):
+        return_code = cmd.run_interactive('python -c "import sys; sys.exit(3)"')
+        assert return_code == 3
+
+    def test_env_passed_to_subprocess(self):
+        return_code = cmd.run_interactive(
+            "python -c \"import os, sys; sys.exit(0 if os.environ['CZ_ITEST_VAR'] == 'val' else 1)\"",
+            env={"CZ_ITEST_VAR": "val"},
+        )
+        assert return_code == 0
+
+    def test_env_merged_with_os_environ(self, monkeypatch):
+        monkeypatch.setenv("CZ_ITEST_EXISTING", "yes")
+        return_code = cmd.run_interactive(
+            "python -c \"import os, sys; sys.exit(0 if os.environ['CZ_ITEST_EXISTING'] == 'yes' else 1)\"",
+            env={"CZ_ITEST_EXTRA": "extra"},
+        )
+        assert return_code == 0