mytype

Author: joyguoguo

fuzz/command_util.py

@@ -1,24 +1,24 @@
-# fuzz/command_util.py
+# command_util.py
 
 import subprocess
-import time
 import logging
+import time
 from pathlib import Path
-from typing import Optional, cast
+from typing import Optional, Tuple, List
 from returns.maybe import Maybe
 from errors import CommandError
 
 def _run_subprocess(
     cmd: str,
-    cwd: Optional[Path] = None,
+    cwd: Optional[Path] = None,  # 修复：添加 Optional 类型
     capture_output: bool = False,
-    timeout: Optional[int] = None,
-    logger: Optional[logging.Logger] = None,
-) -> tuple[int, list[str]]:
+    timeout: Optional[int] = None,  # 修复：添加 Optional 类型
+    logger: Optional[logging.Logger] = None,  # 修复：添加 Optional 类型
+) -> Tuple[int, List[str]]:  # 建议使用 Tuple 替代 tuple
     """
     低层执行子进程命令
     - capture_output=True：捕获 stdout，返回输出列表
-    - timeout 秒超时（无超时则None）
+    - timeout 秒超时（无超时则 None）
     - logger 用于实时打印输出
     返回：(退出码, 输出行列表)
     """
@@ -33,28 +33,29 @@ def _run_subprocess(
         errors="replace",
     )
 
-    output_lines = []
+    output_lines: List[str] = []
     start_time = time.time()
 
     try:
         if capture_output:
-            # 类型断言确保 stdout 不为 None
-            stdout = cast(Optional[subprocess.PIPE], process.stdout)
-            if stdout is None:
-                raise RuntimeError("Stdout should not be None when capture_output is True")
+            # 确保 stdout 不是 None
+            if process.stdout is None:
+                raise RuntimeError("stdout is unexpectedly None in capture mode")
 
             while True:
-                line = stdout.readline()
+                line = process.stdout.readline()
                 if line:
-                    output_lines.append(line.rstrip())
+                    line_stripped = line.rstrip()
+                    output_lines.append(line_stripped)
                     if logger:
-                        logger.debug(line.rstrip())
+                        logger.debug(line_stripped)
                 elif process.poll() is not None:
                     break
 
-                if timeout is not None and (time.time() - start_time) > timeout:
+                # 处理超时逻辑
+                if timeout and (time.time() - start_time) > timeout:
                     if logger:
-                        logger.error(f"⌛ Command timed out after {timeout} seconds")
+                        logger.error(f"⌛ 命令超时，耗时 {timeout} 秒")
                     process.terminate()
                     try:
                         process.wait(timeout=5)
@@ -68,9 +69,9 @@ def _run_subprocess(
 
     except Exception as e:
         if logger:
-            logger.exception(f"Error during command execution: {e}")
+            logger.exception(f"执行命令时出错：{e}")
         process.kill()
-        raise e
+        raise CommandError(f"Command failed: {str(e)}") from e
 
     return process.returncode, output_lines
 
@@ -79,15 +80,15 @@ def run_command_build_fuzz(
     cmd: str,
     oss_fuzz_dir: Path,
     project: str = "",
-    allowed_exit_codes: Maybe[list[int]] = Maybe.empty(),
+    allowed_exit_codes: Maybe[List[int]] = Maybe.empty,
     skip_yes: bool = False
 ) -> int:
     """build_fuzz.py 中使用的 run_command，简化版，抛异常"""
     allowed_codes = allowed_exit_codes.value_or([0])
     cmd_str = f"yes | {cmd}" if not skip_yes else cmd
     exit_code, _ = _run_subprocess(cmd_str, cwd=oss_fuzz_dir)
     if exit_code not in allowed_codes:
-        error_msg = f"Command failed (exit code: {exit_code})"
+        error_msg = f"命令失败（退出码：{exit_code}）"
         if project:
             error_msg += f" for project: {project}"
         raise CommandError(error_msg, project=project, exit_code=exit_code)
@@ -98,16 +99,22 @@ def run_command_fuzz_all_targets(
     cmd: str,
     log_msg: str,
     logger: logging.Logger,
-    allowed_exit_codes: Maybe[list[int]] = Maybe.empty(),
+    allowed_exit_codes: Maybe[List[int]] = Maybe.empty,
     timeout: int = 3600,
 ) -> bool:
-    """run_fuzz_all_targets_print1.py 中使用，带实时日志与超时，返回bool"""
+    """run_fuzz_all_targets_print1.py 中使用，带实时日志与超时，返回 bool"""
     logger.info(f"▶️ {log_msg}...")
     logger.debug(f"   $ {cmd}")
 
     allowed_codes = allowed_exit_codes.value_or([])
-    exit_code, _ = _run_subprocess(cmd, capture_output=True, timeout=timeout, logger=logger)
+    exit_code, _ = _run_subprocess(
+        cmd, 
+        capture_output=True, 
+        timeout=timeout, 
+        logger=logger
+    )
+    
     if exit_code not in [0, *allowed_codes]:
-        logger.error(f"❌ Command execution failed, exit code: {exit_code}")
+        logger.error(f"❌ 命令执行失败，退出码：{exit_code}")
         return False
     return True