248 lines
7.4 KiB
Python
248 lines
7.4 KiB
Python
import asyncio
|
|
from collections.abc import Sequence
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from agentci.integrations.git import Git, GitError
|
|
|
|
|
|
class FakeProcess:
|
|
def __init__(
|
|
self,
|
|
stdout: bytes = b"",
|
|
stderr: bytes = b"",
|
|
returncode: int = 0,
|
|
) -> None:
|
|
self.stdout = stdout
|
|
self.stderr = stderr
|
|
self.returncode = returncode
|
|
|
|
async def communicate(self) -> tuple[bytes, bytes]:
|
|
return self.stdout, self.stderr
|
|
|
|
|
|
class SubprocessRecorder:
|
|
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 git_client(tmp_path: Path) -> Git:
|
|
return Git(
|
|
gitea_url="https://git.example.test/",
|
|
username="agent-user",
|
|
token="secret-token",
|
|
askpass_path=tmp_path / "askpass.sh",
|
|
commit_name="Agent CI",
|
|
commit_email="agent@example.test",
|
|
)
|
|
|
|
|
|
def install_recorder(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
outcomes: Sequence[FakeProcess | OSError],
|
|
) -> SubprocessRecorder:
|
|
recorder = SubprocessRecorder(outcomes)
|
|
monkeypatch.setattr(asyncio, "create_subprocess_exec", recorder)
|
|
return recorder
|
|
|
|
|
|
async def test_clone_uses_authenticated_remote_and_returns_head(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
for name in (
|
|
"GIT_ASKPASS",
|
|
"GIT_TERMINAL_PROMPT",
|
|
"AGENTCI_GIT_USERNAME",
|
|
"AGENTCI_GIT_PASSWORD",
|
|
):
|
|
monkeypatch.delenv(name, raising=False)
|
|
recorder = install_recorder(
|
|
monkeypatch,
|
|
[FakeProcess(), FakeProcess(stdout=b"abc123\n")],
|
|
)
|
|
destination = tmp_path / "workspaces" / "repo"
|
|
|
|
sha = await git_client(tmp_path).clone("org", "repo", "main", destination)
|
|
|
|
assert sha == "abc123"
|
|
assert destination.parent.is_dir()
|
|
assert recorder.commands == [
|
|
(
|
|
"git",
|
|
"clone",
|
|
"--branch",
|
|
"main",
|
|
"--single-branch",
|
|
"https://git.example.test/org/repo.git",
|
|
str(destination),
|
|
),
|
|
("git", "rev-parse", "HEAD"),
|
|
]
|
|
clone_kwargs = recorder.calls[0][1]
|
|
assert clone_kwargs["cwd"] == destination.parent
|
|
assert clone_kwargs["stdout"] is asyncio.subprocess.PIPE
|
|
assert clone_kwargs["stderr"] is asyncio.subprocess.PIPE
|
|
assert (
|
|
clone_kwargs["env"]
|
|
| {
|
|
"GIT_ASKPASS": str(tmp_path / "askpass.sh"),
|
|
"GIT_TERMINAL_PROMPT": "0",
|
|
"AGENTCI_GIT_USERNAME": "agent-user",
|
|
"AGENTCI_GIT_PASSWORD": "secret-token",
|
|
}
|
|
== clone_kwargs["env"]
|
|
)
|
|
current_sha_environment = recorder.calls[1][1]["env"]
|
|
assert "AGENTCI_GIT_PASSWORD" not in current_sha_environment
|
|
|
|
|
|
async def test_sync_branch_resets_and_cleans_before_returning_sha(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
recorder = install_recorder(
|
|
monkeypatch,
|
|
[FakeProcess(), FakeProcess(), FakeProcess(), FakeProcess(stdout=b"new-sha\n")],
|
|
)
|
|
workspace = tmp_path / "repo"
|
|
|
|
sha = await git_client(tmp_path).sync_branch(workspace, "feature")
|
|
|
|
assert sha == "new-sha"
|
|
assert recorder.commands == [
|
|
(
|
|
"git",
|
|
"fetch",
|
|
"origin",
|
|
"refs/heads/feature:refs/remotes/origin/feature",
|
|
),
|
|
("git", "reset", "--hard", "origin/feature"),
|
|
("git", "clean", "-fd"),
|
|
("git", "rev-parse", "HEAD"),
|
|
]
|
|
assert all(call[1]["cwd"] == workspace for call in recorder.calls)
|
|
assert recorder.calls[0][1]["env"]["AGENTCI_GIT_PASSWORD"] == "secret-token"
|
|
|
|
|
|
async def test_branch_status_and_diff_commands(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
recorder = install_recorder(
|
|
monkeypatch,
|
|
[FakeProcess(), FakeProcess(stdout=b" M src/app.py\n"), FakeProcess()],
|
|
)
|
|
workspace = tmp_path / "repo"
|
|
git = git_client(tmp_path)
|
|
|
|
await git.create_branch(workspace, "agent/issue-1")
|
|
changed = await git.has_changes(workspace)
|
|
await git.diff_check(workspace)
|
|
|
|
assert changed is True
|
|
assert recorder.commands == [
|
|
("git", "switch", "-c", "agent/issue-1"),
|
|
("git", "status", "--porcelain"),
|
|
("git", "diff", "--check"),
|
|
]
|
|
|
|
|
|
async def test_has_changes_is_false_for_clean_status(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
install_recorder(monkeypatch, [FakeProcess(stdout=b"\n")])
|
|
|
|
assert await git_client(tmp_path).has_changes(tmp_path / "repo") is False
|
|
|
|
|
|
async def test_commit_stages_all_changes_sets_identity_and_returns_sha(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
recorder = install_recorder(
|
|
monkeypatch,
|
|
[FakeProcess(), FakeProcess(), FakeProcess(stdout=b"commit-sha\n")],
|
|
)
|
|
workspace = tmp_path / "repo"
|
|
|
|
sha = await git_client(tmp_path).commit(workspace, "agent: Fix widget")
|
|
|
|
assert sha == "commit-sha"
|
|
assert recorder.commands == [
|
|
("git", "add", "-A"),
|
|
(
|
|
"git",
|
|
"-c",
|
|
"user.name=Agent CI",
|
|
"-c",
|
|
"user.email=agent@example.test",
|
|
"commit",
|
|
"-m",
|
|
"agent: Fix widget",
|
|
),
|
|
("git", "rev-parse", "HEAD"),
|
|
]
|
|
|
|
|
|
async def test_push_command_supports_initial_and_existing_branches(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
recorder = install_recorder(monkeypatch, [FakeProcess(), FakeProcess()])
|
|
workspace = tmp_path / "repo"
|
|
git = git_client(tmp_path)
|
|
|
|
await git.push(workspace, "agent/new", set_upstream=True)
|
|
await git.push(workspace, "agent/existing")
|
|
|
|
assert recorder.commands == [
|
|
("git", "push", "--set-upstream", "origin", "agent/new"),
|
|
("git", "push", "origin", "HEAD:agent/existing"),
|
|
]
|
|
assert all(call[1]["env"]["GIT_TERMINAL_PROMPT"] == "0" for call in recorder.calls)
|
|
|
|
|
|
async def test_git_translates_process_start_failure(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
install_recorder(monkeypatch, [OSError("git executable missing")])
|
|
|
|
with pytest.raises(GitError, match="Could not run git rev-parse") as error:
|
|
await git_client(tmp_path).current_sha(tmp_path / "repo")
|
|
|
|
assert isinstance(error.value.__cause__, OSError)
|
|
|
|
|
|
async def test_git_translates_nonzero_exit_and_stderr(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
install_recorder(
|
|
monkeypatch,
|
|
[FakeProcess(stderr=b"fatal: invalid diff\n", returncode=2)],
|
|
)
|
|
|
|
with pytest.raises(GitError, match="git diff failed: fatal: invalid diff"):
|
|
await git_client(tmp_path).diff_check(tmp_path / "repo")
|
|
|
|
|
|
async def test_commit_failure_identifies_commit_operation(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
install_recorder(
|
|
monkeypatch,
|
|
[FakeProcess(), FakeProcess(stderr=b"nothing to commit\n", returncode=1)],
|
|
)
|
|
|
|
with pytest.raises(GitError, match="git commit failed: nothing to commit"):
|
|
await git_client(tmp_path).commit(tmp_path / "repo", "agent: change")
|