cleveragents-core/scripts/tdd_quality_gate.py

#!/usr/bin/env python3
"""TDD bug tag quality gate for bug fix PRs.

Enforces the TDD bug fix workflow rules described in CONTRIBUTING.md:

1. Parses the PR description for closing keywords that reference bug issues
   (``Closes #N``, ``Fixes #N``, ``Resolves #N``, ``ISSUES CLOSED: #N``).
2. Searches the codebase for tests tagged ``@tdd_bug_N`` (Behave ``.feature``
   files) or ``tdd_bug_N`` (Robot ``.robot`` files).
3. Verifies that every such test has had its ``@tdd_expected_fail`` /
   ``tdd_expected_fail`` tag removed — the fix PR must remove the
   expected-fail marker as proof the bug is now fixed.

Exit codes:
  0 — All checks passed (or PR references no bugs).
  1 — One or more violations detected.

Usage:
    PR_DESCRIPTION="Fixes #42" python scripts/tdd_quality_gate.py

Or via nox::

    PR_DESCRIPTION="Fixes #42" nox -s tdd_quality_gate
"""

from __future__ import annotations

import functools
import os
import re
import subprocess
import sys
from pathlib import Path

# ---------------------------------------------------------------------------
# PR description parsing
# ---------------------------------------------------------------------------

# Matches: Closes #N, Fixes #N, Resolves #N (case-insensitive)
_CLOSING_KEYWORD_RE = re.compile(
    r"\b(?:close[sd]?|fix(?:e[sd])?|resolve[sd]?)\b\s+#(\d+)",
    re.IGNORECASE,
)

# Matches: ISSUES CLOSED: #N, #M, ...
_ISSUES_CLOSED_RE = re.compile(
    r"ISSUES\s+CLOSED\s*:\s*((?:#\d+[\s,]*)+)",
    re.IGNORECASE,
)

# Extracts individual issue numbers from the ISSUES CLOSED value
_ISSUE_NUMBER_RE = re.compile(r"#(\d+)")


@functools.lru_cache(maxsize=64)
def _tag_token_pattern(tag: str) -> re.Pattern[str]:
    """Return a compiled regex that matches ``tag`` as a full token."""
    escaped = re.escape(tag)
    return re.compile(rf"(?<![A-Za-z0-9_]){escaped}(?![A-Za-z0-9_])")


def _contains_tag_token(content: str, tag: str) -> bool:
    """Return True when ``tag`` appears as a full token in ``content``."""
    return _tag_token_pattern(tag).search(content) is not None


def _collect_pr_diff(search_root: Path, base_ref: str) -> str:
    """Collect unified diff between the PR branch and the base branch."""
    if not isinstance(search_root, Path):
        raise TypeError(f"search_root must be a Path, got {type(search_root).__name__}")
    if not isinstance(base_ref, str):
        raise TypeError(f"base_ref must be a str, got {type(base_ref).__name__}")
    if base_ref.strip() == "":
        raise ValueError("base_ref must not be empty")

    ranges = (f"origin/{base_ref}...HEAD", f"{base_ref}...HEAD")
    for ref_range in ranges:
        try:
            proc = subprocess.run(
                [
                    "git",
                    "diff",
                    "--no-color",
                    ref_range,
                    "--",
                    "*.feature",
                    "*.robot",
                ],
                cwd=search_root,
                check=True,
                capture_output=True,
                text=True,
            )
            return proc.stdout
        except (OSError, subprocess.CalledProcessError):
            continue

    raise RuntimeError(
        "Unable to compute PR diff against base branch. "
        "Ensure git history for the base branch is available and retry."
    )


def _diff_has_expected_fail_removal_for_bug(pr_diff: str, bug_number: int) -> bool:
    """Return True when PR diff removes expected-fail for ``bug_number``."""
    if not isinstance(pr_diff, str):
        raise TypeError(f"pr_diff must be a str, got {type(pr_diff).__name__}")
    if (
        isinstance(bug_number, bool)
        or not isinstance(bug_number, int)
        or bug_number < 1
    ):
        raise ValueError(f"bug_number must be a positive integer, got {bug_number!r}")

    current_suffix = ""
    in_hunk = False
    file_has_bug_tag = False
    file_removed_expected_fail = False

    for line in pr_diff.splitlines():
        if line.startswith("+++ "):
            if file_has_bug_tag and file_removed_expected_fail:
                return True
            raw_path = line[4:]
            if raw_path.startswith("b/"):
                raw_path = raw_path[2:]
            current_suffix = Path(raw_path).suffix.lower()
            in_hunk = False
            file_has_bug_tag = False
            file_removed_expected_fail = False
            continue

        if current_suffix not in {".feature", ".robot"}:
            continue

        if line.startswith("@@"):
            in_hunk = True
            continue

        if not in_hunk:
            continue

        if not line or line[0] not in {" ", "+", "-"}:
            continue

        content = line[1:]
        if current_suffix == ".feature":
            bug_tag = f"@tdd_bug_{bug_number}"
            expected_fail_tag = "@tdd_expected_fail"
        else:
            bug_tag = f"tdd_bug_{bug_number}"
            expected_fail_tag = "tdd_expected_fail"

        if _contains_tag_token(content, bug_tag):
            file_has_bug_tag = True
        if (
            line[0] == "-"
            and _contains_tag_token(content, expected_fail_tag)
            and _contains_tag_token(content, bug_tag)
        ):
            file_removed_expected_fail = True

    return file_has_bug_tag and file_removed_expected_fail


def parse_bug_refs(pr_description: str) -> list[int]:
    """Extract bug issue numbers from PR closing keywords.

    Recognises ``Closes #N``, ``Fixes #N``, ``Resolves #N``
    (case-insensitive) and ``ISSUES CLOSED: #N, #M``.

    Returns a deduplicated, sorted list of issue numbers.
    """
    if not isinstance(pr_description, str):
        raise TypeError(
            f"pr_description must be a str, got {type(pr_description).__name__}"
        )

    refs: set[int] = set()

    # Standard closing keywords
    for match in _CLOSING_KEYWORD_RE.finditer(pr_description):
        num = int(match.group(1))
        if num > 0:
            refs.add(num)

    # ISSUES CLOSED: #N, #M block
    for block_match in _ISSUES_CLOSED_RE.finditer(pr_description):
        block = block_match.group(1)
        for num_match in _ISSUE_NUMBER_RE.finditer(block):
            num = int(num_match.group(1))
            if num > 0:
                refs.add(num)

    return sorted(refs)


# ---------------------------------------------------------------------------
# TDD test search
# ---------------------------------------------------------------------------


def find_tdd_tests(
    bug_number: int,
    search_root: Path,
) -> list[Path]:
    """Find test files tagged with ``@tdd_bug_<bug_number>``.

    Searches ``.feature`` files for ``@tdd_bug_<N>`` and ``.robot``
    files for ``tdd_bug_<N>``.

    Returns a list of paths that contain the tag.
    """
    if (
        isinstance(bug_number, bool)
        or not isinstance(bug_number, int)
        or bug_number < 1
    ):
        raise ValueError(f"bug_number must be a positive integer, got {bug_number!r}")
    if not isinstance(search_root, Path):
        raise TypeError(f"search_root must be a Path, got {type(search_root).__name__}")

    tag_behave = f"@tdd_bug_{bug_number}"
    tag_robot = f"tdd_bug_{bug_number}"
    matches: list[Path] = []

    # Search .feature files
    for feature_file in sorted(search_root.rglob("*.feature")):
        try:
            content = feature_file.read_text(encoding="utf-8")
        except (OSError, UnicodeDecodeError):
            continue
        if _contains_tag_token(content, tag_behave):
            matches.append(feature_file)

    # Search .robot files
    for robot_file in sorted(search_root.rglob("*.robot")):
        try:
            content = robot_file.read_text(encoding="utf-8")
        except (OSError, UnicodeDecodeError):
            continue
        if _contains_tag_token(content, tag_robot):
            matches.append(robot_file)

    return matches


# ---------------------------------------------------------------------------
# Tag removal verification
# ---------------------------------------------------------------------------


def check_expected_fail_removed(
    test_files: list[Path],
    bug_number: int,
) -> list[str]:
    """Verify ``@tdd_expected_fail`` has been removed from test files.

    Returns a list of error messages for files that still contain the
    expected-fail tag.
    """
    if not isinstance(test_files, list):
        raise TypeError(f"test_files must be a list, got {type(test_files).__name__}")
    if (
        isinstance(bug_number, bool)
        or not isinstance(bug_number, int)
        or bug_number < 1
    ):
        raise ValueError(f"bug_number must be a positive integer, got {bug_number!r}")

    errors: list[str] = []

    for path in test_files:
        try:
            content = path.read_text(encoding="utf-8")
        except (OSError, UnicodeDecodeError):
            continue

        suffix = path.suffix.lower()
        if suffix == ".feature":
            tag = "@tdd_expected_fail"
        elif suffix == ".robot":
            tag = "tdd_expected_fail"
        else:
            continue

        if _contains_tag_token(content, tag):
            bug_tag_display = (
                f"@tdd_bug_{bug_number}"
                if suffix == ".feature"
                else f"tdd_bug_{bug_number}"
            )
            errors.append(
                f"Bug fix PR must remove the {tag} tag from tests tagged "
                f"{bug_tag_display}. "
                f"See CONTRIBUTING.md > Bug Fix Workflow."
            )

    return errors


# ---------------------------------------------------------------------------
# Main gate logic
# ---------------------------------------------------------------------------


def run_quality_gate(
    pr_description: str,
    search_root: Path,
    *,
    pr_diff: str | None = None,
    base_ref: str = "master",
) -> tuple[list[str], list[int]]:
    """Run the full TDD quality gate.

    Returns a ``(errors, bug_refs)`` tuple.  An empty error list means all
    checks passed.  ``bug_refs`` contains the parsed bug issue numbers so
    callers can avoid re-parsing the PR description.
    """
    if not isinstance(pr_description, str):
        raise TypeError(
            f"pr_description must be a str, got {type(pr_description).__name__}"
        )
    if not isinstance(search_root, Path):
        raise TypeError(f"search_root must be a Path, got {type(search_root).__name__}")
    if pr_diff is not None and not isinstance(pr_diff, str):
        raise TypeError(f"pr_diff must be a str or None, got {type(pr_diff).__name__}")
    if not isinstance(base_ref, str):
        raise TypeError(f"base_ref must be a str, got {type(base_ref).__name__}")
    if base_ref.strip() == "":
        raise ValueError("base_ref must not be empty")

    bug_refs = parse_bug_refs(pr_description)
    if not bug_refs:
        return [], bug_refs

    if pr_diff is None:
        try:
            pr_diff = _collect_pr_diff(search_root, base_ref)
        except RuntimeError as exc:
            return [str(exc)], bug_refs

    all_errors: list[str] = []

    for bug_num in bug_refs:
        test_files = find_tdd_tests(bug_num, search_root)

        if not test_files:
            all_errors.append(
                f"No TDD test found for bug #{bug_num}. "
                f"The TDD workflow requires a test tagged @tdd_bug_{bug_num} "
                f"to exist before the bug can be fixed. "
                f"See CONTRIBUTING.md > Bug Fix Workflow."
            )
            continue

        removal_errors = check_expected_fail_removed(test_files, bug_num)
        all_errors.extend(removal_errors)

        # Only check the diff when the file-level check found no tag issues;
        # otherwise the diff error would be redundant.
        if not removal_errors and not _diff_has_expected_fail_removal_for_bug(
            pr_diff, bug_num
        ):
            all_errors.append(
                "No removal of @tdd_expected_fail / tdd_expected_fail detected "
                f"in PR diff for bug #{bug_num}. "
                "The bug fix PR must remove the expected-fail tag in this branch. "
                "See CONTRIBUTING.md > Bug Fix Workflow."
            )

    return all_errors, bug_refs


# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------


def main() -> int:
    """CLI entry point.  Reads PR_DESCRIPTION from the environment."""
    pr_description = os.environ.get("PR_DESCRIPTION", "")
    base_ref = os.environ.get("PR_BASE_REF", "master")
    search_root = Path.cwd()

    errors, bug_refs = run_quality_gate(pr_description, search_root, base_ref=base_ref)

    if errors:
        for error in errors:
            print(f"ERROR: {error}", file=sys.stderr)
        return 1

    if bug_refs:
        print(f"TDD quality gate passed for bug(s): {bug_refs}")
    else:
        print("TDD quality gate: no bug references found in PR description (pass).")

    return 0


if __name__ == "__main__":
    sys.exit(main())