diff --git a/tests/tools/test_base_tool_dependencies.py b/tests/tools/test_base_tool_dependencies.py new file mode 100644 index 00000000..79505a2a --- /dev/null +++ b/tests/tools/test_base_tool_dependencies.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import subprocess +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT)) + +from tools.base_tool import BaseTool, DependencyError, ToolResult + + +class DummyTool(BaseTool): + def execute(self, inputs: dict) -> ToolResult: + return ToolResult(success=True) + + +class BinaryDependencyTests(unittest.TestCase): + def test_binary_dependency_prefix_is_checked_like_cmd(self) -> None: + tool = DummyTool() + tool.dependencies = ["binary:definitely-not-installed-openmontage-test"] + tool.install_instructions = "install it" + + with patch("tools.base_tool.shutil.which", return_value=None): + with self.assertRaises(DependencyError): + tool.check_dependencies() + + def test_binary_dependency_prefix_accepts_available_command(self) -> None: + tool = DummyTool() + tool.dependencies = ["binary:ffmpeg"] + tool.install_instructions = "install ffmpeg" + + with patch("tools.base_tool.shutil.which", return_value="/usr/bin/ffmpeg"): + tool.check_dependencies() + def test_run_command_error_preserves_called_process_error_type(self) -> None: + tool = DummyTool() + + 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() diff --git a/tests/tools/test_cost_tracker_governance.py b/tests/tools/test_cost_tracker_governance.py new file mode 100644 index 00000000..e97e4783 --- /dev/null +++ b/tests/tools/test_cost_tracker_governance.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import json +import sys +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT)) + +from lib.config_model import BudgetMode +from tools.cost_tracker import CostTracker + + +class CostTrackerGovernanceTests(unittest.TestCase): + def test_warn_mode_marks_over_budget_reservation(self) -> None: + with self.subTest("warning is recorded and persisted"): + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as temp_dir: + log_path = Path(temp_dir) / "cost_log.json" + tracker = CostTracker( + budget_total_usd=1.0, + reserve_pct=0.0, + single_action_approval_usd=99.0, + require_approval_for_new_paid_tool=False, + mode=BudgetMode.WARN, + cost_log_path=log_path, + ) + entry_id = tracker.estimate("paid_video", "generate", 2.0) + + tracker.reserve(entry_id) + + entry = tracker.entries[0] + self.assertEqual(entry["status"], "reserved") + self.assertEqual(entry["reserved_usd"], 2.0) + self.assertTrue(entry["budget_warning"]) + self.assertIn("exceeds usable budget", entry["budget_warning_message"]) + persisted = json.loads(log_path.read_text()) + self.assertTrue(persisted["entries"][0]["budget_warning"]) + + def test_approved_tools_persist_across_tracker_restarts(self) -> None: + import tempfile + from pathlib import Path + + with tempfile.TemporaryDirectory() as temp_dir: + log_path = Path(temp_dir) / "cost_log.json" + tracker = CostTracker(cost_log_path=log_path) + tracker.approve_tool("paid_video") + + restarted = CostTracker(cost_log_path=log_path) + + entry_id = restarted.estimate("paid_video", "generate", 0.01) + restarted.reserve(entry_id) + self.assertEqual(restarted.entries[-1]["status"], "reserved") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/tools/test_scene_detect_lavfi_escape.py b/tests/tools/test_scene_detect_lavfi_escape.py new file mode 100644 index 00000000..717b1611 --- /dev/null +++ b/tests/tools/test_scene_detect_lavfi_escape.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import sys +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(ROOT)) + +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" + + escaped = SceneDetect._escape_lavfi_movie_path(raw) + + self.assertIn("\\,", escaped) + self.assertIn("\\[", escaped) + self.assertIn("\\]", escaped) + self.assertIn("\\;", escaped) + self.assertIn("\\:", 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__": + unittest.main() diff --git a/tools/analysis/scene_detect.py b/tools/analysis/scene_detect.py index 71062a4f..8ca2ecb5 100644 --- a/tools/analysis/scene_detect.py +++ b/tools/analysis/scene_detect.py @@ -164,11 +164,26 @@ class SceneDetect(BaseTool): return scenes + @staticmethod + 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 "\\:,[];": + escaped.append("\\" + char) + else: + escaped.append(char) + return "".join(escaped) + def _detect_ffmpeg(self, inputs: dict[str, Any]) -> list[dict]: """Fallback: use FFmpeg scene change filter.""" input_path = str(inputs["input_path"]) threshold = inputs.get("threshold", 0.3) min_scene_len = inputs.get("min_scene_length_seconds", 1.0) + escaped_input = self._escape_lavfi_movie_path(input_path) cmd = [ "ffprobe", @@ -176,7 +191,7 @@ class SceneDetect(BaseTool): "-show_entries", "frame=pts_time", "-of", "json", "-f", "lavfi", - f"movie='{input_path.replace(chr(92), '/').replace(':', chr(92)+':')}',select='gt(scene,{threshold})'", + f"movie='{escaped_input}',select='gt(scene,{threshold})'", ] try: diff --git a/tools/base_tool.py b/tools/base_tool.py index 1194e97d..15d24e6c 100644 --- a/tools/base_tool.py +++ b/tools/base_tool.py @@ -304,8 +304,9 @@ class BaseTool(ABC): def check_dependencies(self) -> None: """Verify all dependencies are installed. Raises DependencyError if not.""" for dep in self.dependencies: - if dep.startswith("cmd:"): - cmd_name = dep[4:] + if dep.startswith(("cmd:", "binary:")): + prefix = "cmd:" if dep.startswith("cmd:") else "binary:" + cmd_name = dep[len(prefix):] if shutil.which(cmd_name) is None: raise DependencyError( f"Command {cmd_name!r} not found. {self.install_instructions}" @@ -424,20 +425,54 @@ class BaseTool(ABC): exe = shutil.which(resolved_cmd[0]) if exe: resolved_cmd[0] = exe - return subprocess.run( - resolved_cmd, - capture_output=True, - text=True, - # Force UTF-8 decoding. The default uses the OS locale (cp1252 on - # Windows), which raises UnicodeDecodeError on a subprocess that - # emits Unicode/emoji (e.g. Remotion's progress output), killing the - # reader thread and potentially swallowing the real error text. - encoding="utf-8", - errors="replace", - timeout=timeout, - cwd=cwd, - check=True, - ) + try: + return subprocess.run( + resolved_cmd, + capture_output=True, + text=True, + # Force UTF-8 decoding. The default uses the OS locale (cp1252 on + # Windows), which raises UnicodeDecodeError on a subprocess that + # emits Unicode/emoji (e.g. Remotion's progress output), killing the + # reader thread and potentially swallowing the real error text. + encoding="utf-8", + errors="replace", + timeout=timeout, + cwd=cwd, + check=True, + ) + except subprocess.CalledProcessError as exc: + stderr = (exc.stderr or "").strip() + stdout = (exc.stdout or "").strip() + detail = stderr or stdout or str(exc) + 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): diff --git a/tools/cost_tracker.py b/tools/cost_tracker.py index ee324efb..8c363d7d 100644 --- a/tools/cost_tracker.py +++ b/tools/cost_tracker.py @@ -141,11 +141,15 @@ class CostTracker: # Check budget if estimated > self.usable_budget_usd: + message = ( + f"Reservation of ${estimated:.2f} exceeds usable budget " + f"${self.usable_budget_usd:.2f}" + ) if self.mode == BudgetMode.CAP: - raise BudgetExceededError( - f"Reservation of ${estimated:.2f} exceeds usable budget " - f"${self.usable_budget_usd:.2f}" - ) + raise BudgetExceededError(message) + if self.mode == BudgetMode.WARN: + entry["budget_warning"] = True + entry["budget_warning_message"] = message entry["status"] = EntryStatus.RESERVED.value entry["reserved_usd"] = estimated @@ -155,6 +159,7 @@ class CostTracker: def approve_tool(self, tool: str) -> None: """Mark a tool as approved for paid operations.""" self._approved_tools.add(tool) + self._save() def reconcile(self, entry_id: str, actual_usd: float, success: bool = True) -> None: """Reconcile actual spend after tool execution.""" @@ -487,6 +492,7 @@ class CostTracker: "budget_total_usd": self.budget_total_usd, "budget_reserved_usd": round(self.budget_reserved_usd, 4), "budget_spent_usd": round(self.budget_spent_usd, 4), + "approved_tools": sorted(self._approved_tools), "entries": self.entries, } self.cost_log_path.parent.mkdir(parents=True, exist_ok=True) @@ -498,6 +504,7 @@ class CostTracker: data = json.load(f) self.entries = data.get("entries", []) self.budget_total_usd = data.get("budget_total_usd", self.budget_total_usd) + self._approved_tools = set(data.get("approved_tools", [])) # ---- Helpers ----