arborist/tests/test_vcs_source.py

"""Tests for the git and Mercurial repo sources.

Each test builds a tiny synthetic repo via the actual `git` / `hg` CLIs in
a tmp_path, ingests it, and asserts the round-trip and supersedes-chain
behavior. If the underlying VCS isn't on PATH, the test is skipped.
"""

from __future__ import annotations

import os
import shutil
import subprocess
from pathlib import Path

import pytest

from aborist.ingest import ingest_source
from aborist.sources.vcs import GitRepoSource, MercurialRepoSource
from aborist.store import connect


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _git_available() -> bool:
    return shutil.which("git") is not None


def _hg_available() -> bool:
    return shutil.which("hg") is not None


def _mk_git_repo(path: Path) -> None:
    path.mkdir(parents=True, exist_ok=True)
    env = {
        **os.environ,
        # Detached author so commits are reproducible and tests don't
        # accidentally pick up the host's git identity.
        "GIT_AUTHOR_NAME": "test",
        "GIT_AUTHOR_EMAIL": "t@example.invalid",
        "GIT_COMMITTER_NAME": "test",
        "GIT_COMMITTER_EMAIL": "t@example.invalid",
        "GIT_AUTHOR_DATE": "2026-01-01T00:00:00Z",
        "GIT_COMMITTER_DATE": "2026-01-01T00:00:00Z",
    }
    subprocess.run(["git", "init", "-q", "-b", "main"], cwd=path, check=True)
    subprocess.run(["git", "config", "user.email", "t@example.invalid"], cwd=path, check=True)
    subprocess.run(["git", "config", "user.name", "test"], cwd=path, check=True)
    (path / "README.md").write_text("# Demo\n\nHello world.\n")
    (path / "src.py").write_text("def hello():\n    return 'hi'\n")
    (path / "logo.bin").write_bytes(b"\x00\x01\x02\x03" * 64)  # binary
    subprocess.run(["git", "add", "-A"], cwd=path, check=True)
    subprocess.run(
        ["git", "commit", "-q", "-m", "initial"],
        cwd=path,
        check=True,
        env=env,
    )


def _commit_file(path: Path, name: str, body: str, msg: str) -> None:
    env = {
        **os.environ,
        "GIT_AUTHOR_NAME": "test",
        "GIT_AUTHOR_EMAIL": "t@example.invalid",
        "GIT_COMMITTER_NAME": "test",
        "GIT_COMMITTER_EMAIL": "t@example.invalid",
        "GIT_AUTHOR_DATE": "2026-02-01T00:00:00Z",
        "GIT_COMMITTER_DATE": "2026-02-01T00:00:00Z",
    }
    (path / name).write_text(body)
    subprocess.run(["git", "add", name], cwd=path, check=True)
    subprocess.run(
        ["git", "commit", "-q", "-m", msg], cwd=path, check=True, env=env
    )


def _mk_hg_repo(path: Path) -> None:
    path.mkdir(parents=True, exist_ok=True)
    subprocess.run(["hg", "init", "-q"], cwd=path, check=True)
    (path / ".hg" / "hgrc").write_text(
        "[ui]\nusername = test <t@example.invalid>\n"
    )
    (path / "README.md").write_text("# Demo\n\nHello hg.\n")
    (path / "src.py").write_text("def world():\n    return 'world'\n")
    subprocess.run(["hg", "add"], cwd=path, check=True)
    subprocess.run(
        ["hg", "commit", "-m", "initial", "-d", "1735689600 0"],
        cwd=path,
        check=True,
    )


# ---------------------------------------------------------------------------
# GitRepoSource
# ---------------------------------------------------------------------------


@pytest.mark.skipif(not _git_available(), reason="git not on PATH")
def test_git_source_yields_text_files_and_skips_binary(tmp_path):
    repo = tmp_path / "demo-repo"
    _mk_git_repo(repo)
    src = GitRepoSource(repo)
    docs = list(src.iter_documents())
    titles = sorted(d.title for d in docs)
    # Binary file (logo.bin) should not appear.
    assert titles == ["README.md", "src.py"]
    readme = next(d for d in docs if d.title == "README.md")
    assert readme.uri == "git://demo-repo/file/README.md"
    assert "Hello world" in readme.content
    assert readme.source_type == "git_repo"
    # Commit metadata is captured (informational; not Merkle-bound).
    assert "commit_hash" in readme.extra
    assert len(readme.extra["commit_hash"]) == 40


@pytest.mark.skipif(not _git_available(), reason="git not on PATH")
def test_git_source_round_trip_and_supersedes_chain(tmp_path):
    repo = tmp_path / "demo-repo"
    _mk_git_repo(repo)
    db = tmp_path / "aborist.db"

    # First ingest: 2 docs (README.md, src.py).
    conn = connect(db)
    try:
        first = ingest_source(conn, GitRepoSource(repo))
    finally:
        conn.close()
    assert first.inserted == 2

    # Modify src.py and re-ingest. Same URI, new content -> new document_root
    # plus an automatic `supersedes` edge to the prior root.
    _commit_file(repo, "src.py", "def hello():\n    return 'updated'\n", "edit src")
    conn = connect(db)
    try:
        second = ingest_source(conn, GitRepoSource(repo))
    finally:
        conn.close()
    # README.md is unchanged so its content_root collides -> skipped.
    # src.py changed -> 1 new doc inserted with a supersedes edge.
    assert second.inserted == 1
    assert second.skipped_duplicate >= 1

    conn = connect(db)
    try:
        edges = conn.execute(
            "SELECT src_root, dst_root FROM edges WHERE edge_type='supersedes'"
        ).fetchall()
    finally:
        conn.close()
    assert len(edges) == 1


@pytest.mark.skipif(not _git_available(), reason="git not on PATH")
def test_git_source_rejects_non_git_path(tmp_path):
    with pytest.raises(FileNotFoundError):
        GitRepoSource(tmp_path)  # tmp_path has no .git


# ---------------------------------------------------------------------------
# MercurialRepoSource
# ---------------------------------------------------------------------------


@pytest.mark.skipif(not _hg_available(), reason="hg not on PATH")
def test_hg_source_yields_text_files(tmp_path):
    repo = tmp_path / "demo-hg"
    _mk_hg_repo(repo)
    src = MercurialRepoSource(repo)
    docs = list(src.iter_documents())
    titles = sorted(d.title for d in docs)
    assert titles == ["README.md", "src.py"]
    readme = next(d for d in docs if d.title == "README.md")
    assert readme.uri == "hg://demo-hg/file/README.md"
    assert "Hello hg" in readme.content
    assert readme.source_type == "hg_repo"
    assert "changeset_hash" in readme.extra


@pytest.mark.skipif(not _hg_available(), reason="hg not on PATH")
def test_hg_source_rejects_non_hg_path(tmp_path):
    with pytest.raises(FileNotFoundError):
        MercurialRepoSource(tmp_path)