test_commit_observer.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tests/tests_git/test_commit_observer.py
 
"""
Purpose
-------
Validate the B1.0 commit observer that detects silent mutations of
tracked-but-unstaged files during ``oct git commit``.
 
Responsibilities
----------------
- Unit tests for ``snapshot_unstaged_tracked`` and ``diff_snapshots``
  in ``oct.git.commit_observer``: snapshot excludes staged files,
  detects content drift, handles oversized files via sentinel.
- Integration tests for the warn/block modes wired into
  ``git_commit_cmd``: a synthetic mutation between snapshot points
  produces the expected stderr warning, and the block mode aborts
  with exit code 6.
 
Diagnostics
-----------
Domain: GIT-TESTS
Levels:
    L2 -- test lifecycle
    L3 -- assertion details
 
Contracts
---------
- All tests use subprocess invocation against ``oct.cli`` for
  integration-level checks; unit tests import the helper directly.
- All tests use tmp_path fixtures so they are automatically cleaned up.
"""
 
import json
import shutil
import subprocess
import sys
from pathlib import Path
 
import pytest
 
from oct.git.commit_observer import (
    _MAX_HASH_BYTES,
    _OVERSIZE_SENTINEL,
    diff_snapshots,
    snapshot_unstaged_tracked,
)
 
 
def _git_available() -> bool:
    return shutil.which("git") is not None
 
 
def _git(repo: Path, *args: str, check: bool = True):
    return subprocess.run(
        ["git", *args], cwd=repo, check=check,
        capture_output=True, text=True,
    )
 
 
@pytest.fixture
def git_repo(tmp_path: Path) -> Path:
    """Bare-bones git repo for unit-level snapshot tests."""
    if not _git_available():
        pytest.skip("git binary not available on PATH")
    repo = tmp_path / "repo"
    repo.mkdir()
    _git(repo, "init", "--quiet", "-b", "main")
    _git(repo, "config", "user.email", "test@example.com")
    _git(repo, "config", "user.name", "Test")
    _git(repo, "config", "commit.gpgsign", "false")
    return repo.resolve()
 
 
# =====================================================================
# snapshot_unstaged_tracked
# =====================================================================
 
 
class TestSnapshot:
 
    @pytest.mark.skipif(not _git_available(), reason="git not on PATH")
    def test_excludes_staged_files(self, git_repo: Path) -> None:
        (git_repo / "tracked.txt").write_text("hello\n", encoding="utf-8")
        (git_repo / "staged.txt").write_text("world\n", encoding="utf-8")
        _git(git_repo, "add", "tracked.txt", "staged.txt")
        _git(git_repo, "commit", "-m", "init", "--quiet")
 
        # Re-stage one file (modify + stage).
        (git_repo / "staged.txt").write_text("modified\n", encoding="utf-8")
        _git(git_repo, "add", "staged.txt")
 
        snap = snapshot_unstaged_tracked(
            git_repo, staged_set={"staged.txt"},
        )
        assert "tracked.txt" in snap
        assert "staged.txt" not in snap
 
    @pytest.mark.skipif(not _git_available(), reason="git not on PATH")
    def test_excludes_untracked(self, git_repo: Path) -> None:
        (git_repo / "tracked.txt").write_text("x\n", encoding="utf-8")
        _git(git_repo, "add", "tracked.txt")
        _git(git_repo, "commit", "-m", "init", "--quiet")
        (git_repo / "untracked.txt").write_text("y\n", encoding="utf-8")
 
        snap = snapshot_unstaged_tracked(git_repo)
        assert "tracked.txt" in snap
        assert "untracked.txt" not in snap
 
    @pytest.mark.skipif(not _git_available(), reason="git not on PATH")
    def test_oversize_file_uses_sentinel(self, git_repo: Path) -> None:
        big = git_repo / "big.bin"
        big.write_bytes(b"x" * (_MAX_HASH_BYTES + 1))
        _git(git_repo, "add", "big.bin")
        _git(git_repo, "commit", "-m", "init", "--quiet")
 
        snap = snapshot_unstaged_tracked(git_repo)
        assert snap["big.bin"] == _OVERSIZE_SENTINEL
 
 
# =====================================================================
# diff_snapshots
# =====================================================================
 
 
class TestDiffSnapshots:
 
    def test_no_change(self) -> None:
        before = {"a.txt": "h1", "b.txt": "h2"}
        after = {"a.txt": "h1", "b.txt": "h2"}
        assert diff_snapshots(before, after) == []
 
    def test_content_change(self) -> None:
        before = {"a.txt": "h1"}
        after = {"a.txt": "h1-CHANGED"}
        assert diff_snapshots(before, after) == ["a.txt"]
 
    def test_deletion(self) -> None:
        before = {"a.txt": "h1", "b.txt": "h2"}
        after = {"a.txt": "h1"}
        assert diff_snapshots(before, after) == ["b.txt"]
 
    def test_oversize_pair_is_ignored(self) -> None:
        # Two oversize files that we cannot compare must NOT register
        # as a mutation.
        before = {"big.bin": _OVERSIZE_SENTINEL}
        after = {"big.bin": _OVERSIZE_SENTINEL}
        assert diff_snapshots(before, after) == []
 
    def test_change_to_or_from_oversize_is_detected(self) -> None:
        before = {"big.bin": "abcd"}
        after = {"big.bin": _OVERSIZE_SENTINEL}
        assert diff_snapshots(before, after) == ["big.bin"]
 
 
# =====================================================================
# Integration: warn mode on a real oct git commit
# =====================================================================
 
 
def _run_commit(
    project_root: Path, *extra_args: str, env: dict | None = None,
) -> subprocess.CompletedProcess:
    return subprocess.run(
        [
            sys.executable, "-m", "oct.cli", "git", "commit",
            *extra_args,
        ],
        cwd=project_root,
        capture_output=True,
        text=True,
        env=env,
    )
 
 
class TestObserverWarnDefault:
    """Default ``warn`` mode prints a warning but does not block."""
 
    @pytest.mark.skipif(not _git_available(), reason="git not on PATH")
    def test_clean_commit_no_warning(
        self, temp_git_project, make_compliant_py,
    ) -> None:
        path = make_compliant_py("src/clean.py")
        _git(temp_git_project, "add", "src/clean.py")
        result = _run_commit(temp_git_project, "-m", "feat: add clean")
        assert result.returncode == 0, result.stderr
        assert "mutated during" not in result.stderr
        assert "tracked file mutation" not in result.stderr
 
    @pytest.mark.skipif(not _git_available(), reason="git not on PATH")
    def test_no_block_on_warn_mode(
        self, temp_git_project, make_compliant_py,
    ) -> None:
        # Stage one compliant file, leave another tracked file modified
        # but NOT staged. The observer should NOT trip because the
        # mutation pre-existed the commit; it only flags mutations that
        # happen *between* the start and end snapshots.
        a = make_compliant_py("src/a.py")
        b = make_compliant_py("src/b.py")
        _git(temp_git_project, "add", "src/a.py", "src/b.py")
        _git(temp_git_project, "commit", "-m", "init", "--quiet")
 
        b.write_text(
            b.read_text(encoding="utf-8") + "\n# drift\n",
            encoding="utf-8",
        )
        a.write_text(
            a.read_text(encoding="utf-8") + "\n# new\n",
            encoding="utf-8",
        )
        _git(temp_git_project, "add", "src/a.py")
 
        result = _run_commit(temp_git_project, "-m", "feat: bump a")
        assert result.returncode == 0, result.stderr
        # Pre-existing drift on b.py is NOT a mid-commit mutation.
        assert "mutated during" not in result.stderr
 
 
# =====================================================================
# Integration: block mode aborts with exit 6 on a manufactured mutation
# =====================================================================
 
 
class TestObserverBlock:
 
    @pytest.mark.skipif(not _git_available(), reason="git not on PATH")
    def test_block_mode_via_octrc(
        self, temp_git_project, make_compliant_py,
    ) -> None:
        # Configure block mode in octrc.
        octrc_path = temp_git_project / ".octrc.json"
        cfg = json.loads(octrc_path.read_text(encoding="utf-8"))
        cfg.setdefault("git", {})["observe_unstaged_mutation"] = "block"
        octrc_path.write_text(
            json.dumps(cfg, indent=2), encoding="utf-8",
        )
 
        # Build two committed files; stage A; mutate B mid-commit via
        # a pre-commit hook that touches it.
        a = make_compliant_py("src/a.py")
        b = make_compliant_py("src/b.py")
        _git(temp_git_project, "add", "src/a.py", "src/b.py")
        _git(temp_git_project, "commit", "-m", "init", "--quiet")
 
        # Install a real git pre-commit hook that mutates b.py.
        hooks_dir = temp_git_project / ".git" / "hooks"
        hooks_dir.mkdir(parents=True, exist_ok=True)
        hook = hooks_dir / "pre-commit"
        rel_b = (b.relative_to(temp_git_project)).as_posix()
        hook.write_text(
            "#!/bin/sh\n"
            f'printf "\\n# injected by hook\\n" >> "{rel_b}"\n',
            encoding="utf-8",
        )
        try:
            hook.chmod(0o755)
        except OSError:
            pass
 
        a.write_text(
            a.read_text(encoding="utf-8") + "\n# bump\n",
            encoding="utf-8",
        )
        _git(temp_git_project, "add", "src/a.py")
 
        result = _run_commit(temp_git_project, "-m", "feat: bump a")
        # On Windows, /bin/sh hooks may not execute; tolerate that case.
        if result.returncode == 0 and "mutated during" not in result.stderr:
            pytest.skip("git pre-commit hook did not fire on this platform")
 
        assert result.returncode == 6, (
            f"expected exit 6 (observer block), got "
            f"{result.returncode}: {result.stderr}"
        )
        assert "mutated during" in result.stderr.lower() \
            or "tracked-file mutation" in result.stderr.lower()
 
 
# =====================================================================
# Override: --ignore-unstaged-mutation downgrades block to warn
# =====================================================================
 
 
class TestObserverOverride:
 
    @pytest.mark.skipif(not _git_available(), reason="git not on PATH")
    def test_flag_downgrades_block_to_warn(
        self, temp_git_project, make_compliant_py,
    ) -> None:
        octrc_path = temp_git_project / ".octrc.json"
        cfg = json.loads(octrc_path.read_text(encoding="utf-8"))
        cfg.setdefault("git", {})["observe_unstaged_mutation"] = "block"
        octrc_path.write_text(
            json.dumps(cfg, indent=2), encoding="utf-8",
        )
 
        a = make_compliant_py("src/a.py")
        _git(temp_git_project, "add", "src/a.py")
        # No mid-commit mutation here — flag should be a no-op when the
        # observer didn't fire. Just verify it parses and the commit
        # still succeeds.
        result = _run_commit(
            temp_git_project, "-m", "feat: add a",
            "--ignore-unstaged-mutation",
        )
        assert result.returncode == 0, result.stderr