Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 22 additions & 5 deletions src/devops_learn/tools/python_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -60,6 +60,7 @@ def _run(
capture_output=True,
text=True,
check=check,
timeout=timeout,
env=environment,
)

Expand Down Expand Up @@ -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,
Expand Down
38 changes: 38 additions & 0 deletions tests/tools/test_real_python_tool.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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"
Loading