From e00f0251933289338f709a94e37fb0d4d3beaeda Mon Sep 17 00:00:00 2001 From: autocommit Date: Sat, 18 Apr 2026 11:38:36 -0700 Subject: [PATCH] =?UTF-8?q?test(rebase-recovery):=20=E2=9C=85=20Add=20test?= =?UTF-8?q?=20case=20for=20rebase=20recovery=20failure=20handling=20and=20?= =?UTF-8?q?validation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Lilith Autocommit --- tests/test_rebase_recovery.py | 205 ++++++++++++++++++++++++++++++++++ 1 file changed, 205 insertions(+) create mode 100644 tests/test_rebase_recovery.py diff --git a/tests/test_rebase_recovery.py b/tests/test_rebase_recovery.py new file mode 100644 index 0000000..ba5e784 --- /dev/null +++ b/tests/test_rebase_recovery.py @@ -0,0 +1,205 @@ +"""Tests for rebase/merge orphan-state recovery logic.""" + +import subprocess +from pathlib import Path +from unittest.mock import AsyncMock, patch + +import pytest + +from auto_commit_service.git.operations import ( + _abort_rebase_verified, + pre_cycle_recover, + pre_cycle_sync, +) + + +def _git(args: list[str], cwd: Path) -> subprocess.CompletedProcess: + return subprocess.run(["git"] + args, cwd=cwd, check=True, capture_output=True) + + +def _make_conflicted_rebase(tmp_path: Path) -> Path: + """Return a repo stuck mid-rebase with a conflict. + + Sets up two divergent branches on the same file, starts rebase, + which halts at the conflict — leaving .git/rebase-merge in place. + """ + repo = tmp_path / "repo" + repo.mkdir() + _git(["init"], repo) + _git(["config", "user.email", "test@test.com"], repo) + _git(["config", "user.name", "Test User"], repo) + + # Base commit + (repo / "file.txt").write_text("base\n") + _git(["add", "."], repo) + _git(["commit", "-m", "base"], repo) + + # Branch A: modify file.txt + _git(["checkout", "-b", "branch-a"], repo) + (repo / "file.txt").write_text("branch-a change\n") + _git(["add", "file.txt"], repo) + _git(["commit", "-m", "branch-a commit"], repo) + + # Back to master: diverge on same file + _git(["checkout", "master"], repo) + (repo / "file.txt").write_text("master change\n") + _git(["add", "file.txt"], repo) + _git(["commit", "-m", "master commit"], repo) + + # Start rebase of master onto branch-a — will conflict and halt + subprocess.run( + ["git", "rebase", "branch-a"], + cwd=repo, + capture_output=True, + # Non-zero expected — conflict halts rebase + ) + + assert (repo / ".git" / "rebase-merge").exists(), "Expected repo to be stuck in rebase" + return repo + + +class TestAbortRebaseVerified: + async def test_abort_rebase_verified_clean_path(self, tmp_path: Path) -> None: + repo = _make_conflicted_rebase(tmp_path) + orig_head_ref = (repo / ".git" / "ORIG_HEAD").read_text().strip() + + result = await _abort_rebase_verified(repo) + + assert result is True + assert not (repo / ".git" / "rebase-merge").exists() + assert not (repo / ".git" / "rebase-apply").exists() + # HEAD should be back at ORIG_HEAD (the pre-rebase tip) + head = subprocess.run( + ["git", "rev-parse", "HEAD"], + cwd=repo, capture_output=True, text=True, check=True, + ).stdout.strip() + assert head == orig_head_ref + + async def test_abort_rebase_verified_escalates_to_quit(self, tmp_path: Path) -> None: + """When --abort is a no-op (leaves state behind), escalate to --quit.""" + repo = _make_conflicted_rebase(tmp_path) + + abort_call_count = 0 + original_run = __import__( + "auto_commit_service.git.operations", fromlist=["_run_git_command"] + )._run_git_command + + async def patched_run(*args, **kwargs): + nonlocal abort_call_count + if args == ("rebase", "--abort") or list(args) == ["rebase", "--abort"]: + abort_call_count += 1 + # First abort call: run the real command but then re-create state + # to simulate a partially-cleared scenario where --abort "succeeded" + # but didn't fully remove the directory. + result = await original_run(*args, **kwargs) + if abort_call_count == 1: + # Re-create the sentinel to simulate abort not cleaning up + rebase_merge = kwargs["cwd"] / ".git" / "rebase-merge" + rebase_merge.mkdir(exist_ok=True) + (rebase_merge / "head-name").write_text("refs/heads/master\n") + return result + return await original_run(*args, **kwargs) + + with patch( + "auto_commit_service.git.operations._run_git_command", + side_effect=patched_run, + ): + result = await _abort_rebase_verified(repo) + + # --quit path ran and cleaned up + assert result is True + assert not (repo / ".git" / "rebase-merge").exists() + + +class TestPreCycleRecover: + async def test_pre_cycle_recover_detects_rebase_state(self, tmp_path: Path) -> None: + repo = tmp_path / "repo" + repo.mkdir() + _git(["init"], repo) + _git(["config", "user.email", "test@test.com"], repo) + _git(["config", "user.name", "Test User"], repo) + (repo / "file.txt").write_text("content\n") + _git(["add", "."], repo) + _git(["commit", "-m", "initial"], repo) + + # Manually plant orphan rebase state + rebase_merge = repo / ".git" / "rebase-merge" + rebase_merge.mkdir() + (rebase_merge / "head-name").write_text("refs/heads/master\n") + (rebase_merge / "onto").write_text("deadbeef\n") + + result = await pre_cycle_recover(repo) + + assert result["was_stuck"] is True + assert result["state"] == "rebase" + # Recovery may fail because ORIG_HEAD doesn't exist, but it must detect + assert "recovered" in result + + async def test_pre_cycle_recover_detects_merge_state(self, tmp_path: Path) -> None: + repo = tmp_path / "repo" + repo.mkdir() + _git(["init"], repo) + _git(["config", "user.email", "test@test.com"], repo) + _git(["config", "user.name", "Test User"], repo) + (repo / "file.txt").write_text("content\n") + _git(["add", "."], repo) + _git(["commit", "-m", "initial"], repo) + + # Plant orphan merge state + merge_head = repo / ".git" / "MERGE_HEAD" + merge_head.write_text("deadbeefdeadbeefdeadbeefdeadbeefdeadbeef\n") + + result = await pre_cycle_recover(repo) + + assert result["was_stuck"] is True + assert result["state"] == "merge" + # merge --abort removes MERGE_HEAD (git will do that even if hash is fake) + assert not merge_head.exists() + assert result["recovered"] is True + + async def test_pre_cycle_recover_clean_repo_noop(self, tmp_path: Path) -> None: + repo = tmp_path / "repo" + repo.mkdir() + _git(["init"], repo) + _git(["config", "user.email", "test@test.com"], repo) + _git(["config", "user.name", "Test User"], repo) + (repo / "file.txt").write_text("content\n") + _git(["add", "."], repo) + _git(["commit", "-m", "initial"], repo) + + result = await pre_cycle_recover(repo) + + assert result["was_stuck"] is False + assert result["state"] == "clean" + + +class TestPreCycleSyncRecoveryGating: + async def test_pre_cycle_sync_refuses_when_recovery_fails( + self, tmp_path: Path + ) -> None: + """pre_cycle_sync must return early with orphan_rebase_unrecovered if recovery fails.""" + repo = tmp_path / "repo" + repo.mkdir() + _git(["init"], repo) + _git(["config", "user.email", "test@test.com"], repo) + _git(["config", "user.name", "Test User"], repo) + (repo / "file.txt").write_text("content\n") + _git(["add", "."], repo) + _git(["commit", "-m", "initial"], repo) + + # Plant orphan rebase state that will survive all abort attempts + rebase_merge = repo / ".git" / "rebase-merge" + rebase_merge.mkdir() + (rebase_merge / "head-name").write_text("refs/heads/master\n") + + # Patch _abort_rebase_verified to always return False (unrecoverable) + with patch( + "auto_commit_service.git.operations._abort_rebase_verified", + new=AsyncMock(return_value=False), + ): + result = await pre_cycle_sync(repo, "origin", "master") + + assert result["error"] == "orphan_rebase_unrecovered" + assert result["fetched"] is False + assert result["pulled"] is False + assert result["recovered"] is False