nisse/nisse-dev/tests/test_git.py

"""Git on the build VM (D4, L13): here, a throwaway sshd on this machine."""

import shlex
import subprocess
from pathlib import Path

import pytest

from nisse_dev.execution import SSHHost
from nisse_dev.git import GitError, GitRepo


@pytest.fixture
def repo(ssh_host: SSHHost, tmp_path: Path) -> GitRepo:
    r = GitRepo(ssh_host, tmp_path / "work")
    r.init()
    r.configure_identity("nisse-dev-testing", "nisse@example.test")
    return r


def bare_remote(path: Path) -> str:
    subprocess.run(["git", "init", "-q", "--bare", "-b", "main", str(path)], check=True)
    return str(path)


def test_init_creates_repo_on_default_branch(repo: GitRepo) -> None:
    assert repo.current_branch() == "main"


def test_commit_all_returns_sha_and_branch_switching(repo: GitRepo) -> None:
    (repo.root / "a.txt").write_text("a")
    sha = repo.commit_all("first")
    assert sha is not None
    assert sha == repo.head_sha()

    repo.checkout_new_branch("milestone-1")
    assert repo.current_branch() == "milestone-1"


def test_commit_with_nothing_staged_returns_none(repo: GitRepo) -> None:
    (repo.root / "a.txt").write_text("a")
    repo.commit_all("first")
    # An empty milestone is a fact to report, not an error to raise.
    assert repo.commit_all("nothing changed") is None


def test_has_changes_tracks_the_worktree(repo: GitRepo) -> None:
    assert not repo.has_changes()
    (repo.root / "a.txt").write_text("a")
    assert repo.has_changes()


def test_git_error_on_bad_command(ssh_host: SSHHost, tmp_path: Path) -> None:
    repo = GitRepo(ssh_host, tmp_path / "not-a-repo")
    repo.root.mkdir()
    with pytest.raises(GitError):
        repo.head_sha()


def test_commits_since_and_remote_sha_against_a_local_remote(
    repo: GitRepo, tmp_path: Path
) -> None:
    remote = bare_remote(tmp_path / "remote.git")
    (repo.root / "a.txt").write_text("a")
    base = repo.commit_all("base")
    repo.push(remote, "main")
    repo._git("update-ref", "refs/remotes/origin/main", "HEAD")
    assert repo.remote_sha(remote, "main") == base
    assert repo.remote_sha(remote, "absent") is None
    (repo.root / "b.txt").write_text("b")
    repo.commit_all("ahead")
    assert repo.commits_since("origin/main") == 1


# --- the push key (#89) ------------------------------------------------------


@pytest.fixture
def ssh_remote(ssh_host: SSHHost, tmp_path: Path) -> tuple[str, str]:
    """A bare repository reached by git over SSH, and the key that may push
    to it: the forge, as the build VM sees it."""
    remote = bare_remote(tmp_path / "forge.git")
    url = f"ssh://{ssh_host.user}@127.0.0.1:{ssh_host.port}{remote}"
    return url, ssh_host._key.read_text()


def test_clone_and_push_with_the_key_store_no_credential(
    ssh_host: SSHHost, tmp_path: Path, ssh_remote: tuple[str, str]
) -> None:
    url, key = ssh_remote
    seed = GitRepo(ssh_host, tmp_path / "seed", known_hosts=str(tmp_path / "kh"))
    seed.init()
    seed.configure_identity("t", "t@example.test")
    (seed.root / "a.txt").write_text("a")
    seed.commit_all("first")
    seed.push(url, "main", key=key)

    clone = GitRepo(ssh_host, tmp_path / "clone", known_hosts=str(tmp_path / "kh"))
    clone.clone(url, key=key)
    assert (clone.root / "a.txt").read_text() == "a"
    config = (clone.root / ".git" / "config").read_text()
    assert "PRIVATE KEY" not in config
    # The URL recorded is the credential-free one we gave.
    assert url in config


def test_the_key_is_gone_when_the_command_ends(
    ssh_host: SSHHost, tmp_path: Path, ssh_remote: tuple[str, str]
) -> None:
    url, key = ssh_remote
    repo = GitRepo(ssh_host, tmp_path / "w", known_hosts=str(tmp_path / "kh"))
    repo.init()
    repo.configure_identity("t", "t@example.test")
    (repo.root / "a.txt").write_text("a")
    repo.commit_all("first")
    before = _agents()
    repo.push(url, "main", key=key)
    assert _agents() == before  # the throwaway ssh-agent was killed


def test_a_failed_push_still_kills_its_agent(
    ssh_host: SSHHost, tmp_path: Path, ssh_remote: tuple[str, str]
) -> None:
    _, key = ssh_remote
    repo = GitRepo(ssh_host, tmp_path / "w", known_hosts=str(tmp_path / "kh"))
    repo.init()
    before = _agents()
    with pytest.raises(GitError):
        repo.push(
            f"ssh://nobody@127.0.0.1:{ssh_host.port}/nowhere.git", "main", key=key
        )
    assert _agents() == before


def _agents() -> set[str]:
    out = subprocess.run(
        ["pgrep", "-x", "ssh-agent"], capture_output=True, text=True, check=False
    ).stdout
    return set(out.split())


def test_a_remote_url_is_quoted_not_interpreted(repo: GitRepo, tmp_path: Path) -> None:
    marker = tmp_path / "pwned"
    with pytest.raises(GitError):
        repo.push(f"x; touch {shlex.quote(str(marker))}", "main")
    assert not marker.exists()