cleveragents-core/features/steps/git_tools_steps.py

"""Step definitions for the Built-in Git Tools feature."""

from __future__ import annotations

import os
import shutil
import subprocess
import tempfile
from pathlib import Path
from typing import Any
from unittest.mock import patch

from behave import given, then, when

from cleveragents.tool.builtins import register_git_tools
from cleveragents.tool.builtins.git_tools import (
    ALL_GIT_TOOLS,
    _map_git_error,
    _resolve_sandbox_root,
    validate_repo_path,
)
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runner import ToolRunner

__all__: list[str] = []


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


def _git(args: list[str], cwd: str) -> subprocess.CompletedProcess[str]:
    """Run a git command in the given directory.

    Sets ``safe.directory`` so git does not refuse to operate in temp
    directories owned by a different user (common in Docker / CI).
    """
    env = {
        **os.environ,
        "GIT_CONFIG_NOSYSTEM": "1",
        "GIT_TERMINAL_PROMPT": "0",
        "GIT_CONFIG_COUNT": "1",
        "GIT_CONFIG_KEY_0": "safe.directory",
        "GIT_CONFIG_VALUE_0": cwd,
    }
    return subprocess.run(
        ["git", *args],
        cwd=cwd,
        capture_output=True,
        text=True,
        check=True,
        timeout=30,
        env=env,
    )


def _run_git_tool(context: Any, tool_name: str, inputs: dict[str, Any]) -> None:
    """Execute a built-in git tool through the runner.

    Lazily initialises a shared ``ToolRunner`` on *context* to avoid
    creating a fresh registry per invocation within the same scenario.

    The temp repo directory is passed as ``_sandbox_root`` so that the
    absolute ``/tmp/…`` path used in tests passes sandbox validation.
    """
    if not hasattr(context, "_git_runner"):
        registry = ToolRegistry()
        register_git_tools(registry)
        context._git_runner = ToolRunner(registry)
    # Point repo_path at the temp git repo unless already set
    if "repo_path" not in inputs:
        inputs["repo_path"] = context.git_repo_dir
    # Allow the absolute temp-dir path through sandbox validation
    if hasattr(context, "git_repo_dir") and "_sandbox_root" not in inputs:
        inputs["_sandbox_root"] = context.git_repo_dir
    context.tool_result = context._git_runner.execute(tool_name, inputs)


# ---------------------------------------------------------------------------
# Givens
# ---------------------------------------------------------------------------


@given("a temporary git repository")
def step_given_temp_git_repo(context: Any) -> None:
    repo_dir = tempfile.mkdtemp()
    context.git_repo_dir = repo_dir
    context._cleanup_handlers.append(
        lambda: shutil.rmtree(repo_dir, ignore_errors=True)
    )
    # Initialise a git repo with an initial commit
    _git(["init"], cwd=repo_dir)
    _git(["config", "user.email", "test@test.com"], cwd=repo_dir)
    _git(["config", "user.name", "Test User"], cwd=repo_dir)
    # Create an initial commit so HEAD exists
    readme = Path(repo_dir) / "README.md"
    readme.write_text("# Test repo\n")
    _git(["add", "."], cwd=repo_dir)
    _git(["commit", "-m", "Initial commit"], cwd=repo_dir)


@given('a tracked file "{name}" with content "{content}"')
def step_given_tracked_file(context: Any, name: str, content: str) -> None:
    path = Path(context.git_repo_dir) / name
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content)
    _git(["add", name], cwd=context.git_repo_dir)
    _git(["commit", "-m", f"Add {name}"], cwd=context.git_repo_dir)


@given('a tracked file "{name}" with multiline content')
def step_given_tracked_file_multiline(context: Any, name: str) -> None:
    path = Path(context.git_repo_dir) / name
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text("line one\nline two\nline three\n")
    _git(["add", name], cwd=context.git_repo_dir)
    _git(["commit", "-m", f"Add {name}"], cwd=context.git_repo_dir)


@given('the file "{name}" is modified to "{content}"')
def step_given_file_modified(context: Any, name: str, content: str) -> None:
    path = Path(context.git_repo_dir) / name
    path.write_text(content)


@given('the file "{name}" is modified and staged to "{content}"')
def step_given_file_modified_and_staged(context: Any, name: str, content: str) -> None:
    path = Path(context.git_repo_dir) / name
    path.write_text(content)
    _git(["add", name], cwd=context.git_repo_dir)


@given("a temporary git repository with multiple commits")
def step_given_temp_git_repo_multi(context: Any) -> None:
    repo_dir = tempfile.mkdtemp()
    context.git_repo_dir = repo_dir
    context._cleanup_handlers.append(
        lambda: shutil.rmtree(repo_dir, ignore_errors=True)
    )
    _git(["init"], cwd=repo_dir)
    _git(["config", "user.email", "test@test.com"], cwd=repo_dir)
    _git(["config", "user.name", "Test User"], cwd=repo_dir)
    for i in range(3):
        f = Path(repo_dir) / f"file{i}.txt"
        f.write_text(f"content {i}")
        _git(["add", "."], cwd=repo_dir)
        _git(["commit", "-m", f"Commit {i}"], cwd=repo_dir)


@given("a temporary non-git directory")
def step_given_temp_non_git_dir(context: Any) -> None:
    # Force /tmp so the directory is NOT inside a git worktree (the nox
    # session creates temp dirs under /app/.nox/… which IS inside one).
    tmp_dir = tempfile.mkdtemp(dir="/tmp")
    context.non_git_dir = tmp_dir
    context._cleanup_handlers.append(lambda: shutil.rmtree(tmp_dir, ignore_errors=True))


@given("a temporary git repository with no commits")
def step_given_temp_git_repo_no_commits(context: Any) -> None:
    # Force /tmp for the same reason as the non-git directory step.
    repo_dir = tempfile.mkdtemp(dir="/tmp")
    context.empty_repo_dir = repo_dir
    context._cleanup_handlers.append(
        lambda: shutil.rmtree(repo_dir, ignore_errors=True)
    )
    _git(["init"], cwd=repo_dir)
    _git(["config", "user.email", "test@test.com"], cwd=repo_dir)
    _git(["config", "user.name", "Test User"], cwd=repo_dir)


@given('an unrecognized git error message "{stderr}"')
def step_given_unrecognized_git_error(context: Any, stderr: str) -> None:
    context.raw_git_stderr = stderr


@given("a sandbox root directory")
def step_given_sandbox_root_dir(context: Any) -> None:
    tmp_dir = tempfile.mkdtemp(dir="/tmp")
    context.sandbox_root_path = Path(tmp_dir)
    context._cleanup_handlers.append(lambda: shutil.rmtree(tmp_dir, ignore_errors=True))


@given("the git subprocess is configured to time out")
def step_given_git_subprocess_timeout(context: Any) -> None:
    context.git_timeout_patch = patch(
        "cleveragents.tool.builtins.git_tools._run_git",
        side_effect=subprocess.TimeoutExpired(cmd=["git"], timeout=30),
    )
    context.git_timeout_mock = context.git_timeout_patch.start()
    context._cleanup_handlers.append(context.git_timeout_patch.stop)


# Note: "a tool registry" step is already defined in tool_runtime_steps.py
# and reused here via Behave's shared step registry.


# ---------------------------------------------------------------------------
# Whens
# ---------------------------------------------------------------------------


@when('I execute the "builtin/git-status" tool')
def step_when_git_status(context: Any) -> None:
    _run_git_tool(context, "builtin/git-status", {})


@when('I execute the "builtin/git-status" tool with short format')
def step_when_git_status_short(context: Any) -> None:
    _run_git_tool(context, "builtin/git-status", {"short": True})


@when('I execute the "builtin/git-status" tool with repo_path "{path}"')
def step_when_git_status_repo_path(context: Any, path: str) -> None:
    _run_git_tool(context, "builtin/git-status", {"repo_path": path})


@when('I execute the "builtin/git-diff" tool')
def step_when_git_diff(context: Any) -> None:
    _run_git_tool(context, "builtin/git-diff", {})


@when('I execute the "builtin/git-diff" tool with staged flag')
def step_when_git_diff_staged(context: Any) -> None:
    _run_git_tool(context, "builtin/git-diff", {"staged": True})


@when('I execute the "builtin/git-diff" tool with stat flag')
def step_when_git_diff_stat(context: Any) -> None:
    _run_git_tool(context, "builtin/git-diff", {"stat": True})


@when('I execute the "builtin/git-log" tool')
def step_when_git_log(context: Any) -> None:
    _run_git_tool(context, "builtin/git-log", {})


@when('I execute the "builtin/git-log" tool with max_count {count:d}')
def step_when_git_log_max(context: Any, count: int) -> None:
    _run_git_tool(context, "builtin/git-log", {"max_count": count})


@when('I execute the "builtin/git-log" tool with oneline format')
def step_when_git_log_oneline(context: Any) -> None:
    _run_git_tool(context, "builtin/git-log", {"oneline": True})


@when('I execute the "builtin/git-blame" tool for "{path}"')
def step_when_git_blame(context: Any, path: str) -> None:
    _run_git_tool(context, "builtin/git-blame", {"path": path})


@when('I execute the "builtin/git-blame" tool for "{path}" lines {start:d} to {end:d}')
def step_when_git_blame_range(context: Any, path: str, start: int, end: int) -> None:
    _run_git_tool(
        context,
        "builtin/git-blame",
        {"path": path, "line_start": start, "line_end": end},
    )


@when('I execute the "builtin/git-diff" tool with ref "{ref}"')
def step_when_git_diff_ref(context: Any, ref: str) -> None:
    _run_git_tool(context, "builtin/git-diff", {"ref": ref})


@when('I execute the "builtin/git-diff" tool with path "{path}"')
def step_when_git_diff_path(context: Any, path: str) -> None:
    _run_git_tool(context, "builtin/git-diff", {"path": path})


@when('I execute the "builtin/git-log" tool with author "{author}"')
def step_when_git_log_author(context: Any, author: str) -> None:
    _run_git_tool(context, "builtin/git-log", {"author": author})


@when('I execute the "builtin/git-log" tool with since "{since}"')
def step_when_git_log_since(context: Any, since: str) -> None:
    _run_git_tool(context, "builtin/git-log", {"since": since})


@when('I execute the "builtin/git-log" tool with until "{until}"')
def step_when_git_log_until(context: Any, until: str) -> None:
    _run_git_tool(context, "builtin/git-log", {"until": until})


@when('I execute the "builtin/git-log" tool with path "{path}"')
def step_when_git_log_path(context: Any, path: str) -> None:
    _run_git_tool(context, "builtin/git-log", {"path": path})


@when(
    'I execute the "builtin/git-blame" tool for "{path}" with only line_start {start:d}'
)
def step_when_git_blame_partial_range(context: Any, path: str, start: int) -> None:
    _run_git_tool(
        context,
        "builtin/git-blame",
        {"path": path, "line_start": start},
    )


@when('I execute the "builtin/git-status" tool in the non-git directory')
def step_when_git_status_non_git(context: Any) -> None:
    _run_git_tool(
        context,
        "builtin/git-status",
        {"repo_path": context.non_git_dir, "_sandbox_root": context.non_git_dir},
    )


@when('I execute the "builtin/git-log" tool in the empty repository')
def step_when_git_log_empty_repo(context: Any) -> None:
    _run_git_tool(
        context,
        "builtin/git-log",
        {
            "repo_path": context.empty_repo_dir,
            "_sandbox_root": context.empty_repo_dir,
        },
    )


@when("I map the git error to a user-friendly message")
def step_when_map_git_error(context: Any) -> None:
    context.mapped_error = _map_git_error(context.raw_git_stderr)


@when("I validate the repo path with no user-supplied path")
def step_when_validate_repo_path_none(context: Any) -> None:
    context.validated_path = validate_repo_path(
        None,
        sandbox_root=context.sandbox_root_path,
    )


@when("I resolve a sandbox root from an empty value")
def step_when_resolve_sandbox_root_empty(context: Any) -> None:
    context.resolved_sandbox_root = _resolve_sandbox_root(None)


@when('I execute the "builtin/git-status" tool against the timed-out subprocess')
def step_when_git_status_timeout(context: Any) -> None:
    _run_git_tool(context, "builtin/git-status", {})


@when('I execute the "builtin/git-diff" tool against the timed-out subprocess')
def step_when_git_diff_timeout(context: Any) -> None:
    _run_git_tool(context, "builtin/git-diff", {})


@when('I execute the "builtin/git-log" tool against the timed-out subprocess')
def step_when_git_log_timeout(context: Any) -> None:
    _run_git_tool(context, "builtin/git-log", {})


@when(
    'I execute the "builtin/git-blame" tool for "{path}"'
    " against the timed-out subprocess"
)
def step_when_git_blame_timeout(context: Any, path: str) -> None:
    _run_git_tool(context, "builtin/git-blame", {"path": path})


@when("I register all git tools")
def step_when_register_git_tools(context: Any) -> None:
    register_git_tools(context.registry)


# ---------------------------------------------------------------------------
# Thens
# ---------------------------------------------------------------------------

# Note: the following step definitions are named uniquely to avoid conflicts
# with existing tool_builtins_steps.py steps.  Generic steps like
# "the tool result should be successful" are already defined there and
# reused via Behave's step registry.


@then('the git output should contain "{text}"')
def step_then_git_output_contains(context: Any, text: str) -> None:
    assert context.tool_result.success, f"Tool failed: {context.tool_result.error}"
    output = context.tool_result.output.get("output", "")
    assert text in output, f"Expected '{text}' in output:\n{output}"


@then("the git output should be empty")
def step_then_git_output_empty(context: Any) -> None:
    assert context.tool_result.success, f"Tool failed: {context.tool_result.error}"
    output = context.tool_result.output.get("output", "")
    assert output.strip() == "", f"Expected empty output, got:\n{output}"


@then('the git output should not contain "{text}"')
def step_then_git_output_not_contains(context: Any, text: str) -> None:
    assert context.tool_result.success, f"Tool failed: {context.tool_result.error}"
    output = context.tool_result.output.get("output", "")
    assert text not in output, f"Did not expect '{text}' in output:\n{output}"


@then("the git log output should contain at most {count:d} commit")
def step_then_git_log_max_commits(context: Any, count: int) -> None:
    assert context.tool_result.success, f"Tool failed: {context.tool_result.error}"
    output = context.tool_result.output.get("output", "")
    # Standard git log separates commits with lines starting with "commit "
    commit_count = output.count("commit ")
    assert commit_count <= count, (
        f"Expected at most {count} commit(s), found {commit_count}:\n{output}"
    )


@then("the git log output should use oneline format")
def step_then_git_log_oneline_format(context: Any) -> None:
    assert context.tool_result.success, f"Tool failed: {context.tool_result.error}"
    output = context.tool_result.output.get("output", "").strip()
    # Oneline format: each non-empty line is "<hash> <message>" — no multi-line
    # commit blocks.  Verify no line starts with "Author:" or "Date:".
    for line in output.splitlines():
        assert not line.startswith("Author:"), (
            f"Oneline format should not contain Author lines:\n{output}"
        )
        assert not line.startswith("Date:"), (
            f"Oneline format should not contain Date lines:\n{output}"
        )


@then("the git blame output should contain exactly {count:d} lines")
def step_then_git_blame_line_count(context: Any, count: int) -> None:
    assert context.tool_result.success, f"Tool failed: {context.tool_result.error}"
    output = context.tool_result.output.get("output", "")
    # git blame produces one output line per source line
    blame_lines = [ln for ln in output.splitlines() if ln.strip()]
    assert len(blame_lines) == count, (
        f"Expected {count} blame line(s), got {len(blame_lines)}:\n{output}"
    )


@then('the mapped error should be "{expected}"')
def step_then_mapped_error_equals(context: Any, expected: str) -> None:
    assert context.mapped_error == expected, (
        f"Expected mapped error '{expected}', got '{context.mapped_error}'"
    )


@then("the validated path should equal the sandbox root")
def step_then_validated_path_equals_sandbox_root(context: Any) -> None:
    assert context.validated_path == context.sandbox_root_path.resolve(), (
        f"Expected {context.sandbox_root_path.resolve()}, got {context.validated_path}"
    )


@then("the resolved sandbox root should be absent")
def step_then_resolved_sandbox_root_is_none(context: Any) -> None:
    assert context.resolved_sandbox_root is None, (
        f"Expected None, got {context.resolved_sandbox_root}"
    )


@then("all registered git tools should be read-only")
def step_then_all_read_only(context: Any) -> None:
    for spec in ALL_GIT_TOOLS:
        tool = context.registry.get(spec.name)
        assert tool is not None, f"Tool {spec.name} not found"
        assert tool.capabilities.read_only is True, f"Tool {spec.name} is not read_only"
        assert tool.capabilities.writes is False, f"Tool {spec.name} has writes=True"