mirror of
https://github.com/calesthio/OpenMontage.git
synced 2026-08-17 22:08:48 +08:00
fix: preserve subprocess error type in tool runner
- raise a CalledProcessError subclass that surfaces stderr/stdout detail - keep existing callers that catch subprocess.CalledProcessError working - reject lavfi movie paths containing single quotes fail-closed - add regression coverage for both review findings Verification: - python3 -m unittest tests.tools.test_base_tool_dependencies tests.tools.test_scene_detect_lavfi_escape tests.tools.test_cost_tracker_governance - python3 -m py_compile tools/base_tool.py tools/analysis/scene_detect.py tools/cost_tracker.py tests/tools/test_base_tool_dependencies.py tests/tools/test_scene_detect_lavfi_escape.py tests/tools/test_cost_tracker_governance.py
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
@@ -33,16 +34,19 @@ class BinaryDependencyTests(unittest.TestCase):
|
||||
|
||||
with patch("tools.base_tool.shutil.which", return_value="/usr/bin/ffmpeg"):
|
||||
tool.check_dependencies()
|
||||
def test_run_command_error_includes_stderr(self) -> None:
|
||||
def test_run_command_error_preserves_called_process_error_type(self) -> None:
|
||||
tool = DummyTool()
|
||||
|
||||
with self.assertRaisesRegex(RuntimeError, "specific stderr"):
|
||||
with self.assertRaises(subprocess.CalledProcessError) as ctx:
|
||||
tool.run_command([
|
||||
sys.executable,
|
||||
"-c",
|
||||
"import sys; print('specific stderr', file=sys.stderr); sys.exit(7)",
|
||||
])
|
||||
|
||||
self.assertEqual(ctx.exception.returncode, 7)
|
||||
self.assertIn("specific stderr", str(ctx.exception))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -12,17 +12,20 @@ from tools.analysis.scene_detect import SceneDetect
|
||||
|
||||
class SceneDetectEscapingTests(unittest.TestCase):
|
||||
def test_lavfi_movie_path_escapes_filtergraph_metacharacters(self) -> None:
|
||||
raw = "/tmp/clip'name,with[bad];chars:01.mov"
|
||||
raw = "/tmp/clip-name,with[bad];chars:01.mov"
|
||||
|
||||
escaped = SceneDetect._escape_lavfi_movie_path(raw)
|
||||
|
||||
self.assertIn("\\'", escaped)
|
||||
self.assertIn("\\,", escaped)
|
||||
self.assertIn("\\[", escaped)
|
||||
self.assertIn("\\]", escaped)
|
||||
self.assertIn("\\;", escaped)
|
||||
self.assertIn("\\:", escaped)
|
||||
self.assertNotIn("clip'name,with[bad];chars:01", escaped)
|
||||
self.assertNotIn("clip-name,with[bad];chars:01", escaped)
|
||||
|
||||
def test_lavfi_movie_path_rejects_single_quote_fail_closed(self) -> None:
|
||||
with self.assertRaisesRegex(ValueError, "single quotes"):
|
||||
SceneDetect._escape_lavfi_movie_path("/tmp/clip'name.mov")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -168,9 +168,11 @@ class SceneDetect(BaseTool):
|
||||
def _escape_lavfi_movie_path(path: str) -> str:
|
||||
"""Escape a path for FFmpeg lavfi movie=... without allowing filter injection."""
|
||||
normalized = path.replace("\\", "/")
|
||||
if "'" in normalized:
|
||||
raise ValueError("FFmpeg lavfi movie paths containing single quotes are unsupported")
|
||||
escaped = []
|
||||
for char in normalized:
|
||||
if char in "\\':,[];":
|
||||
if char in "\\:,[];":
|
||||
escaped.append("\\" + char)
|
||||
else:
|
||||
escaped.append(char)
|
||||
|
||||
@@ -349,11 +349,37 @@ class BaseTool(ABC):
|
||||
stderr = (exc.stderr or "").strip()
|
||||
stdout = (exc.stdout or "").strip()
|
||||
detail = stderr or stdout or str(exc)
|
||||
raise RuntimeError(
|
||||
f"Command failed with exit code {exc.returncode}: {' '.join(resolved_cmd)}\n{detail}"
|
||||
raise ToolCommandError(
|
||||
exc.returncode,
|
||||
exc.cmd,
|
||||
output=exc.output,
|
||||
stderr=exc.stderr,
|
||||
detail=detail,
|
||||
) from exc
|
||||
|
||||
|
||||
class ToolCommandError(subprocess.CalledProcessError):
|
||||
"""CalledProcessError with stderr/stdout surfaced in str(error)."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
returncode: int,
|
||||
cmd: list[str],
|
||||
*,
|
||||
output: Optional[str] = None,
|
||||
stderr: Optional[str] = None,
|
||||
detail: str = "",
|
||||
) -> None:
|
||||
super().__init__(returncode, cmd, output=output, stderr=stderr)
|
||||
self.detail = detail
|
||||
|
||||
def __str__(self) -> str:
|
||||
base = super().__str__()
|
||||
if self.detail:
|
||||
return f"{base}\n{self.detail}"
|
||||
return base
|
||||
|
||||
|
||||
class DependencyError(Exception):
|
||||
"""Raised when a tool's dependency is not satisfied."""
|
||||
pass
|
||||
|
||||
Reference in New Issue
Block a user