diff --git a/src/devops_learn/tools/python_tool.py b/src/devops_learn/tools/python_tool.py index 455e623..61496ce 100644 --- a/src/devops_learn/tools/python_tool.py +++ b/src/devops_learn/tools/python_tool.py @@ -48,7 +48,7 @@ def _python_executable() -> str: def _run( - command: list[str], *, cwd: str | None, check: bool = False + command: list[str], *, cwd: str | None, check: bool = False, timeout: float = 60.0 ) -> subprocess.CompletedProcess[str]: environment = os.environ.copy() # User-installed pytest plugins are outside a scanned project's dependency @@ -60,6 +60,7 @@ def _run( capture_output=True, text=True, check=check, + timeout=timeout, env=environment, ) @@ -137,29 +138,45 @@ def execute( ) cwd: str | None = params.get("path") + timeout = float(params.get("timeout", 60.0)) try: if operation == "run_tests": tests_path = (Path(cwd or ".").resolve() / "tests") command = [_python_executable(), "-m", "pytest", str(tests_path)] - completed = _run(command, cwd=cwd) + completed = _run(command, cwd=cwd, timeout=timeout) success = completed.returncode == 0 summary = ( completed.stdout.strip().splitlines()[-1] if completed.stdout else "no output" ) details = {"returncode": completed.returncode, "stderr": completed.stderr} elif operation == "run_lint": - lint_paths = params.get("paths", ["src", "tests"]) + raw_paths = params.get("paths", ["src", "tests"]) + if isinstance(raw_paths, str): + lint_paths = [raw_paths] + elif isinstance(raw_paths, (list, tuple)): + lint_paths = [str(p) for p in raw_paths] + else: + lint_paths = ["src", "tests"] command = ["flake8", "--jobs", "1"] + lint_paths - completed = _run(command, cwd=cwd) + completed = _run(command, cwd=cwd, timeout=timeout) success = completed.returncode == 0 summary = completed.stdout.strip() or "No lint issues found" details = {"returncode": completed.returncode, "stderr": completed.stderr} else: # run_typecheck command = ["mypy"] - completed = _run(command, cwd=cwd) + completed = _run(command, cwd=cwd, timeout=timeout) success = completed.returncode == 0 summary = completed.stdout.strip() or "No type issues found" details = {"returncode": completed.returncode, "stderr": completed.stderr} + except subprocess.TimeoutExpired: + return ToolResult( + success=False, + summary=f"(real, failed) Execution timed out after {timeout}s", + details={"error": "timeout", "timeout": timeout}, + risk_level=spec.risk_level, + was_dry_run=dry_run, + approval=approval, + ) except (OSError, FileNotFoundError) as exc: return ToolResult( success=False, diff --git a/tests/tools/test_real_python_tool.py b/tests/tools/test_real_python_tool.py index 4cda3c7..963dd63 100644 --- a/tests/tools/test_real_python_tool.py +++ b/tests/tools/test_real_python_tool.py @@ -1,5 +1,7 @@ from pathlib import Path +import pytest + from devops_learn.tools.approval import ApprovalRecord from devops_learn.tools.base import RiskLevel from devops_learn.tools.python_tool import RealPythonTool, SimulatedPythonTool @@ -45,3 +47,39 @@ def test_real_python_tool_reports_failure_for_bad_path() -> None: approval=None, ) assert not result.success + + +def test_real_python_tool_runs_lint_on_single_and_multiple_paths() -> None: + tool = RealPythonTool() + result = tool.execute( + "run_lint", + {"paths": "src"}, + dry_run=False, + approval=None, + ) + has_summary = "flake8" in result.summary + has_code = result.details.get("returncode") is not None + assert result.success or has_summary or has_code + + +def test_real_python_tool_timeout_returns_failure_result( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import subprocess + tool = RealPythonTool() + + def mock_run(*args: object, **kwargs: object) -> None: + raise subprocess.TimeoutExpired(cmd=["test"], timeout=0.1) + + import devops_learn.tools.python_tool as pt_module + monkeypatch.setattr(pt_module, "_run", mock_run) + + result = tool.execute( + "run_tests", + {"path": ".", "timeout": 0.1}, + dry_run=False, + approval=None, + ) + assert not result.success + assert "Execution timed out" in result.summary + assert result.details.get("error") == "timeout"