152 lines
4.7 KiB
Python
152 lines
4.7 KiB
Python
import asyncio
|
|
from collections.abc import Sequence
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from agentci.integrations.codegraph import CodeGraph, CodeGraphError
|
|
|
|
|
|
class FakeProcess:
|
|
def __init__(self, returncode: int = 0, stderr: bytes = b"") -> None:
|
|
self.returncode = returncode
|
|
self.stderr = stderr
|
|
|
|
async def communicate(self) -> tuple[bytes, bytes]:
|
|
return b"", self.stderr
|
|
|
|
|
|
class CodeGraphProcessRecorder:
|
|
def __init__(self, outcomes: Sequence[FakeProcess | OSError]) -> None:
|
|
self.outcomes = list(outcomes)
|
|
self.calls: list[tuple[tuple[Any, ...], dict[str, Any]]] = []
|
|
|
|
async def __call__(self, *args: Any, **kwargs: Any) -> FakeProcess:
|
|
self.calls.append((args, kwargs))
|
|
outcome = self.outcomes.pop(0)
|
|
if isinstance(outcome, OSError):
|
|
raise outcome
|
|
return outcome
|
|
|
|
@property
|
|
def commands(self) -> list[tuple[Any, ...]]:
|
|
return [args for args, _ in self.calls]
|
|
|
|
|
|
def install_recorder(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
outcomes: Sequence[FakeProcess | OSError],
|
|
) -> CodeGraphProcessRecorder:
|
|
recorder = CodeGraphProcessRecorder(outcomes)
|
|
monkeypatch.setattr(
|
|
"agentci.integrations.codegraph.asyncio.create_subprocess_exec",
|
|
recorder,
|
|
)
|
|
return recorder
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("database_exists", "initial_exclude", "expected_command", "expected_exclude"),
|
|
[
|
|
pytest.param(False, None, "init", ".codegraph/\n", id="incomplete-index"),
|
|
pytest.param(
|
|
True,
|
|
"# local excludes\n.codegraph/\n",
|
|
"sync",
|
|
"# local excludes\n.codegraph/\n",
|
|
id="existing-index",
|
|
),
|
|
],
|
|
)
|
|
async def test_prepare_selects_init_or_sync_and_updates_git_exclude(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
database_exists: bool,
|
|
initial_exclude: str | None,
|
|
expected_command: str,
|
|
expected_exclude: str,
|
|
) -> None:
|
|
workspace = tmp_path / "repo"
|
|
(workspace / ".git" / "info").mkdir(parents=True)
|
|
(workspace / ".codegraph").mkdir()
|
|
if database_exists:
|
|
(workspace / ".codegraph" / "codegraph.db").touch()
|
|
exclude = workspace / ".git" / "info" / "exclude"
|
|
if initial_exclude is not None:
|
|
exclude.write_text(initial_exclude)
|
|
recorder = install_recorder(monkeypatch, [FakeProcess()])
|
|
|
|
await CodeGraph().prepare(workspace)
|
|
|
|
assert recorder.commands == [("codegraph", expected_command, str(workspace))]
|
|
kwargs = recorder.calls[0][1]
|
|
assert kwargs["cwd"] == workspace
|
|
assert kwargs["env"]["CODEGRAPH_TELEMETRY"] == "0"
|
|
assert kwargs["stdout"] is asyncio.subprocess.PIPE
|
|
assert kwargs["stderr"] is asyncio.subprocess.PIPE
|
|
assert exclude.read_text() == expected_exclude
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("initial", "expected"),
|
|
[
|
|
("*.pyc", "*.pyc\n.codegraph/\n"),
|
|
("# .codegraph/ is documented here\n", "# .codegraph/ is documented here\n.codegraph/\n"),
|
|
(" .codegraph/ \n", " .codegraph/ \n"),
|
|
],
|
|
)
|
|
async def test_handles_exclude_file_boundaries(
|
|
tmp_path: Path,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
initial: str,
|
|
expected: str,
|
|
) -> None:
|
|
workspace = tmp_path / "repo"
|
|
exclude = workspace / ".git" / "info" / "exclude"
|
|
exclude.parent.mkdir(parents=True)
|
|
exclude.write_text(initial)
|
|
|
|
install_recorder(monkeypatch, [FakeProcess()])
|
|
|
|
await CodeGraph().prepare(workspace)
|
|
|
|
assert exclude.read_text() == expected
|
|
|
|
|
|
async def test_reports_missing_executable(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
workspace = tmp_path / "repo"
|
|
workspace.mkdir()
|
|
|
|
install_recorder(
|
|
monkeypatch,
|
|
[FileNotFoundError(2, "No such file or directory", "codegraph")],
|
|
)
|
|
|
|
with pytest.raises(CodeGraphError, match="Could not run CodeGraph:.*codegraph") as raised:
|
|
await CodeGraph().prepare(workspace)
|
|
|
|
assert isinstance(raised.value.__cause__, FileNotFoundError)
|
|
assert (workspace / ".git" / "info" / "exclude").read_text() == ".codegraph/\n"
|
|
|
|
|
|
async def test_reports_nonzero_exit_with_bounded_non_utf8_stderr(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
workspace = tmp_path / "repo"
|
|
workspace.mkdir()
|
|
stderr = b"discarded-prefix" + (b"x" * 1200) + b"\xff useful-tail"
|
|
|
|
install_recorder(
|
|
monkeypatch,
|
|
[FakeProcess(returncode=7, stderr=stderr)],
|
|
)
|
|
|
|
with pytest.raises(CodeGraphError, match="codegraph init failed") as raised:
|
|
await CodeGraph().prepare(workspace)
|
|
|
|
message = str(raised.value)
|
|
assert "discarded-prefix" not in message
|
|
assert "useful-tail" in message
|
|
assert len(message.removeprefix("codegraph init failed: ")) == 1000
|