cleveragents-core/features/steps/tdd_quality_gate_steps.py

"""Step definitions for TDD quality gate feature tests."""

from __future__ import annotations

import os
import shutil
import sys
import tempfile
from pathlib import Path

from behave import given, then, when

# Ensure the project root is importable.
_ROOT = str(Path(__file__).resolve().parents[2])
if _ROOT not in sys.path:
    sys.path.insert(0, _ROOT)

from scripts.tdd_quality_gate import (  # noqa: E402
    check_expected_fail_removed,
    find_tdd_tests,
    parse_bug_refs,
    run_quality_gate,
)
from scripts.tdd_quality_gate import main as quality_gate_main  # noqa: E402

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _ensure_temp_dir(context: object) -> Path:
    """Return (and lazily create) the temporary directory on context."""
    tmp: Path | None = getattr(context, "temp_dir", None)
    if tmp is None:
        tmp = Path(tempfile.mkdtemp())
        context.temp_dir = tmp  # type: ignore[attr-defined]
    return tmp


def _cleanup_temp_dir(context: object) -> None:
    """Remove the temporary directory if it exists."""
    tmp: Path | None = getattr(context, "temp_dir", None)
    if tmp is not None and tmp.exists():
        shutil.rmtree(tmp, ignore_errors=True)


def _default_pr_diff_for_bug_refs(
    bug_refs: list[int], search_root: Path | None = None
) -> str:
    """Return a synthetic PR diff that removes expected-fail tags.

    When *search_root* is provided the helper inspects the temp tree to
    decide whether each bug's TDD test lives in a ``.feature`` or
    ``.robot`` file and emits the diff in the matching format.
    """
    chunks: list[str] = []
    for bug_num in bug_refs:
        use_robot = False
        if search_root is not None:
            robot_hits = list(search_root.rglob("*.robot"))
            for rp in robot_hits:
                try:
                    if f"tdd_bug_{bug_num}" in rp.read_text(encoding="utf-8"):
                        use_robot = True
                        break
                except (OSError, UnicodeDecodeError):
                    continue

        if use_robot:
            chunks.append(
                "\n".join(
                    [
                        f"diff --git a/robot/bug{bug_num}.robot b/robot/bug{bug_num}.robot",
                        f"--- a/robot/bug{bug_num}.robot",
                        f"+++ b/robot/bug{bug_num}.robot",
                        "@@ -1 +1 @@",
                        f"-tdd_expected_fail    tdd_bug    tdd_bug_{bug_num}",
                        f"+tdd_bug    tdd_bug_{bug_num}",
                    ]
                )
            )
        else:
            chunks.append(
                "\n".join(
                    [
                        f"diff --git a/features/bug{bug_num}.feature b/features/bug{bug_num}.feature",
                        f"--- a/features/bug{bug_num}.feature",
                        f"+++ b/features/bug{bug_num}.feature",
                        "@@ -1 +1 @@",
                        f"-@tdd_expected_fail @tdd_bug @tdd_bug_{bug_num}",
                        f"+@tdd_bug @tdd_bug_{bug_num}",
                    ]
                )
            )
    return "\n".join(chunks)


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


@given('a PR description "{description}"')
def step_given_pr_description(context: object, description: str) -> None:
    context.pr_description = description  # type: ignore[attr-defined]


@given("a multiline PR description")
def step_given_multiline_pr_description(context: object) -> None:
    context.pr_description = context.text  # type: ignore[attr-defined]


@when("I parse the bug references")
def step_when_parse_bug_refs(context: object) -> None:
    context.bug_refs = parse_bug_refs(  # type: ignore[attr-defined]
        context.pr_description  # type: ignore[attr-defined]
    )


@then("the bug references should be [{refs}]")
def step_then_bug_refs_should_be(context: object, refs: str) -> None:
    if refs.strip() == "":
        expected: list[int] = []
    else:
        expected = [int(x.strip()) for x in refs.split(",")]
    actual: list[int] = context.bug_refs  # type: ignore[attr-defined]
    if actual != expected:
        raise AssertionError(f"Expected bug refs {expected}, got {actual}")


@then("the bug references should be []")
def step_then_bug_refs_empty(context: object) -> None:
    actual: list[int] = context.bug_refs  # type: ignore[attr-defined]
    if actual != []:
        raise AssertionError(f"Expected empty bug refs, got {actual}")


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


@given('a temporary directory with a file "{filepath}" containing "{content}"')
def step_given_temp_dir_with_file(context: object, filepath: str, content: str) -> None:
    _cleanup_temp_dir(context)
    tmp = _ensure_temp_dir(context)
    full_path = tmp / filepath
    full_path.parent.mkdir(parents=True, exist_ok=True)
    full_path.write_text(content, encoding="utf-8")


@given('a temporary directory also has a file "{filepath}" containing "{content}"')
def step_given_temp_dir_also_has_file(
    context: object, filepath: str, content: str
) -> None:
    tmp = _ensure_temp_dir(context)
    full_path = tmp / filepath
    full_path.parent.mkdir(parents=True, exist_ok=True)
    full_path.write_text(content, encoding="utf-8")


@when("I search for TDD tests for bug {bug_num:d}")
def step_when_search_tdd_tests(context: object, bug_num: int) -> None:
    tmp = _ensure_temp_dir(context)
    context.found_tests = find_tdd_tests(bug_num, tmp)  # type: ignore[attr-defined]


@then("the search should find {count:d} test file")
def step_then_search_finds_count(context: object, count: int) -> None:
    actual = len(context.found_tests)  # type: ignore[attr-defined]
    if actual != count:
        raise AssertionError(f"Expected {count} test file(s), found {actual}")


@then("the search should find {count:d} test files")
def step_then_search_finds_count_plural(context: object, count: int) -> None:
    actual = len(context.found_tests)  # type: ignore[attr-defined]
    if actual != count:
        raise AssertionError(f"Expected {count} test file(s), found {actual}")


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


@when("I check expected fail removal for bug {bug_num:d}")
def step_when_check_removal(context: object, bug_num: int) -> None:
    tmp = _ensure_temp_dir(context)
    # Filter by bug tag first — matching the production path in run_quality_gate.
    test_files = find_tdd_tests(bug_num, tmp)
    context.removal_errors = check_expected_fail_removed(  # type: ignore[attr-defined]
        test_files, bug_num
    )


@then("there should be {count:d} removal error")
def step_then_removal_errors_count(context: object, count: int) -> None:
    actual = len(context.removal_errors)  # type: ignore[attr-defined]
    if actual != count:
        raise AssertionError(
            f"Expected {count} removal error(s), got {actual}: {context.removal_errors}"  # type: ignore[attr-defined]
        )


@then("there should be {count:d} removal errors")
def step_then_removal_errors_count_plural(context: object, count: int) -> None:
    actual = len(context.removal_errors)  # type: ignore[attr-defined]
    if actual != count:
        raise AssertionError(
            f"Expected {count} removal error(s), got {actual}: {context.removal_errors}"  # type: ignore[attr-defined]
        )


@then('the removal error should mention "{text}"')
def step_then_removal_error_mentions(context: object, text: str) -> None:
    errors: list[str] = context.removal_errors  # type: ignore[attr-defined]
    found = any(text in err for err in errors)
    if not found:
        raise AssertionError(
            f"Expected removal error mentioning '{text}', got: {errors}"
        )


# ---------------------------------------------------------------------------
# Full quality gate
# ---------------------------------------------------------------------------


@given("a temporary search root")
def step_given_temp_search_root(context: object) -> None:
    _cleanup_temp_dir(context)
    _ensure_temp_dir(context)
    context.pr_diff = None  # type: ignore[attr-defined]


@given('a temporary search root with file "{filepath}" containing "{content}"')
def step_given_temp_search_root_with_file(
    context: object, filepath: str, content: str
) -> None:
    _cleanup_temp_dir(context)
    tmp = _ensure_temp_dir(context)
    context.pr_diff = None  # type: ignore[attr-defined]
    full_path = tmp / filepath
    full_path.parent.mkdir(parents=True, exist_ok=True)
    full_path.write_text(content, encoding="utf-8")


@given('the search root also has file "{filepath}" containing "{content}"')
def step_given_search_root_also_has(
    context: object, filepath: str, content: str
) -> None:
    tmp = _ensure_temp_dir(context)
    full_path = tmp / filepath
    full_path.parent.mkdir(parents=True, exist_ok=True)
    full_path.write_text(content, encoding="utf-8")


@given("the PR diff does not remove expected fail tags")
def step_given_pr_diff_no_expected_fail_removal(context: object) -> None:
    context.pr_diff = ""  # type: ignore[attr-defined]


@when("I run the quality gate")
def step_when_run_quality_gate(context: object) -> None:
    tmp = _ensure_temp_dir(context)
    pr_desc: str = getattr(context, "pr_description", "")
    pr_diff: str | None = getattr(context, "pr_diff", None)
    if pr_diff is None:
        bug_refs = parse_bug_refs(pr_desc)
        pr_diff = _default_pr_diff_for_bug_refs(bug_refs, tmp)
    errors, _bug_refs = run_quality_gate(pr_desc, tmp, pr_diff=pr_diff)
    context.gate_errors = errors  # type: ignore[attr-defined]


@then("the quality gate should pass")
def step_then_gate_passes(context: object) -> None:
    errors: list[str] = context.gate_errors  # type: ignore[attr-defined]
    if errors:
        raise AssertionError(f"Quality gate should pass but got errors: {errors}")


@then("the quality gate should fail")
def step_then_gate_fails(context: object) -> None:
    errors: list[str] = context.gate_errors  # type: ignore[attr-defined]
    if not errors:
        raise AssertionError("Quality gate should fail but passed with no errors")


@then('the quality gate errors should mention "{text}"')
def step_then_gate_errors_mention(context: object, text: str) -> None:
    errors: list[str] = context.gate_errors  # type: ignore[attr-defined]
    found = any(text in err for err in errors)
    if not found:
        raise AssertionError(
            f"Expected quality gate error mentioning '{text}', got: {errors}"
        )


# ---------------------------------------------------------------------------
# Argument validation
# ---------------------------------------------------------------------------


@when("I call parse_bug_refs with a non-string argument")
def step_when_parse_bug_refs_non_string(context: object) -> None:
    context.caught_exception = None  # type: ignore[attr-defined]
    try:
        parse_bug_refs(123)  # type: ignore[arg-type]
    except TypeError as exc:
        context.caught_exception = exc  # type: ignore[attr-defined]


@when("I call find_tdd_tests with bug number {num:d}")
def step_when_find_tdd_tests_invalid(context: object, num: int) -> None:
    context.caught_exception = None  # type: ignore[attr-defined]
    try:
        find_tdd_tests(num, Path("/tmp/nonexistent"))
    except ValueError as exc:
        context.caught_exception = exc  # type: ignore[attr-defined]


@when("I call find_tdd_tests with a non-Path search root")
def step_when_find_tdd_tests_non_path(context: object) -> None:
    context.caught_exception = None  # type: ignore[attr-defined]
    try:
        find_tdd_tests(1, "/tmp/nonexistent")  # type: ignore[arg-type]
    except TypeError as exc:
        context.caught_exception = exc  # type: ignore[attr-defined]


@then("a TypeError should be raised by the quality gate")
def step_then_type_error_raised(context: object) -> None:
    exc = getattr(context, "caught_exception", None)
    if exc is None:
        raise AssertionError("Expected TypeError but no exception was raised")
    if not isinstance(exc, TypeError):
        raise AssertionError(f"Expected TypeError, got {type(exc).__name__}: {exc}")


@then("a ValueError should be raised by the quality gate")
def step_then_value_error_raised(context: object) -> None:
    exc = getattr(context, "caught_exception", None)
    if exc is None:
        raise AssertionError("Expected ValueError but no exception was raised")
    if not isinstance(exc, ValueError):
        raise AssertionError(f"Expected ValueError, got {type(exc).__name__}: {exc}")


# ---------------------------------------------------------------------------
# Robot-format diff, unreadable files, and extra argument validation
# ---------------------------------------------------------------------------


@given("the PR diff removes expected fail for robot bug {bug_num:d}")
def step_given_robot_diff_removes_expected_fail(context: object, bug_num: int) -> None:
    context.pr_diff = "\n".join(  # type: ignore[attr-defined]
        [
            f"diff --git a/robot/bug{bug_num}.robot b/robot/bug{bug_num}.robot",
            f"--- a/robot/bug{bug_num}.robot",
            f"+++ b/robot/bug{bug_num}.robot",
            "@@ -1 +1 @@",
            f"-tdd_expected_fail    tdd_bug    tdd_bug_{bug_num}",
            f"+tdd_bug    tdd_bug_{bug_num}",
        ]
    )


@given("a temporary directory with an unreadable feature file for bug {bug_num:d}")
def step_given_unreadable_feature_file(context: object, bug_num: int) -> None:
    _cleanup_temp_dir(context)
    tmp = _ensure_temp_dir(context)
    full_path = tmp / "features" / "bug.feature"
    full_path.parent.mkdir(parents=True, exist_ok=True)
    # Write invalid UTF-8 bytes so read_text(encoding="utf-8") raises
    # UnicodeDecodeError (caught as OSError subclass).  This is root-safe
    # unlike chmod(0o000) which root bypasses.
    full_path.write_bytes(
        f"@tdd_expected_fail @tdd_bug @tdd_bug_{bug_num}".encode() + b"\xff\xfe"
    )


@when("I call check_expected_fail_removed with a non-list argument")
def step_when_check_expected_fail_removed_non_list(context: object) -> None:
    context.caught_exception = None  # type: ignore[attr-defined]
    try:
        check_expected_fail_removed("not-a-list", 1)  # type: ignore[arg-type]
    except TypeError as exc:
        context.caught_exception = exc  # type: ignore[attr-defined]


@when("I call check_expected_fail_removed with bug number {num:d}")
def step_when_check_expected_fail_removed_bad_bug(context: object, num: int) -> None:
    context.caught_exception = None  # type: ignore[attr-defined]
    try:
        check_expected_fail_removed([], num)
    except ValueError as exc:
        context.caught_exception = exc  # type: ignore[attr-defined]


# ---------------------------------------------------------------------------
# Bool type guard
# ---------------------------------------------------------------------------


@when("I call find_tdd_tests with boolean True as bug number")
def step_when_find_tdd_tests_bool(context: object) -> None:
    context.caught_exception = None  # type: ignore[attr-defined]
    try:
        find_tdd_tests(True, Path("/tmp/nonexistent"))  # type: ignore[arg-type]
    except (TypeError, ValueError) as exc:
        context.caught_exception = exc  # type: ignore[attr-defined]


@when("I call check_expected_fail_removed with boolean True as bug number")
def step_when_check_expected_fail_removed_bool(context: object) -> None:
    context.caught_exception = None  # type: ignore[attr-defined]
    try:
        check_expected_fail_removed([], True)  # type: ignore[arg-type]
    except (TypeError, ValueError) as exc:
        context.caught_exception = exc  # type: ignore[attr-defined]


# ---------------------------------------------------------------------------
# Co-located bug false positive guard (M1)
# ---------------------------------------------------------------------------


@given("the PR diff only removes expected fail for bug {other:d} not bug {target:d}")
def step_given_pr_diff_removes_wrong_bug(
    context: object, other: int, target: int
) -> None:
    # Craft a diff that removes expected-fail for bug `other` but NOT for
    # bug `target`.  The diff must mention bug `other`'s tag on the removed
    # line, so the gate should NOT count this as a removal for `target`.
    context.pr_diff = "\n".join(  # type: ignore[attr-defined]
        [
            "diff --git a/features/bugs99.feature b/features/bugs99.feature",
            "--- a/features/bugs99.feature",
            "+++ b/features/bugs99.feature",
            "@@ -1 +1 @@",
            f"-@tdd_expected_fail @tdd_bug @tdd_bug_{other}",
            f"+@tdd_bug @tdd_bug_{other}",
        ]
    )


# ---------------------------------------------------------------------------
# run_quality_gate argument validation (L4)
# ---------------------------------------------------------------------------


@when("I call run_quality_gate with a non-string PR description")
def step_when_run_quality_gate_non_str_desc(context: object) -> None:
    context.caught_exception = None  # type: ignore[attr-defined]
    try:
        run_quality_gate(123, Path("/tmp"))  # type: ignore[arg-type]
    except TypeError as exc:
        context.caught_exception = exc  # type: ignore[attr-defined]


@when("I call run_quality_gate with a non-Path search root")
def step_when_run_quality_gate_non_path_root(context: object) -> None:
    context.caught_exception = None  # type: ignore[attr-defined]
    try:
        run_quality_gate("desc", "/tmp")  # type: ignore[arg-type]
    except TypeError as exc:
        context.caught_exception = exc  # type: ignore[attr-defined]


@when("I call run_quality_gate with an empty base_ref")
def step_when_run_quality_gate_empty_base_ref(context: object) -> None:
    context.caught_exception = None  # type: ignore[attr-defined]
    try:
        run_quality_gate("desc", Path("/tmp"), base_ref="  ")
    except ValueError as exc:
        context.caught_exception = exc  # type: ignore[attr-defined]


@when("I call run_quality_gate with a non-string pr_diff")
def step_when_run_quality_gate_non_str_diff(context: object) -> None:
    context.caught_exception = None  # type: ignore[attr-defined]
    try:
        run_quality_gate("desc", Path("/tmp"), pr_diff=123)  # type: ignore[arg-type]
    except TypeError as exc:
        context.caught_exception = exc  # type: ignore[attr-defined]


# ---------------------------------------------------------------------------
# main() CLI entry point (M2)
# ---------------------------------------------------------------------------


@when('I call main with PR_DESCRIPTION "{description}"')
def step_when_call_main_with_desc(context: object, description: str) -> None:
    tmp = _ensure_temp_dir(context)
    saved_desc = os.environ.get("PR_DESCRIPTION")
    saved_cwd = os.getcwd()
    try:
        os.environ["PR_DESCRIPTION"] = description
        os.chdir(tmp)
        context.main_exit_code = quality_gate_main()  # type: ignore[attr-defined]
    finally:
        os.chdir(saved_cwd)
        if saved_desc is None:
            os.environ.pop("PR_DESCRIPTION", None)
        else:
            os.environ["PR_DESCRIPTION"] = saved_desc


@when('I call main with PR_DESCRIPTION "{description}" in an empty search root')
def step_when_call_main_missing_test(context: object, description: str) -> None:
    _cleanup_temp_dir(context)
    tmp = _ensure_temp_dir(context)
    saved_desc = os.environ.get("PR_DESCRIPTION")
    saved_base = os.environ.get("PR_BASE_REF")
    saved_cwd = os.getcwd()
    try:
        os.environ["PR_DESCRIPTION"] = description
        # Use a base_ref that won't exist in the temp dir (not a git repo),
        # so _collect_pr_diff will fail and produce an error about the diff.
        os.environ["PR_BASE_REF"] = "master"
        os.chdir(tmp)
        context.main_exit_code = quality_gate_main()  # type: ignore[attr-defined]
    finally:
        os.chdir(saved_cwd)
        if saved_desc is None:
            os.environ.pop("PR_DESCRIPTION", None)
        else:
            os.environ["PR_DESCRIPTION"] = saved_desc
        if saved_base is None:
            os.environ.pop("PR_BASE_REF", None)
        else:
            os.environ["PR_BASE_REF"] = saved_base


@then("the main exit code should be {code:d}")
def step_then_main_exit_code(context: object, code: int) -> None:
    actual = context.main_exit_code  # type: ignore[attr-defined]
    if actual != code:
        raise AssertionError(f"Expected main exit code {code}, got {actual}")