test(conflict-resolution): ✅ Add comprehensive tests for regen timeout edge cases in conflict resolution
Co-Authored-By: Lilith Autocommit <noreply@atlilith.com>
This commit is contained in:
parent
e00f025193
commit
e747f78a70
1 changed files with 51 additions and 36 deletions
|
|
@ -182,31 +182,39 @@ class TestTryAutoResolveLockfiles:
|
||||||
async def test_try_auto_resolve_regen_timeout(
|
async def test_try_auto_resolve_regen_timeout(
|
||||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||||
) -> None:
|
) -> None:
|
||||||
|
"""Regen command that times out results in regen_failed entry."""
|
||||||
|
import asyncio as _asyncio
|
||||||
|
import auto_commit_service.git.conflict_resolution as cr_mod
|
||||||
|
from auto_commit_service.git.conflict_resolution import LockfileStrategy
|
||||||
|
|
||||||
repo = _make_bun_lock_conflict(tmp_path)
|
repo = _make_bun_lock_conflict(tmp_path)
|
||||||
|
|
||||||
# Stub bun to hang (sleeping long enough to trigger timeout)
|
short_strategy = LockfileStrategy(
|
||||||
bin_dir = tmp_path / "stub-bin"
|
"bun.lock", "theirs", ("bun", "install"), regen_timeout_sec=1
|
||||||
bin_dir.mkdir()
|
)
|
||||||
import os, stat as _stat
|
monkeypatch.setattr(cr_mod, "STRATEGY_BY_FILENAME", {"bun.lock": short_strategy})
|
||||||
bun = bin_dir / "bun"
|
|
||||||
bun.write_text("#!/bin/sh\nsleep 999\n")
|
|
||||||
bun.chmod(bun.stat().st_mode | _stat.S_IEXEC | _stat.S_IXGRP | _stat.S_IXOTH)
|
|
||||||
monkeypatch.setenv("PATH", f"{bin_dir}:{os.environ['PATH']}")
|
|
||||||
|
|
||||||
# Patch timeout to 0 seconds so it fires immediately
|
_real_exec = _asyncio.create_subprocess_exec
|
||||||
import auto_commit_service.git.conflict_resolution as cr_mod
|
|
||||||
original_strategies = cr_mod.LOCKFILE_STRATEGIES
|
class _HangingProc:
|
||||||
from auto_commit_service.git.conflict_resolution import LockfileStrategy
|
returncode = None
|
||||||
monkeypatch.setattr(
|
_killed = False
|
||||||
cr_mod,
|
|
||||||
"LOCKFILE_STRATEGIES",
|
async def communicate(self, **kwargs):
|
||||||
(LockfileStrategy("bun.lock", "theirs", ("bun", "install"), regen_timeout_sec=1),),
|
if self._killed:
|
||||||
)
|
return b"", b""
|
||||||
monkeypatch.setattr(
|
await _asyncio.sleep(9999)
|
||||||
cr_mod,
|
return b"", b""
|
||||||
"STRATEGY_BY_FILENAME",
|
|
||||||
{"bun.lock": LockfileStrategy("bun.lock", "theirs", ("bun", "install"), regen_timeout_sec=1)},
|
def kill(self):
|
||||||
)
|
self._killed = True
|
||||||
|
|
||||||
|
async def _fake_exec(*args, **kwargs):
|
||||||
|
if args[0] == "git":
|
||||||
|
return await _real_exec(*args, **kwargs)
|
||||||
|
return _HangingProc()
|
||||||
|
|
||||||
|
monkeypatch.setattr(cr_mod.asyncio, "create_subprocess_exec", _fake_exec)
|
||||||
|
|
||||||
result = await try_auto_resolve_lockfiles(repo)
|
result = await try_auto_resolve_lockfiles(repo)
|
||||||
|
|
||||||
|
|
@ -216,15 +224,29 @@ class TestTryAutoResolveLockfiles:
|
||||||
async def test_try_auto_resolve_regen_nonzero_exit(
|
async def test_try_auto_resolve_regen_nonzero_exit(
|
||||||
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
||||||
) -> None:
|
) -> None:
|
||||||
|
"""Regen command that exits non-zero populates regen_failed."""
|
||||||
|
import asyncio as _asyncio
|
||||||
|
import auto_commit_service.git.conflict_resolution as cr_mod
|
||||||
|
|
||||||
repo = _make_bun_lock_conflict(tmp_path)
|
repo = _make_bun_lock_conflict(tmp_path)
|
||||||
|
|
||||||
bin_dir = tmp_path / "stub-bin"
|
_real_exec = _asyncio.create_subprocess_exec
|
||||||
bin_dir.mkdir()
|
|
||||||
import os, stat as _stat
|
class _FailingProc:
|
||||||
bun = bin_dir / "bun"
|
returncode = 1
|
||||||
bun.write_text("#!/bin/sh\nexit 1\n")
|
|
||||||
bun.chmod(bun.stat().st_mode | _stat.S_IEXEC | _stat.S_IXGRP | _stat.S_IXOTH)
|
async def communicate(self, **kwargs):
|
||||||
monkeypatch.setenv("PATH", f"{bin_dir}:{os.environ['PATH']}")
|
return b"", b"regen error output"
|
||||||
|
|
||||||
|
def kill(self):
|
||||||
|
pass
|
||||||
|
|
||||||
|
async def _fake_exec(*args, **kwargs):
|
||||||
|
if args[0] == "git":
|
||||||
|
return await _real_exec(*args, **kwargs)
|
||||||
|
return _FailingProc()
|
||||||
|
|
||||||
|
monkeypatch.setattr(cr_mod.asyncio, "create_subprocess_exec", _fake_exec)
|
||||||
|
|
||||||
result = await try_auto_resolve_lockfiles(repo)
|
result = await try_auto_resolve_lockfiles(repo)
|
||||||
|
|
||||||
|
|
@ -237,13 +259,6 @@ class TestGitPullRebaseAutoResolves:
|
||||||
self, tmp_path: Path, fake_bun_binary: Path
|
self, tmp_path: Path, fake_bun_binary: Path
|
||||||
) -> None:
|
) -> None:
|
||||||
"""End-to-end: pull --rebase with a bun.lock conflict resolves automatically."""
|
"""End-to-end: pull --rebase with a bun.lock conflict resolves automatically."""
|
||||||
repo = _make_bun_lock_conflict.__wrapped__(tmp_path) if hasattr(
|
|
||||||
_make_bun_lock_conflict, "__wrapped__"
|
|
||||||
) else None
|
|
||||||
|
|
||||||
# Build scenario from scratch — _make_bun_lock_conflict leaves mid-rebase state,
|
|
||||||
# but git_pull_rebase starts by running pull --rebase itself. Set up divergence
|
|
||||||
# without starting rebase manually.
|
|
||||||
bare = tmp_path / "remote.git"
|
bare = tmp_path / "remote.git"
|
||||||
local = tmp_path / "local"
|
local = tmp_path / "local"
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue