test_git.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_core/test_git.py
 
"""
Purpose
-------
Validate the V4.0 Phase 4A Git integration foundation in
``oct.core.git``: exception hierarchy, subprocess discipline,
credential redaction, and the 11 read-only state query functions.
 
Responsibilities
----------------
- Confirm the ``GitError`` hierarchy (G-A1).
- Confirm ``git_run`` enforces argv-only, timeout, text mode, utf-8
  with ``errors="replace"``, and translates subprocess failures to
  ``GitNotFoundError`` / ``GitTimeoutError`` / ``GitCommandError``
  (G-A2).
- Confirm every query function behaves correctly against a real git
  repo, raising the right error on missing repo / missing binary
  (G-A3).
- Confirm path-escape filtering in ``git_changed_files`` and
  ``git_staged_files`` (G-A3 + G-A4 defense-in-depth).
- Confirm ``_redact_git_url`` scrubs http(s) credentials while leaving
  SSH URLs intact (G-A5).
- Confirm ``_git_changed_files`` shim in the linter continues to
  return ``[]`` on all ``GitError`` subtypes (G-A7).
 
Diagnostics
-----------
Domain: CORE-TESTS
Levels:
    L2 — test lifecycle
    L3 — per-case setup
    L4 — fine-grained assertions
 
Contracts
---------
- Tests never write outside of ``tmp_path`` / pytest-managed temp dirs.
- Tests that require a real ``git`` binary are skipped via the
  ``git_repo`` fixture when git is missing.
- Symlink tests skip when the platform cannot create symlinks.
"""
 
import os
import subprocess
import tempfile
from pathlib import Path
from unittest.mock import patch
 
import pytest
 
 
def _platform_can_symlink() -> bool:
    """Probe whether this process can create a real OS symlink.
 
    Equivalent to the session-scoped ``can_symlink`` fixture in
    ``tests_core/conftest.py`` but evaluated at module import so the
    result is usable in ``@pytest.mark.skipif`` decorators (which run
    at collection time and cannot consume runtime fixtures). On
    Windows without admin elevation or developer-mode this returns
    ``False``; on POSIX it returns ``True`` unless the filesystem
    refuses the syscall (e.g. some FAT mounts).
    """
    try:
        with tempfile.TemporaryDirectory() as d:
            target = Path(d) / "target.txt"
            link = Path(d) / "link.txt"
            target.write_text("x", encoding="utf-8")
            os.symlink(target, link)
            return link.exists()
    except (OSError, NotImplementedError):
        return False
 
 
_CAN_SYMLINK: bool = _platform_can_symlink()
_SYMLINK_SKIP_REASON: str = (
    "Platform cannot create symlinks without elevation "
    "(Windows non-admin / non-developer-mode, or FS refuses the syscall)."
)
 
from oct.core import git as git_mod
from oct.core.git import (
    DEFAULT_GIT_TIMEOUT,
    GitCommandError,
    GitError,
    GitNotARepoError,
    GitNotFoundError,
    GitTimeoutError,
    _redact_argv,
    _redact_git_url,
    current_branch,
    get_repo_root,
    git_changed_files,
    git_run,
    git_staged_files,
    has_uncommitted_changes,
    has_untracked_files,
    is_detached_head,
    is_git_repo,
    last_commit_message,
    last_commit_sha,
    remote_urls,
)
from oct.core.project_root import is_within_project_root
 
 
# ============================================================
# G-A1: Exception hierarchy
# ============================================================
 
 
def test_git_error_is_base_of_all_subclasses():
    assert issubclass(GitNotFoundError, GitError)
    assert issubclass(GitNotARepoError, GitError)
    assert issubclass(GitTimeoutError, GitError)
    assert issubclass(GitCommandError, GitError)
 
 
def test_git_command_error_preserves_context():
    err = GitCommandError(
        ["git", "status"], returncode=128,
        stdout="out", stderr="fatal: not a repo",
    )
    assert err.returncode == 128
    assert err.stdout == "out"
    assert err.stderr == "fatal: not a repo"
    assert err.argv == ["git", "status"]
    assert "128" in str(err)
    assert "status" in str(err)
 
 
# ============================================================
# G-A5: Credential redaction
# ============================================================
 
 
def test_redact_strips_http_creds():
    out = _redact_git_url("clone http://user:pass@example.com/r.git ok")
    assert "user:pass" not in out
    assert "http://***@example.com/r.git" in out
 
 
def test_redact_strips_https_creds():
    out = _redact_git_url("https://alice:s3cr3t@github.com/x/y.git")
    assert "alice" not in out
    assert "s3cr3t" not in out
    assert "https://***@github.com/x/y.git" in out
 
 
def test_redact_preserves_ssh_urls():
    url = "git@github.com:user/repo.git"
    assert _redact_git_url(url) == url
 
 
def test_redact_handles_empty_string():
    assert _redact_git_url("") == ""
 
 
def test_redact_handles_none_safely():
    # None-like falsy case: empty string is the only contract; None
    # would not come from subprocess stdout but we still guard it.
    assert _redact_git_url("") == ""
 
 
def test_redact_multiline_output():
    text = (
        "origin\thttps://u:t@github.com/x/y.git (fetch)\n"
        "origin\thttps://u:t@github.com/x/y.git (push)\n"
    )
    out = _redact_git_url(text)
    assert "u:t" not in out
    assert out.count("***@github.com") == 2
 
 
def test_redact_preserves_non_url_content():
    text = "some random commit message mentioning @user and email@host"
    # No scheme prefix → nothing to redact.
    assert _redact_git_url(text) == text
 
 
# ============================================================
# OI-507: _redact_argv + credential-safe _dbg/TimeoutExpired/GitCommandError
# ============================================================
 
 
_TOKEN_URL = "https://oauth2:ghp_SUPERSECRET@github.com/foo/bar.git"
 
 
def test_redact_argv_redacts_url_elements():
    argv = ["git", "push", _TOKEN_URL]
    out = _redact_argv(argv)
    assert "ghp_SUPERSECRET" not in " ".join(out)
    assert "oauth2" not in " ".join(out)
    assert "***@github.com" in out[-1]
 
 
def test_redact_argv_noop_for_plain_argv():
    argv = ["git", "status", "--porcelain"]
    assert _redact_argv(argv) == argv
 
 
def test_git_run_dbg_does_not_leak_token(tmp_path, monkeypatch, caplog):
    """_dbg logs argv before the subprocess runs; it must be redacted."""
    captured: dict[str, str] = {}
 
    def fake_dbg(domain, func, msg, level, *args, **kwargs):
        captured.setdefault(f"{domain}:{func}:{level}", "")
        captured[f"{domain}:{func}:{level}"] += str(msg) + "\n"
 
    def fake_run(*a, **kw):
        return _FakeCompleted(returncode=0, stdout="", stderr="")
 
    monkeypatch.setattr(git_mod, "_dbg", fake_dbg)
    monkeypatch.setattr(subprocess, "run", fake_run)
    git_run(["push", _TOKEN_URL], cwd=tmp_path)
 
    combined = "\n".join(captured.values())
    assert "ghp_SUPERSECRET" not in combined
    assert "oauth2" not in combined
 
 
def test_git_command_error_str_redacts_token_in_argv():
    err = GitCommandError(
        ["git", "push", _TOKEN_URL],
        returncode=1,
        stdout="",
        stderr="fatal: auth failed",
    )
    rendered = str(err)
    assert "ghp_SUPERSECRET" not in rendered
    assert "oauth2" not in rendered
    # .argv attribute must also be redacted.
    assert "ghp_SUPERSECRET" not in " ".join(err.argv)
 
 
def test_timeout_message_redacts_token(tmp_path, monkeypatch):
    def fake_run(*a, **kw):
        raise subprocess.TimeoutExpired(cmd="git", timeout=1)
 
    monkeypatch.setattr(subprocess, "run", fake_run)
    with pytest.raises(GitTimeoutError) as exc_info:
        git_run(["push", _TOKEN_URL], cwd=tmp_path, timeout=1)
 
    rendered = str(exc_info.value)
    assert "ghp_SUPERSECRET" not in rendered
    assert "oauth2" not in rendered
 
 
# ============================================================
# G-A2: git_run subprocess wrapper (mocked)
# ============================================================
 
 
class _FakeCompleted:
    def __init__(self, returncode=0, stdout="", stderr=""):
        self.returncode = returncode
        self.stdout = stdout
        self.stderr = stderr
 
 
def test_git_run_rejects_non_list_argv(tmp_path):
    with pytest.raises(GitError):
        git_run("status", cwd=tmp_path)  # type: ignore[arg-type]
 
 
def test_git_run_happy_path(tmp_path, monkeypatch):
    calls = {}
 
    def fake_run(argv, **kwargs):
        calls["argv"] = argv
        calls["kwargs"] = kwargs
        return _FakeCompleted(returncode=0, stdout="main\n", stderr="")
 
    monkeypatch.setattr(subprocess, "run", fake_run)
    proc = git_run(["rev-parse", "--abbrev-ref", "HEAD"], cwd=tmp_path)
    assert proc.stdout == "main\n"
    assert calls["argv"][0] == "git"
    assert calls["kwargs"]["shell"] is False
    assert calls["kwargs"]["text"] is True
    assert calls["kwargs"]["encoding"] == "utf-8"
    assert calls["kwargs"]["errors"] == "replace"
    assert calls["kwargs"]["timeout"] == DEFAULT_GIT_TIMEOUT
 
 
def test_git_run_raises_git_not_found(tmp_path, monkeypatch):
    def fake_run(*a, **kw):
        raise FileNotFoundError("git")
    monkeypatch.setattr(subprocess, "run", fake_run)
    with pytest.raises(GitNotFoundError):
        git_run(["status"], cwd=tmp_path)
 
 
def test_git_run_raises_timeout(tmp_path, monkeypatch):
    def fake_run(*a, **kw):
        raise subprocess.TimeoutExpired(cmd="git", timeout=1)
    monkeypatch.setattr(subprocess, "run", fake_run)
    with pytest.raises(GitTimeoutError):
        git_run(["status"], cwd=tmp_path, timeout=1)
 
 
def test_git_run_raises_on_nonzero_exit(tmp_path, monkeypatch):
    def fake_run(*a, **kw):
        return _FakeCompleted(returncode=128, stdout="", stderr="fatal: boom")
    monkeypatch.setattr(subprocess, "run", fake_run)
    with pytest.raises(GitCommandError) as exc_info:
        git_run(["status"], cwd=tmp_path)
    assert exc_info.value.returncode == 128
    assert "boom" in exc_info.value.stderr
 
 
def test_git_run_nocheck_returns_process(tmp_path, monkeypatch):
    def fake_run(*a, **kw):
        return _FakeCompleted(returncode=1, stdout="", stderr="")
    monkeypatch.setattr(subprocess, "run", fake_run)
    proc = git_run(["rev-parse"], cwd=tmp_path, check=False)
    assert proc.returncode == 1
 
 
def test_git_run_redacts_stderr_on_error(tmp_path, monkeypatch):
    def fake_run(*a, **kw):
        return _FakeCompleted(
            returncode=1, stdout="",
            stderr="fatal: could not read https://bob:token@host/x.git\n",
        )
    monkeypatch.setattr(subprocess, "run", fake_run)
    with pytest.raises(GitCommandError) as exc_info:
        git_run(["fetch"], cwd=tmp_path)
    assert "bob" not in exc_info.value.stderr
    assert "token" not in exc_info.value.stderr
    assert "***@host" in exc_info.value.stderr
 
 
def test_git_run_redacts_stdout_on_success(tmp_path, monkeypatch):
    def fake_run(*a, **kw):
        return _FakeCompleted(
            returncode=0,
            stdout="origin\thttps://u:p@github.com/x.git (fetch)\n",
            stderr="",
        )
    monkeypatch.setattr(subprocess, "run", fake_run)
    proc = git_run(["remote", "-v"], cwd=tmp_path)
    assert "u:p" not in proc.stdout
    assert "***@github.com" in proc.stdout
 
 
# ============================================================
# G-A3: is_git_repo (never-raise)
# ============================================================
 
 
def test_is_git_repo_true_inside_repo(git_repo):
    assert is_git_repo(git_repo) is True
 
 
def test_is_git_repo_false_outside_repo(tmp_path):
    assert is_git_repo(tmp_path) is False
 
 
def test_is_git_repo_false_when_git_missing(tmp_path, monkeypatch):
    def fake_run(*a, **kw):
        raise FileNotFoundError
    monkeypatch.setattr(subprocess, "run", fake_run)
    assert is_git_repo(tmp_path) is False
 
 
def test_is_git_repo_false_on_any_exception(tmp_path, monkeypatch):
    def fake_run(*a, **kw):
        raise OSError("permission denied")
    monkeypatch.setattr(subprocess, "run", fake_run)
    assert is_git_repo(tmp_path) is False
 
 
# ============================================================
# G-A3: query functions (real git integration)
# ============================================================
 
 
def test_get_repo_root_returns_absolute_path(git_repo):
    assert get_repo_root(git_repo) == git_repo
 
 
def test_get_repo_root_raises_not_a_repo(tmp_path):
    with pytest.raises(GitNotARepoError):
        get_repo_root(tmp_path)
 
 
def test_current_branch_returns_name(git_repo):
    from tests.tests_core.conftest import commit_file
    commit_file(git_repo, "a.txt", "hello", "init")
    assert current_branch(git_repo) == "main"
 
 
def test_is_detached_head_false_on_branch(git_repo):
    from tests.tests_core.conftest import commit_file
    commit_file(git_repo, "a.txt", "hello", "init")
    assert is_detached_head(git_repo) is False
 
 
def test_is_detached_head_true_after_checkout_sha(git_repo):
    from tests.tests_core.conftest import commit_file
    commit_file(git_repo, "a.txt", "hello", "init")
    sha = last_commit_sha(git_repo)
    subprocess.run(
        ["git", "checkout", "--quiet", sha],
        cwd=git_repo, check=True, capture_output=True, text=True,
    )
    assert is_detached_head(git_repo) is True
 
 
def test_has_uncommitted_changes_false_on_clean_repo(git_repo):
    from tests.tests_core.conftest import commit_file
    commit_file(git_repo, "a.txt", "hello", "init")
    assert has_uncommitted_changes(git_repo) is False
 
 
def test_has_uncommitted_changes_true_after_edit(git_repo):
    from tests.tests_core.conftest import commit_file
    commit_file(git_repo, "a.txt", "hello", "init")
    (git_repo / "a.txt").write_text("changed", encoding="utf-8")
    assert has_uncommitted_changes(git_repo) is True
 
 
def test_has_untracked_files_false_on_fresh_repo(git_repo):
    assert has_untracked_files(git_repo) is False
 
 
def test_has_untracked_files_true_for_new_file(git_repo):
    (git_repo / "untracked.txt").write_text("x", encoding="utf-8")
    assert has_untracked_files(git_repo) is True
 
 
def test_git_changed_files_returns_absolute_paths(git_repo):
    (git_repo / "new.py").write_text("print()", encoding="utf-8")
    result = git_changed_files(git_repo)
    assert len(result) >= 1
    for p in result:
        assert p.is_absolute()
 
 
def test_git_changed_files_filters_by_extension(git_repo):
    (git_repo / "new.py").write_text("print()", encoding="utf-8")
    (git_repo / "other.txt").write_text("x", encoding="utf-8")
    py_only = git_changed_files(git_repo, extensions={".py"})
    assert any(p.suffix == ".py" for p in py_only)
    assert not any(p.suffix == ".txt" for p in py_only)
 
 
def test_git_changed_files_empty_on_clean_repo(git_repo):
    from tests.tests_core.conftest import commit_file
    commit_file(git_repo, "a.txt", "hello", "init")
    assert git_changed_files(git_repo) == []
 
 
def test_git_staged_files_only_returns_staged(git_repo):
    (git_repo / "staged.py").write_text("print()", encoding="utf-8")
    (git_repo / "unstaged.py").write_text("print()", encoding="utf-8")
    subprocess.run(
        ["git", "add", "staged.py"],
        cwd=git_repo, check=True, capture_output=True, text=True,
    )
    staged = git_staged_files(git_repo)
    names = {p.name for p in staged}
    assert "staged.py" in names
    assert "unstaged.py" not in names
 
 
def test_last_commit_sha_empty_on_fresh_repo(git_repo):
    assert last_commit_sha(git_repo) == ""
 
 
def test_last_commit_sha_full_and_short(git_repo):
    from tests.tests_core.conftest import commit_file
    commit_file(git_repo, "a.txt", "hello", "init")
    full = last_commit_sha(git_repo)
    short = last_commit_sha(git_repo, short=True)
    assert len(full) == 40
    assert len(short) < len(full)
    assert full.startswith(short)
 
 
def test_last_commit_message_empty_on_fresh_repo(git_repo):
    assert last_commit_message(git_repo) == ""
 
 
def test_last_commit_message_returns_subject(git_repo):
    from tests.tests_core.conftest import commit_file
    commit_file(git_repo, "a.txt", "hello", "initial commit")
    assert "initial commit" in last_commit_message(git_repo)
 
 
def test_remote_urls_empty_on_fresh_repo(git_repo):
    assert remote_urls(git_repo) == {}
 
 
def test_remote_urls_redacts_http_creds(git_repo):
    subprocess.run(
        [
            "git", "remote", "add", "origin",
            "https://user:token@example.com/x/y.git",
        ],
        cwd=git_repo, check=True, capture_output=True, text=True,
    )
    urls = remote_urls(git_repo)
    assert "origin" in urls
    assert "user" not in urls["origin"]
    assert "token" not in urls["origin"]
    assert "***@example.com" in urls["origin"]
 
 
# ============================================================
# G-A4: is_within_project_root predicate
# ============================================================
 
 
def test_is_within_project_root_true_for_child(tmp_path):
    child = tmp_path / "sub" / "file.txt"
    child.parent.mkdir()
    child.write_text("x", encoding="utf-8")
    assert is_within_project_root(child, tmp_path) is True
 
 
def test_is_within_project_root_true_for_self(tmp_path):
    assert is_within_project_root(tmp_path, tmp_path) is True
 
 
def test_is_within_project_root_false_for_sibling(tmp_path):
    sibling = tmp_path.parent / "outside-of-tmp"
    assert is_within_project_root(sibling, tmp_path) is False
 
 
def test_is_within_project_root_handles_nonexistent_paths(tmp_path):
    # Nonexistent children of tmp_path still resolve as inside.
    ghost = tmp_path / "does" / "not" / "exist.py"
    assert is_within_project_root(ghost, tmp_path) is True
 
 
@pytest.mark.skipif(not _CAN_SYMLINK, reason=_SYMLINK_SKIP_REASON)
def test_is_within_project_root_false_for_symlink_escape(tmp_path):
    """Symlinks pointing outside the project root must not be accepted.
 
    Skipped on platforms without symlink capability — this is an
    environmental constraint, not a defect in the tested behaviour.
    """
    repo = tmp_path / "repo"
    repo.mkdir()
    outside = tmp_path / "outside.txt"
    outside.write_text("secret", encoding="utf-8")
    link = repo / "link.txt"
    os.symlink(outside, link)
    assert is_within_project_root(link, repo) is False
 
 
@pytest.mark.skipif(not _CAN_SYMLINK, reason=_SYMLINK_SKIP_REASON)
def test_git_changed_files_filters_out_path_escapes(git_repo, tmp_path):
    """``git_changed_files`` must filter symlinks that escape the repo.
 
    Skipped on platforms without symlink capability — this is an
    environmental constraint, not a defect in the tested behaviour.
    """
    outside = tmp_path / "outside_secret.txt"
    outside.write_text("secret", encoding="utf-8")
    link = git_repo / "link.txt"
    os.symlink(outside, link)
    result = git_changed_files(git_repo)
    assert all("outside_secret" not in str(p) for p in result)
 
 
# ============================================================
# G-A7: legacy shim (linter wrapper)
# ============================================================
 
 
def test_legacy_shim_returns_empty_on_missing_git(tmp_path, monkeypatch):
    from oct.linter import oct_lint
    from oct.core import git as core_git
 
    def fake_git_changed_files(*a, **kw):
        raise GitNotFoundError("no git")
 
    monkeypatch.setattr(core_git, "git_changed_files", fake_git_changed_files)
    assert oct_lint._git_changed_files(tmp_path) == []
 
 
def test_legacy_shim_returns_empty_on_not_a_repo(tmp_path, monkeypatch):
    from oct.linter import oct_lint
    from oct.core import git as core_git
 
    def fake_git_changed_files(*a, **kw):
        raise GitNotARepoError("nope")
 
    monkeypatch.setattr(core_git, "git_changed_files", fake_git_changed_files)
    assert oct_lint._git_changed_files(tmp_path) == []
 
 
def test_legacy_shim_returns_empty_on_command_error(tmp_path, monkeypatch):
    from oct.linter import oct_lint
    from oct.core import git as core_git
 
    def fake_git_changed_files(*a, **kw):
        raise GitCommandError(
            ["git", "diff"], returncode=128, stdout="", stderr="boom",
        )
 
    monkeypatch.setattr(core_git, "git_changed_files", fake_git_changed_files)
    assert oct_lint._git_changed_files(tmp_path) == []
 
 
def test_legacy_shim_passes_py_extension_filter(tmp_path, monkeypatch):
    from oct.linter import oct_lint
    from oct.core import git as core_git
 
    received = {}
 
    def fake_git_changed_files(repo_root, extensions=None):
        received["extensions"] = extensions
        return []
 
    monkeypatch.setattr(core_git, "git_changed_files", fake_git_changed_files)
    oct_lint._git_changed_files(tmp_path)
    assert received["extensions"] == {".py"}