tools/vcs/tests/test_vcs.py

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

import os
import subprocess
from pathlib import Path

import pytest

from magpie_vcs import (
    BACKENDS,
    FossilBackend,
    GitBackend,
    MercurialBackend,
    SubversionBackend,
    VCSError,
    detect_backend,
    get_backend,
    main,
)

# Tests shell out to `git` in temp repos. If the process inherits
# location-redirecting git env vars (e.g. when the suite runs inside a git
# pre-commit hook), git would operate on the outer repo instead. Strip them
# for the whole module so the fixture's own `git init` is unaffected; the tool
# itself does the same scrub at runtime (see _clean_env).
for _var in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GIT_COMMON_DIR", "GIT_PREFIX"):
    os.environ.pop(_var, None)

git_required = pytest.mark.skipif(not GitBackend.is_available(), reason="git not installed")
hg_required = pytest.mark.skipif(not MercurialBackend.is_available(), reason="hg not installed")


def _git(repo: Path, *args: str) -> None:
    subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True)


def _hg(repo: Path, *args: str) -> None:
    subprocess.run(["hg", *args], cwd=repo, check=True, capture_output=True)


@pytest.fixture
def git_repo(tmp_path: Path) -> Path:
    repo = tmp_path / "repo"
    repo.mkdir()
    _git(repo, "init", "-q", "-b", "main")
    _git(repo, "config", "user.email", "t@example.com")
    _git(repo, "config", "user.name", "Tester")
    _git(repo, "config", "commit.gpgsign", "false")
    (repo / "file.txt").write_text("hello\n")
    _git(repo, "add", "file.txt")
    _git(repo, "commit", "-q", "-m", "initial commit")
    return repo


@pytest.fixture
def hg_repo(tmp_path: Path) -> Path:
    repo = tmp_path / "repo_hg"
    repo.mkdir()
    _hg(repo, "init")
    with open(repo / ".hg" / "hgrc", "w") as f:
        f.write("[ui]\nusername = Tester <t@example.com>\n")
    (repo / "file.txt").write_text("hello\n")
    _hg(repo, "add", "file.txt")
    _hg(repo, "commit", "-m", "initial commit")
    return repo


# -- detection -------------------------------------------------------------


def test_detect_git(git_repo: Path) -> None:
    backend = detect_backend(git_repo)
    assert backend is not None
    assert backend.name == "git"
    assert backend.root == git_repo


def test_detect_marks_hg_svn_fossil(tmp_path: Path) -> None:
    (tmp_path / ".hg").mkdir()
    backend = detect_backend(tmp_path)
    assert isinstance(backend, MercurialBackend)

    svn = tmp_path / "svn_wc"
    svn.mkdir()
    (svn / ".svn").mkdir()
    assert isinstance(detect_backend(svn), SubversionBackend)

    fossil_dir = tmp_path / "fossil_wc"
    fossil_dir.mkdir()
    (fossil_dir / ".fslckg").touch()
    assert isinstance(detect_backend(fossil_dir), FossilBackend)

    fossil_dir2 = tmp_path / "fossil_wc_win"
    fossil_dir2.mkdir()
    (fossil_dir2 / "_FOSSIL_").touch()
    assert isinstance(detect_backend(fossil_dir2), FossilBackend)


def test_detect_none(tmp_path: Path) -> None:
    assert detect_backend(tmp_path) is None


def test_nested_innermost_wins(git_repo: Path) -> None:
    inner = git_repo / "inner"
    inner.mkdir()
    (inner / ".hg").mkdir()
    backend = detect_backend(inner)
    assert isinstance(backend, MercurialBackend)
    assert backend.root == inner


def test_get_backend_override(git_repo: Path) -> None:
    assert get_backend(git_repo, override="git").name == "git"
    with pytest.raises(VCSError, match="unknown backend"):
        get_backend(git_repo, override="bzr")


def test_get_backend_env(git_repo: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("MAGPIE_VCS", "hg")
    assert get_backend(git_repo).name == "hg"


def test_get_backend_no_repo(tmp_path: Path) -> None:
    with pytest.raises(VCSError, match="no supported VCS"):
        get_backend(tmp_path)


# -- git backend operations ------------------------------------------------


@git_required
def test_git_clean_then_dirty(git_repo: Path) -> None:
    backend = GitBackend(git_repo)
    assert backend.is_clean()
    (git_repo / "file.txt").write_text("changed\n")
    assert not backend.is_clean()
    assert "file.txt" in backend.status()


@git_required
def test_git_branch_and_commit(git_repo: Path) -> None:
    backend = GitBackend(git_repo)
    assert backend.current_branch() == "main"
    backend.create_branch("fix/thing")
    assert backend.current_branch() == "fix/thing"
    (git_repo / "new.txt").write_text("x\n")
    backend.stage(["new.txt"])
    assert "new.txt" in backend.diff(cached=True)
    backend.commit("add new.txt")
    assert backend.is_clean()
    assert "add new.txt" in backend.log(max_count=1)


@git_required
def test_git_ignores_inherited_git_dir(git_repo: Path, monkeypatch: pytest.MonkeyPatch) -> None:
    # Simulate running inside an outer git context (e.g. a pre-commit hook):
    # the tool must resolve the repo from its cwd, not from $GIT_DIR.
    monkeypatch.setenv("GIT_DIR", str(git_repo.parent / "elsewhere" / ".git"))
    monkeypatch.setenv("GIT_INDEX_FILE", str(git_repo.parent / "elsewhere" / "index"))
    backend = GitBackend(git_repo)
    assert backend.current_branch() == "main"
    assert backend.is_clean()


@git_required
def test_git_log_grep(git_repo: Path) -> None:
    backend = GitBackend(git_repo)
    assert "initial commit" in backend.log(grep="initial")
    assert backend.log(grep="nonexistent-token").strip() == ""


@git_required
def test_git_stage_nothing_rejected(git_repo: Path) -> None:
    with pytest.raises(VCSError, match="refusing to stage nothing"):
        GitBackend(git_repo).stage([])


@git_required
def test_git_reset_worktree(git_repo: Path) -> None:
    backend = GitBackend(git_repo)
    (git_repo / "file.txt").write_text("dirty\n")
    (git_repo / "untracked.txt").write_text("junk\n")
    backend.reset_worktree()
    assert backend.is_clean()
    assert not (git_repo / "untracked.txt").exists()
    assert (git_repo / "file.txt").read_text() == "hello\n"


@git_required
def test_git_cat(git_repo: Path) -> None:
    backend = GitBackend(git_repo)
    assert backend.cat("HEAD", "file.txt") == "hello\n"


# -- unimplemented backends ------------------------------------------------


def test_unimplemented_raise_with_issue(tmp_path: Path) -> None:
    svn = SubversionBackend(tmp_path)
    with pytest.raises(VCSError, match=r"apache/magpie#602"):
        svn.commit("x")
    assert svn.distributed is False  # centralized model flagged


# -- hg backend operations -------------------------------------------------


@hg_required
def test_hg_clean_then_dirty(hg_repo: Path) -> None:
    backend = MercurialBackend(hg_repo)
    assert backend.is_clean()
    (hg_repo / "file.txt").write_text("changed\n")
    assert not backend.is_clean()
    assert "file.txt" in backend.status()


@hg_required
def test_hg_bookmark_and_commit(hg_repo: Path) -> None:
    backend = MercurialBackend(hg_repo)
    assert backend.current_branch() == "default"
    backend.create_branch("fix-bookmark")
    assert backend.current_branch() == "fix-bookmark"
    (hg_repo / "new.txt").write_text("x\n")
    backend.stage(["new.txt"])
    assert "new.txt" in backend.diff()
    backend.commit("add new.txt")
    assert backend.is_clean()
    assert "add new.txt" in backend.log(max_count=1)


@hg_required
def test_hg_cached_diff_raises(hg_repo: Path) -> None:
    backend = MercurialBackend(hg_repo)
    with pytest.raises(VCSError, match="does not support staging area"):
        backend.diff(cached=True)


@hg_required
def test_hg_reset_worktree(hg_repo: Path) -> None:
    backend = MercurialBackend(hg_repo)
    (hg_repo / ".hgignore").write_text("ignored.txt\n")
    backend.stage([".hgignore"])
    backend.commit("add hgignore")

    (hg_repo / "file.txt").write_text("dirty\n")
    (hg_repo / "untracked.txt").write_text("junk\n")
    (hg_repo / "ignored.txt").write_text("ignored\n")
    backend.reset_worktree()
    assert backend.is_clean()
    assert not (hg_repo / "untracked.txt").exists()
    assert (hg_repo / "file.txt").read_text() == "hello\n"
    assert (hg_repo / "ignored.txt").exists()  # ignored files should be preserved


def test_registry_unique_names() -> None:
    names = [b.name for b in BACKENDS]
    assert names == sorted(set(names), key=names.index)
    assert "git" in names


# -- CLI -------------------------------------------------------------------


@git_required
def test_cli_detect_and_status(git_repo: Path, capsys: pytest.CaptureFixture[str]) -> None:
    assert main(["-C", str(git_repo), "detect"]) == 0
    assert capsys.readouterr().out.strip() == "git"

    (git_repo / "file.txt").write_text("z\n")
    assert main(["-C", str(git_repo), "clean"]) == 1
    main(["-C", str(git_repo), "reset-worktree"])
    assert main(["-C", str(git_repo), "clean"]) == 0


def test_cli_backends_lists_all(capsys: pytest.CaptureFixture[str]) -> None:
    assert main(["backends"]) == 0
    out = capsys.readouterr().out
    for name in ("git", "hg", "svn", "fossil"):
        assert name in out


def test_cli_unknown_backend_errors(git_repo: Path, capsys: pytest.CaptureFixture[str]) -> None:
    assert main(["-C", str(git_repo), "--backend", "bzr", "status"]) == 2
    assert "unknown backend" in capsys.readouterr().err


def test_cli_unimplemented_backend_errors(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
    (tmp_path / ".svn").mkdir()
    assert main(["-C", str(tmp_path), "status"]) == 2
    assert "apache/magpie#602" in capsys.readouterr().err


@git_required
def test_cli_cat(git_repo: Path, capsys: pytest.CaptureFixture[str]) -> None:
    assert main(["-C", str(git_repo), "cat", "HEAD", "file.txt"]) == 0
    assert capsys.readouterr().out == "hello\n"


def test_fossil_parser_status(monkeypatch: pytest.MonkeyPatch) -> None:
    def mock_run(args: list[str], cwd: str | None = None, **kwargs: object) -> str:
        if "changes" in args:
            return "EDITED     file1.txt\nADDED      file2.txt\nDELETED    file3.txt\n"
        if "extras" in args:
            return "untracked1.txt\nuntracked2.txt\n"
        return ""

    monkeypatch.setattr("magpie_vcs._run", mock_run)
    backend = FossilBackend(Path("/tmp/repo"))
    status_out = backend.status()
    assert "M file1.txt" in status_out
    assert "A file2.txt" in status_out
    assert "D file3.txt" in status_out
    assert "? untracked1.txt" in status_out
    assert "? untracked2.txt" in status_out


def test_fossil_parser_status_error(monkeypatch: pytest.MonkeyPatch) -> None:
    def mock_run(args: list[str], cwd: str | None = None, **kwargs: object) -> str:
        raise VCSError("command failed")

    monkeypatch.setattr("magpie_vcs._run", mock_run)
    backend = FossilBackend(Path("/tmp/repo"))
    assert backend.status() == ""


def test_fossil_parser_log(monkeypatch: pytest.MonkeyPatch) -> None:
    sample_timeline = """
=== 2026-07-08 ===
17:00:00 [a1b2c3d4e5f6] Initial commit (user: arnav tags: trunk)
16:30:00 [f6e5d4c3b2a1] Fix a bug in forum post rendering (user: jsmith tags: trunk)
16:00:00 [1234567890ab] Add a new feature [options] (user: alice tags: trunk)
"""

    def mock_run(args: list[str], cwd: str | None = None, **kwargs: object) -> str:
        if "timeline" in args:
            return sample_timeline
        return ""

    monkeypatch.setattr("magpie_vcs._run", mock_run)
    backend = FossilBackend(Path("/tmp/repo"))

    # Test basic log
    log_out = backend.log()
    assert "a1b2c3d4e5f6 Initial commit" in log_out
    assert "f6e5d4c3b2a1 Fix a bug in forum post rendering" in log_out
    assert "1234567890ab Add a new feature [options]" in log_out

    # Test grep search
    grep_out = backend.log(grep="bug")
    assert "f6e5d4c3b2a1 Fix a bug in forum post rendering" in grep_out
    assert "a1b2c3d4e5f6" not in grep_out