tools: fix CI git diff path parsing

5 files changed, 348 insertions, 32 deletions

diff --git a/tools/ci/compile_bsp_with_drivers.py b/tools/ci/compile_bsp_with_drivers.py
index 0ddc7c58e6..fcceea4b31 100644
--- a/tools/ci/compile_bsp_with_drivers.py
+++ b/tools/ci/compile_bsp_with_drivers.py
@@ -8,11 +8,15 @@
 # 2023-06-27     dejavudwh    the first version
 #
 
-import subprocess
 import logging
 import os
 import sys
 
+try:
+    from git_utils import get_changed_files
+except ImportError:
+    from tools.ci.git_utils import get_changed_files
+
 CONFIG_BSP_USING_X = ["CONFIG_BSP_USING_UART", "CONFIG_BSP_USING_I2C", "CONFIG_BSP_USING_SPI", "CONFIG_BSP_USING_ADC", "CONFIG_BSP_USING_DAC"]
 
 def init_logger():
@@ -24,25 +28,35 @@ def init_logger():
                         )
 
 def diff():
-    result = subprocess.run(['git', 'diff', '--name-only', 'HEAD', 'origin/master', '--diff-filter=ACMR', '--no-renames', '--full-index'], stdout = subprocess.PIPE)
-    file_list = result.stdout.decode().strip().split('\n')
+    try:
+        file_list = get_changed_files(
+            "HEAD",
+            "origin/master",
+            diff_filter="ACMR",
+            no_renames=True,
+            full_index=True,
+        )
+    except RuntimeError as e:
+        logging.error(e)
+        return set()
+
     logging.info(file_list)
     bsp_paths = set()
     for file in file_list:
         if "bsp/" in file:
             logging.info("Modifed file: {}".format(file))
             bsp_paths.add(file)
-    
+
     dirs = set()
     for dir in bsp_paths:
-        dir = os.path.dirname(dir)    
+        dir = os.path.dirname(dir)
         while "bsp/" in dir:
             files = os.listdir(dir)
             if ".config" in files and "rt-thread.elf" not in files and not dir.endswith("bsp"):
                 logging.info("Found bsp path: {}".format(dir))
                 dirs.add(dir)
                 break
-            new_dir = os.path.dirname(dir)    
+            new_dir = os.path.dirname(dir)
             dir = new_dir
 
     return dirs
diff --git a/tools/ci/format_ignore.py b/tools/ci/format_ignore.py
index a7f9be2cbf..2bb5f8e146 100644
--- a/tools/ci/format_ignore.py
+++ b/tools/ci/format_ignore.py
@@ -11,7 +11,11 @@
 import yaml
 import logging
 import os
-import subprocess
+
+try:
+    from git_utils import get_changed_files
+except ImportError:
+    from tools.ci.git_utils import get_changed_files
 
 def init_logger():
     log_format = "[%(filename)s %(lineno)d %(levelname)s] %(message)s "
@@ -72,13 +76,23 @@ class CheckOut:
         return 1
     
     def get_new_file(self):
-        result = subprocess.run(['git', 'diff', '--name-only', 'HEAD', 'origin/master', '--diff-filter=ACMR', '--no-renames', '--full-index'], stdout = subprocess.PIPE)
-        file_list = result.stdout.decode().strip().split('\n')
+        try:
+            file_list = get_changed_files(
+                "HEAD",
+                "origin/master",
+                diff_filter="ACMR",
+                no_renames=True,
+                full_index=True,
+            )
+        except RuntimeError as e:
+            logging.error(e)
+            return []
+
         new_files = []
         for line in file_list:
             logging.info("modified file -> {}".format(line))
             result = self.__exclude_file(line)
             if result != 0:
                 new_files.append(line)
-        
-        return new_files
\ No newline at end of file
+
+        return new_files
diff --git a/tools/ci/git_diff_show.py b/tools/ci/git_diff_show.py
index 435823c702..2b5aacc265 100644
--- a/tools/ci/git_diff_show.py
+++ b/tools/ci/git_diff_show.py
@@ -17,6 +17,16 @@ from typing import List, Dict
 
 import json
 from typing import List
+
+try:
+    from git_utils import get_blob_size
+    from git_utils import get_merge_base as git_get_merge_base
+    from git_utils import get_name_status_lines
+except ImportError:
+    from tools.ci.git_utils import get_blob_size
+    from tools.ci.git_utils import get_merge_base as git_get_merge_base
+    from tools.ci.git_utils import get_name_status_lines
+
 class FileDiff:
     def __init__(self, path: str, status: str, size_change: int = 0, old_size: int = 0, new_size: int = 0):
         self.path = path
@@ -49,14 +59,13 @@ class GitDiffAnalyzer:
             sys.exit(1)
             
         # 获取差异文件列表
-        diff_cmd = f"git diff --name-status {merge_base} HEAD"
-        print(diff_cmd)
+        print("git diff --name-status {} HEAD".format(merge_base))
         try:
-            output = subprocess.check_output(diff_cmd.split(), stderr=subprocess.STDOUT)
-            output = output.decode(self.encoding).strip()
+            lines = get_name_status_lines(merge_base, "HEAD", self.encoding)
+            output = "\n".join(lines)
             print(output)
-        except subprocess.CalledProcessError as e:
-            print(f"Error executing git diff: {e.output.decode(self.encoding)}")
+        except RuntimeError as e:
+            print("Error executing git diff: {}".format(e))
             sys.exit(1)
             
         if not output:
@@ -110,24 +119,15 @@ class GitDiffAnalyzer:
     def get_merge_base(self) -> str:
         """获取当前分支和目标分支的最近共同祖先"""
         try:
-            cmd = f"git merge-base {self.target_branch} HEAD"
-            print(cmd)
-            output = subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
-            return output.decode(self.encoding).strip()
-        except subprocess.CalledProcessError as e:
-            print(f"Error executing git merge-base: {e.output.decode(self.encoding)}")
+            print("git merge-base {} HEAD".format(self.target_branch))
+            return git_get_merge_base(self.target_branch, "HEAD", self.encoding)
+        except RuntimeError as e:
+            print("Error executing git merge-base: {}".format(e))
             return None
 
     def get_file_size(self, path: str, ref: str) -> int:
-        """获取指定分支上文件的大小"""
-        try:
-            # 使用 git cat-file 来获取文件内容，然后计算其大小
-            cmd = f"git cat-file blob {ref}:{path}"
-            output = subprocess.check_output(cmd.split(), stderr=subprocess.STDOUT)
-            return len(output)
-        except subprocess.CalledProcessError:
-            # 如果文件不存在或无法获取，返回0
-            return 0
+        """Get the size of a file at the specified git ref."""
+        return get_blob_size(ref, path, self.encoding)
 
 def format_size(size: int) -> str:
     """将字节大小转换为人类可读的格式"""
diff --git a/tools/ci/git_utils.py b/tools/ci/git_utils.py
new file mode 100644
index 0000000000..cf1f8d4b4d
--- /dev/null
+++ b/tools/ci/git_utils.py
@@ -0,0 +1,112 @@
+#
+# Copyright (c) 2006-2026, RT-Thread Development Team
+#
+# SPDX-License-Identifier: Apache-2.0
+#
+
+import locale
+import subprocess
+
+
+def _default_encoding():
+    return locale.getpreferredencoding()
+
+
+def _decode_output(data, encoding=None):
+    if data is None:
+        return ""
+
+    if encoding is None:
+        encoding = _default_encoding()
+
+    return data.decode(encoding, errors="replace")
+
+
+def _run_git(args, encoding=None):
+    result = subprocess.run(
+        ["git"] + args,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=False,
+    )
+    stdout = _decode_output(getattr(result, "stdout", b""), encoding)
+    stderr = _decode_output(getattr(result, "stderr", b""), encoding)
+    returncode = getattr(result, "returncode", 0)
+
+    if returncode != 0:
+        raise RuntimeError(
+            "git {} failed with code {}: {}".format(
+                " ".join(args),
+                returncode,
+                stderr.strip(),
+            )
+        )
+
+    return stdout
+
+
+def split_nonempty_lines(output):
+    return [line.strip() for line in output.splitlines() if line.strip()]
+
+
+def get_changed_files(base, head, diff_filter="ACMR", no_renames=True, full_index=True, encoding=None):
+    args = [
+        "diff",
+        "--name-only",
+        base,
+        head,
+        "--diff-filter={}".format(diff_filter),
+    ]
+
+    if no_renames:
+        args.append("--no-renames")
+    if full_index:
+        args.append("--full-index")
+
+    output = _run_git(args, encoding=encoding)
+    return split_nonempty_lines(output)
+
+
+def get_name_status_lines(base, head="HEAD", encoding=None):
+    output = _run_git(
+        ["diff", "--name-status", base, head],
+        encoding=encoding,
+    )
+    return split_nonempty_lines(output)
+
+
+def get_merge_base(target_branch, head="HEAD", encoding=None):
+    output = _run_git(
+        ["merge-base", target_branch, head],
+        encoding=encoding,
+    )
+    return output.strip()
+
+
+def _parse_ls_tree_size(output, encoding=None):
+    text = _decode_output(output, encoding).strip()
+    if not text:
+        return 0
+
+    fields = text.split(None, 4)
+    if len(fields) < 4:
+        return 0
+
+    try:
+        return int(fields[3])
+    except ValueError:
+        return 0
+
+
+def get_blob_size(ref, path, encoding=None):
+    result = subprocess.run(
+        ["git", "ls-tree", "-l", ref, "--", path],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=False,
+    )
+
+    if result.returncode != 0:
+        return 0
+
+    return _parse_ls_tree_size(result.stdout, encoding)
diff --git a/tools/testcases/test_ci_diff_paths.py b/tools/testcases/test_ci_diff_paths.py
new file mode 100644
index 0000000000..52bed39bba
--- /dev/null
+++ b/tools/testcases/test_ci_diff_paths.py
@@ -0,0 +1,176 @@
+import importlib.util
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+
+def load_module(module_name, path):
+    path = Path(path)
+    repo_root = path.parents[2]
+    module_dir = path.parent
+
+    for item in (str(repo_root), str(module_dir)):
+        if item not in sys.path:
+            sys.path.insert(0, item)
+
+    spec = importlib.util.spec_from_file_location(module_name, path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class CIDiffPathParsingTestCase(unittest.TestCase):
+    def setUp(self):
+        self.repo_root = Path(__file__).resolve().parents[2]
+        self.old_cwd = os.getcwd()
+
+    def tearDown(self):
+        os.chdir(self.old_cwd)
+
+    def test_git_utils_splits_empty_diff_output(self):
+        module = load_module(
+            "git_utils",
+            self.repo_root / "tools" / "ci" / "git_utils.py",
+        )
+
+        self.assertEqual(module.split_nonempty_lines(""), [])
+        self.assertEqual(module.split_nonempty_lines("\n\r\n"), [])
+        self.assertEqual(module.split_nonempty_lines("a.c\n\n b.c \n"), ["a.c", "b.c"])
+
+    def test_format_ignore_returns_empty_list_for_empty_diff(self):
+        module = load_module(
+            "format_ignore",
+            self.repo_root / "tools" / "ci" / "format_ignore.py",
+        )
+
+        if hasattr(module, "get_changed_files"):
+            patcher = mock.patch.object(module, "get_changed_files", return_value=[])
+        else:
+            class FakeResult:
+                returncode = 0
+                stdout = b""
+                stderr = b""
+
+            patcher = mock.patch.object(module.subprocess, "run", return_value=FakeResult())
+
+        with patcher:
+            files = module.CheckOut().get_new_file()
+
+        self.assertEqual(files, [])
+
+    def test_compile_bsp_diff_returns_empty_set_for_empty_diff(self):
+        module = load_module(
+            "compile_bsp_with_drivers",
+            self.repo_root / "tools" / "ci" / "compile_bsp_with_drivers.py",
+        )
+
+        if hasattr(module, "get_changed_files"):
+            patcher = mock.patch.object(module, "get_changed_files", return_value=[])
+        else:
+            class FakeResult:
+                returncode = 0
+                stdout = b""
+                stderr = b""
+
+            patcher = mock.patch.object(module.subprocess, "run", return_value=FakeResult())
+
+        with patcher:
+            dirs = module.diff()
+
+        self.assertEqual(dirs, set())
+
+    def test_git_diff_show_handles_paths_with_spaces(self):
+        if shutil.which("git") is None:
+            self.skipTest("git is not available")
+
+        module = load_module(
+            "git_diff_show",
+            self.repo_root / "tools" / "ci" / "git_diff_show.py",
+        )
+
+        tempdir = tempfile.mkdtemp()
+        workdir = Path(tempdir)
+
+        try:
+            def run(cmd):
+                result = subprocess.run(
+                    cmd,
+                    cwd=str(workdir),
+                    text=True,
+                    stdout=subprocess.PIPE,
+                    stderr=subprocess.PIPE,
+                    check=False,
+                )
+                self.assertEqual(
+                    result.returncode,
+                    0,
+                    msg="command failed: {}\nstdout:\n{}\nstderr:\n{}".format(
+                        " ".join(cmd),
+                        result.stdout,
+                        result.stderr,
+                    ),
+                )
+                return result
+
+            run(["git", "init"])
+            run(["git", "checkout", "-b", "master"])
+
+            space_dir = workdir / "dir with space"
+            space_dir.mkdir()
+            file_path = space_dir / "file with space.txt"
+            file_path.write_text("old\n", encoding="utf-8")
+
+            run(["git", "add", "."])
+            run(
+                [
+                    "git",
+                    "-c",
+                    "user.name=Test",
+                    "-c",
+                    "[email protected]",
+                    "commit",
+                    "-m",
+                    "base",
+                ]
+            )
+
+            run(["git", "checkout", "-b", "feature"])
+            file_path.write_text("old\nnew content\n", encoding="utf-8")
+            run(["git", "add", "."])
+            run(
+                [
+                    "git",
+                    "-c",
+                    "user.name=Test",
+                    "-c",
+                    "[email protected]",
+                    "commit",
+                    "-m",
+                    "modify space path",
+                ]
+            )
+
+            os.chdir(workdir)
+            diffs = module.GitDiffAnalyzer("master").get_diff_files()
+        finally:
+            os.chdir(self.old_cwd)
+            shutil.rmtree(workdir, ignore_errors=True)
+
+        self.assertEqual(len(diffs), 1)
+        self.assertEqual(diffs[0].path, "dir with space/file with space.txt")
+        self.assertEqual(diffs[0].status, "M")
+        self.assertEqual(diffs[0].old_size, len("old\n".encode("utf-8")))
+        self.assertEqual(diffs[0].new_size, len("old\nnew content\n".encode("utf-8")))
+        self.assertEqual(
+            diffs[0].size_change,
+            len("old\nnew content\n".encode("utf-8")) - len("old\n".encode("utf-8")),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()