band-aider-pub/band_aider/judgement/python_coverage_strategy.py

# SPDX-FileCopyrightText: 2026 Art Sea LLC
#
# SPDX-License-Identifier: AGPL-3.0-only

from __future__ import annotations

import json
from dataclasses import dataclass, field
from pathlib import Path

from band_aider.context.class_meta import ClassMeta
from band_aider.judgement.coverage_strategy import CoverageStrategy
from band_aider.judgement.dependency_discovery_strategy import DependencyDiscoveryStrategy
from band_aider.judgement.test_environment import TEST_ENV_VARIABLES_TO_CLEAR, sanitized_test_environment
from band_aider.judgement.repo_exec import RepoExec, current_repo_exec

MINIMUM_BRANCH_COVERAGE = 0.80


@dataclass
class PythonCoverageStrategy(CoverageStrategy):
    """PythonCoverageStrategy checks corresponding tests and measures branch coverage through coverage.py."""

    dependency_strategy: DependencyDiscoveryStrategy
    repo_exec: RepoExec = field(default_factory=current_repo_exec)

    def corresponding_tests_for(
        self,
        source_file: ClassMeta,
        test_files: list[ClassMeta],
        repo_root: Path,
        src_root: Path,
        test_root: Path,
    ) -> list[ClassMeta]:
        """Return discovered direct tests limited to the supplied test-file set."""

        allowed_locations = {
            test_file.location
            for test_file in test_files
        }
        return [
            test_file
            for test_file in self.dependency_strategy.direct_tests_for(
                [source_file],
                test_files,
                src_root,
                test_root,
            )
            if test_file.location in allowed_locations
        ]

    def _source_report_keys(
        self,
        source_files: list[ClassMeta],
        report_files: dict[str, object],
        repo_root: Path,
    ) -> set[str]:
        """Return coverage report keys matching the supplied source files."""

        source_paths = {
            str(repo_root / source_file.location)
            for source_file in source_files
        }
        source_locations = {
            source_file.location
            for source_file in source_files
        }
        matches: set[str] = set()
        for key in report_files:
            path = Path(key)
            resolved = str(path) if path.is_absolute() else ""
            if resolved in source_paths or key in source_locations:
                matches.add(key)
                continue

            if any(key.endswith(location) for location in source_locations):
                matches.add(key)

        return matches

    def branch_coverage_for(
        self,
        source_files: list[ClassMeta],
        test_files: list[ClassMeta],
        repo_root: Path,
        src_root: Path,
        test_root: Path,
    ) -> float:
        """Return measured branch coverage for source files using supplied test files."""

        if not source_files or not test_files:
            return 0.0

        test_paths = [
            repo_root / test_file.location
            for test_file in test_files
        ]
        if not all(self.repo_exec.exists(test_path) for test_path in test_paths):
            return 0.0

        coverage_file = repo_root / ".band-aider-coverage"
        report_file = repo_root / ".band-aider-coverage.json"
        try:
            env = sanitized_test_environment(repo_exec=self.repo_exec) if self.repo_exec.preserves_host_environment else {}
            environment_kwargs = (
                {"unset_env": TEST_ENV_VARIABLES_TO_CLEAR}
                if not self.repo_exec.preserves_host_environment
                else {}
            )
            env["COVERAGE_FILE"] = self.repo_exec.path_in_exec(coverage_file)
            run_command = [
                self.repo_exec.python_bin(),
                "-m",
                "coverage",
                "run",
                "--branch",
                "-m",
                "pytest",
                *[str(path) for path in test_paths],
            ]
            run_result = self.repo_exec.run_cli(
                run_command,
                cwd=repo_root,
                env=env,
                **environment_kwargs,
                capture_output=True,
                text=True,
                check=False,
            )
            if run_result.returncode != 0:
                return 0.0

            json_command = [
                self.repo_exec.python_bin(),
                "-m",
                "coverage",
                "json",
                "-o",
                self.repo_exec.path_in_exec(report_file),
            ]
            json_result = self.repo_exec.run_cli(
                json_command,
                cwd=repo_root,
                env=env,
                **environment_kwargs,
                capture_output=True,
                text=True,
                check=False,
            )
            if json_result.returncode != 0 or not self.repo_exec.exists(report_file):
                return 0.0

            try:
                report = json.loads(self.repo_exec.read_text(report_file, encoding="utf-8"))
            except (OSError, json.JSONDecodeError):
                return 0.0
        finally:
            self.repo_exec.delete(report_file, missing_ok=True)
            self.repo_exec.delete(coverage_file, missing_ok=True)

        report_files = report.get("files")
        if not isinstance(report_files, dict):
            return 0.0

        covered_branches = 0
        total_branches = 0
        covered_lines = 0
        total_lines = 0
        for key in self._source_report_keys(source_files, report_files, repo_root):
            file_report = report_files.get(key)
            if not isinstance(file_report, dict):
                continue

            summary = file_report.get("summary")
            if not isinstance(summary, dict):
                continue

            covered_branches += int(summary.get("covered_branches") or 0)
            total_branches += int(summary.get("num_branches") or 0)
            covered_lines += int(summary.get("covered_lines") or 0)
            total_lines += int(summary.get("num_statements") or 0)

        if total_branches:
            return covered_branches / total_branches

        if total_lines:
            return covered_lines / total_lines

        return 0.0

    def coverage_is_sufficient(
        self,
        source_files: list[ClassMeta],
        test_files: list[ClassMeta],
        repo_root: Path,
        src_root: Path,
        test_root: Path,
    ) -> bool:
        """Return whether measured branch coverage satisfies the configured acceptance gate."""

        return (
            self.branch_coverage_for(
                source_files,
                test_files,
                repo_root,
                src_root,
                test_root,
            )
            >= MINIMUM_BRANCH_COVERAGE
        )