Coverage for src / domain / validation / helpers.py: 18%
77 statements
« prev ^ index » next coverage.py v7.13.0, created at 2026-01-04 04:43 +0000
1"""Shared helper functions for validation runners.
3These utilities are used by SpecValidationRunner and related modules.
4"""
6from __future__ import annotations
8import json
9import shutil
10from pathlib import Path
11from typing import TYPE_CHECKING
13if TYPE_CHECKING:
14 from src.core.protocols import CommandRunnerPort
17def tail(text: str, max_chars: int = 800, max_lines: int = 20) -> str:
18 """Truncate text to last N lines and M characters.
20 Args:
21 text: The text to truncate.
22 max_chars: Maximum number of characters to keep.
23 max_lines: Maximum number of lines to keep.
25 Returns:
26 The truncated text.
27 """
28 if not text:
29 return ""
30 lines = text.splitlines()
31 if len(lines) > max_lines:
32 lines = lines[-max_lines:]
33 clipped = "\n".join(lines)
34 if len(clipped) > max_chars:
35 return clipped[-max_chars:]
36 return clipped
39def decode_timeout_output(data: bytes | str | None) -> str:
40 """Decode TimeoutExpired stdout/stderr which may be bytes, str, or None.
42 Args:
43 data: The output data from TimeoutExpired exception.
45 Returns:
46 Decoded and truncated string.
47 """
48 if data is None:
49 return ""
50 if isinstance(data, str):
51 return tail(data)
52 return tail(data.decode())
55def format_step_output(
56 stdout_tail: str,
57 stderr_tail: str,
58 max_chars: int = 300,
59 max_lines: int = 6,
60) -> str:
61 """Format step output for error messages.
63 Prefers stderr over stdout when both are available.
65 Args:
66 stdout_tail: Truncated stdout.
67 stderr_tail: Truncated stderr.
68 max_chars: Maximum characters for output.
69 max_lines: Maximum lines for output.
71 Returns:
72 Formatted output string.
73 """
74 parts = []
75 if stderr_tail:
76 parts.append(
77 f"stderr: {tail(stderr_tail, max_chars=max_chars, max_lines=max_lines)}"
78 )
79 if stdout_tail and not stderr_tail:
80 parts.append(
81 f"stdout: {tail(stdout_tail, max_chars=max_chars, max_lines=max_lines)}"
82 )
83 return " | ".join(parts)
86def check_e2e_prereqs(env: dict[str, str]) -> str | None:
87 """Check prerequisites for E2E validation.
89 Args:
90 env: Environment variables (unused, kept for API compatibility).
92 Returns:
93 Error message if prereqs not met, None otherwise.
94 """
95 if not shutil.which("mala"):
96 return "E2E prereq missing: mala CLI not found in PATH"
97 if not shutil.which("bd"):
98 return "E2E prereq missing: bd CLI not found in PATH"
99 return None
102def _generate_fixture_programmatically(repo_path: Path) -> None:
103 """Generate E2E fixture files programmatically.
105 This is a fallback when the fixture template directory is not available
106 (e.g., in installed packages where only src/ is included).
108 Args:
109 repo_path: Path to create the fixture repository in.
110 """
111 # Create src/app.py with a bug (subtracts instead of adds)
112 src_dir = repo_path / "src"
113 src_dir.mkdir(parents=True, exist_ok=True)
114 (src_dir / "app.py").write_text(
115 "def add(a: int, b: int) -> int:\n return a - b\n"
116 )
118 # Create tests/test_app.py
119 tests_dir = repo_path / "tests"
120 tests_dir.mkdir(parents=True, exist_ok=True)
121 (tests_dir / "test_app.py").write_text(
122 """import pytest
124from app import add
127@pytest.mark.unit
128def test_add():
129 assert add(2, 2) == 4
130"""
131 )
133 # Create pyproject.toml
134 (repo_path / "pyproject.toml").write_text(
135 """[project]
136name = "mala-e2e-fixture"
137version = "0.0.0"
138description = "Fixture repo for mala e2e validation"
139requires-python = ">=3.11"
140dependencies = []
142[project.optional-dependencies]
143dev = ["pytest>=8.0.0", "pytest-cov>=4.1.0", "pytest-xdist>=3.8.0"]
145[tool.pytest.ini_options]
146pythonpath = ["src"]
147markers = [
148 "unit: fast, isolated tests (default)",
149 "integration: tests that exercise multiple components",
150]
152[dependency-groups]
153dev = [
154 "pytest>=9.0.2",
155 "pytest-cov>=6.0.0",
156 "pytest-xdist>=3.8.0",
157]
158"""
159 )
161 # Create mala.yaml
162 (repo_path / "mala.yaml").write_text(
163 """preset: python-uv
165run_level_commands:
166 test: "uv run pytest --cov=src --cov-report=xml:coverage.xml --cov-fail-under=0 -o cache_dir=/tmp/pytest-${AGENT_ID:-default}"
168coverage:
169 format: xml
170 file: coverage.xml
171 threshold: 0
172"""
173 )
176def write_fixture_repo(repo_path: Path) -> None:
177 """Create a minimal fixture repository for E2E testing.
179 Creates a simple Python project with a failing test that the
180 implementer agent needs to fix. Uses src/ layout for compatibility
181 with mala's coverage checking (--cov=src).
183 Fixture files are sourced from tests/fixtures/e2e-fixture if available,
184 otherwise generated programmatically (for installed packages).
186 Args:
187 repo_path: Path to create the fixture repository in.
188 """
189 fixture_root = (
190 Path(__file__).resolve().parents[3] / "tests" / "fixtures" / "e2e-fixture"
191 )
192 if fixture_root.exists():
193 shutil.copytree(fixture_root, repo_path, dirs_exist_ok=True)
194 else:
195 # Fallback to programmatic generation when fixture template is not available
196 # (e.g., in installed packages where only src/ is included in the wheel)
197 _generate_fixture_programmatically(repo_path)
200def init_fixture_repo(
201 repo_path: Path,
202 command_runner: CommandRunnerPort,
203) -> str | None:
204 """Initialize a fixture repository with git and beads.
206 Args:
207 repo_path: Path to the fixture repository.
208 command_runner: CommandRunnerPort for running commands.
210 Returns:
211 Error message if initialization failed, None on success.
212 """
213 runner = command_runner
215 for cmd in (
216 ["git", "init"],
217 ["git", "config", "user.email", "mala-e2e@example.com"],
218 ["git", "config", "user.name", "Mala E2E"],
219 ["git", "add", "."],
220 ["git", "commit", "-m", "initial"],
221 ["bd", "init"],
222 ["bd", "create", "Fix failing add() test", "-p", "1"],
223 ):
224 result = runner.run(cmd, cwd=repo_path)
225 if not result.ok:
226 stderr = result.stderr.strip()
227 reason = (
228 f"E2E fixture setup failed: {' '.join(cmd)} (exit {result.returncode})"
229 )
230 if stderr:
231 reason = f"{reason}: {tail(stderr)}"
232 return reason
233 return None
236def get_ready_issue_id(
237 repo_path: Path,
238 command_runner: CommandRunnerPort,
239) -> str | None:
240 """Get the first ready issue ID from a repository.
242 Args:
243 repo_path: Path to the repository.
244 command_runner: CommandRunnerPort for running commands.
246 Returns:
247 Issue ID if found, None otherwise.
248 """
249 runner = command_runner
251 result = runner.run(["bd", "ready", "--json"], cwd=repo_path)
252 if not result.ok:
253 return None
254 try:
255 issues = json.loads(result.stdout)
256 except json.JSONDecodeError:
257 return None
258 for issue in issues:
259 issue_id = issue.get("id")
260 if isinstance(issue_id, str):
261 return issue_id
262 return None
265def annotate_issue(
266 repo_path: Path,
267 issue_id: str,
268 command_runner: CommandRunnerPort,
269) -> None:
270 """Add test plan notes to an issue.
272 Args:
273 repo_path: Path to the repository.
274 issue_id: Issue ID to annotate.
275 command_runner: CommandRunnerPort for running commands.
276 """
277 runner = command_runner
279 notes = "\n".join(
280 [
281 "Context:",
282 "- Tests are failing for add() in app.py",
283 "",
284 "Acceptance Criteria:",
285 "- Fix add() so tests pass",
286 "- Run full validation suite",
287 "",
288 "Test Plan:",
289 "- uv run pytest",
290 ]
291 )
292 runner.run(["bd", "update", issue_id, "--notes", notes], cwd=repo_path)