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, HTTPSToken, SSHKey


@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, SSHKey]:
    """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, SSHKey(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, SSHKey]
) -> 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", credential=key)

    clone = GitRepo(ssh_host, tmp_path / "clone", known_hosts=str(tmp_path / "kh"))
    clone.clone(url, credential=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, SSHKey]
) -> 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", credential=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, SSHKey]
) -> 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",
            credential=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()


class RecordingHost:
    """Captures what would be sent to the build VM."""

    def __init__(self) -> None:
        self.calls: list[dict] = []

    def execute(self, command: str, **kwargs):
        from nisse_dev.execution import Result

        self.calls.append({"command": command, **kwargs})
        return Result(0, "", "")


def test_an_https_token_travels_only_as_a_secret(tmp_path: Path) -> None:
    host = RecordingHost()
    repo = GitRepo(host, tmp_path)  # type: ignore[arg-type]
    token = "tok-" + "9f3a" * 6
    repo.push(
        "https://gitea.test/o/r.git", "m1", credential=HTTPSToken("pusher", token)
    )
    [call] = host.calls
    assert token not in call["command"]  # never on a command line
    assert "pusher" not in call["command"]
    assert call["secrets"] == {"NISSE_PUSH_USER": "pusher", "NISSE_PUSH_TOKEN": token}
    # Git asks a helper that reads the environment, and any helper configured
    # on the VM is cleared first.
    assert "credential.helper= " in call["command"]
    assert "$NISSE_PUSH_TOKEN" in call["command"]
    assert "GIT_TERMINAL_PROMPT=0" in call["command"]


def test_an_https_token_answers_git_through_the_helper(
    ssh_host: SSHHost, tmp_path: Path
) -> None:
    # The helper as git runs it, on the build host: asked to "get", it
    # answers from the environment; asked anything else, nothing.
    from nisse_dev.git import _HELPER

    body = _HELPER.removeprefix("!")
    asked = ssh_host.execute(
        f"{body} get",
        secrets={"NISSE_PUSH_USER": "pusher", "NISSE_PUSH_TOKEN": "tok-1234567"},
    )
    # Both travel as secrets, so both are masked in what comes back.
    assert asked.stdout == "username=[redacted]\npassword=[redacted]\n"
    stored = ssh_host.execute(f"{body} store", secrets={"NISSE_PUSH_TOKEN": "x"})
    assert stored.stdout == ""