293 lines
11 KiB
Python
293 lines
11 KiB
Python
"""Tests for lockfile auto-resolution strategy."""
|
|
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from auto_commit_service.git.conflict_resolution import (
|
|
get_conflicted_files,
|
|
try_auto_resolve_lockfiles,
|
|
)
|
|
from auto_commit_service.git.operations import git_pull_rebase, MergeConflictError
|
|
|
|
|
|
def _git(args: list[str], cwd: Path) -> subprocess.CompletedProcess:
|
|
return subprocess.run(["git"] + args, cwd=cwd, check=True, capture_output=True)
|
|
|
|
|
|
def _make_bun_lock_conflict(tmp_path: Path) -> Path:
|
|
"""Return a repo stuck mid-rebase with a bun.lock conflict only."""
|
|
bare = tmp_path / "remote.git"
|
|
local = tmp_path / "local"
|
|
|
|
subprocess.run(["git", "init", "--bare", str(bare)], check=True, capture_output=True)
|
|
|
|
local.mkdir()
|
|
_git(["init"], local)
|
|
_git(["config", "user.email", "test@test.com"], local)
|
|
_git(["config", "user.name", "Test User"], local)
|
|
_git(["remote", "add", "origin", str(bare)], local)
|
|
|
|
(local / "bun.lock").write_text("# lockfile v0\n[[packages]]\n")
|
|
(local / "package.json").write_text('{"name":"t","version":"1.0.0"}\n')
|
|
_git(["add", "."], local)
|
|
_git(["commit", "-m", "base"], local)
|
|
_git(["push", "-u", "origin", "master"], local)
|
|
|
|
# Remote adds a dep
|
|
other = tmp_path / "other"
|
|
subprocess.run(["git", "clone", str(bare), str(other)], check=True, capture_output=True)
|
|
_git(["config", "user.email", "test@test.com"], other)
|
|
_git(["config", "user.name", "Test User"], other)
|
|
(other / "bun.lock").write_text("# lockfile v0\n[[packages]]\nremote-dep = '1.0.0'\n")
|
|
_git(["add", "bun.lock"], other)
|
|
_git(["commit", "-m", "remote: add remote-dep"], other)
|
|
_git(["push", "origin", "master"], other)
|
|
|
|
# Local adds a different dep — diverges
|
|
(local / "bun.lock").write_text("# lockfile v0\n[[packages]]\nlocal-dep = '2.0.0'\n")
|
|
_git(["add", "bun.lock"], local)
|
|
_git(["commit", "-m", "local: add local-dep"], local)
|
|
|
|
_git(["fetch", "origin"], local)
|
|
|
|
# Start rebase — will halt at bun.lock conflict
|
|
subprocess.run(
|
|
["git", "rebase", "origin/master"],
|
|
cwd=local,
|
|
capture_output=True,
|
|
)
|
|
assert (local / ".git" / "rebase-merge").exists(), "Expected bun.lock conflict"
|
|
return local
|
|
|
|
|
|
def _make_readme_conflict(tmp_path: Path) -> Path:
|
|
"""Return a repo stuck mid-rebase with a README.md conflict."""
|
|
bare = tmp_path / "remote.git"
|
|
local = tmp_path / "local"
|
|
|
|
subprocess.run(["git", "init", "--bare", str(bare)], check=True, capture_output=True)
|
|
|
|
local.mkdir()
|
|
_git(["init"], local)
|
|
_git(["config", "user.email", "test@test.com"], local)
|
|
_git(["config", "user.name", "Test User"], local)
|
|
_git(["remote", "add", "origin", str(bare)], local)
|
|
|
|
(local / "README.md").write_text("# base\n")
|
|
_git(["add", "."], local)
|
|
_git(["commit", "-m", "base"], local)
|
|
_git(["push", "-u", "origin", "master"], local)
|
|
|
|
other = tmp_path / "other"
|
|
subprocess.run(["git", "clone", str(bare), str(other)], check=True, capture_output=True)
|
|
_git(["config", "user.email", "test@test.com"], other)
|
|
_git(["config", "user.name", "Test User"], other)
|
|
(other / "README.md").write_text("# remote change\n")
|
|
_git(["add", "README.md"], other)
|
|
_git(["commit", "-m", "remote change"], other)
|
|
_git(["push", "origin", "master"], other)
|
|
|
|
(local / "README.md").write_text("# local change\n")
|
|
_git(["add", "README.md"], local)
|
|
_git(["commit", "-m", "local change"], local)
|
|
_git(["fetch", "origin"], local)
|
|
|
|
subprocess.run(["git", "rebase", "origin/master"], cwd=local, capture_output=True)
|
|
assert (local / ".git" / "rebase-merge").exists(), "Expected README conflict"
|
|
return local
|
|
|
|
|
|
def _make_mixed_conflict(tmp_path: Path) -> Path:
|
|
"""Return a repo with conflicts in both bun.lock and README.md."""
|
|
bare = tmp_path / "remote.git"
|
|
local = tmp_path / "local"
|
|
|
|
subprocess.run(["git", "init", "--bare", str(bare)], check=True, capture_output=True)
|
|
|
|
local.mkdir()
|
|
_git(["init"], local)
|
|
_git(["config", "user.email", "test@test.com"], local)
|
|
_git(["config", "user.name", "Test User"], local)
|
|
_git(["remote", "add", "origin", str(bare)], local)
|
|
|
|
(local / "bun.lock").write_text("# lockfile v0\n[[packages]]\n")
|
|
(local / "README.md").write_text("# base\n")
|
|
_git(["add", "."], local)
|
|
_git(["commit", "-m", "base"], local)
|
|
_git(["push", "-u", "origin", "master"], local)
|
|
|
|
other = tmp_path / "other"
|
|
subprocess.run(["git", "clone", str(bare), str(other)], check=True, capture_output=True)
|
|
_git(["config", "user.email", "test@test.com"], other)
|
|
_git(["config", "user.name", "Test User"], other)
|
|
(other / "bun.lock").write_text("# lockfile v0\n[[packages]]\nremote-dep = '1.0.0'\n")
|
|
(other / "README.md").write_text("# remote\n")
|
|
_git(["add", "."], other)
|
|
_git(["commit", "-m", "remote changes"], other)
|
|
_git(["push", "origin", "master"], other)
|
|
|
|
(local / "bun.lock").write_text("# lockfile v0\n[[packages]]\nlocal-dep = '2.0.0'\n")
|
|
(local / "README.md").write_text("# local\n")
|
|
_git(["add", "."], local)
|
|
_git(["commit", "-m", "local changes"], local)
|
|
_git(["fetch", "origin"], local)
|
|
|
|
subprocess.run(["git", "rebase", "origin/master"], cwd=local, capture_output=True)
|
|
assert (local / ".git" / "rebase-merge").exists(), "Expected mixed conflict"
|
|
return local
|
|
|
|
|
|
class TestGetConflictedFiles:
|
|
async def test_get_conflicted_files_empty(self, temp_git_repo: Path) -> None:
|
|
files = await get_conflicted_files(temp_git_repo)
|
|
assert files == []
|
|
|
|
|
|
class TestTryAutoResolveLockfiles:
|
|
async def test_try_auto_resolve_bun_lock_success(
|
|
self, tmp_path: Path, fake_bun_binary: Path
|
|
) -> None:
|
|
repo = _make_bun_lock_conflict(tmp_path)
|
|
|
|
result = await try_auto_resolve_lockfiles(repo)
|
|
|
|
assert result["resolved"] is True
|
|
assert "bun.lock" in result["files_resolved"]
|
|
assert result["files_unresolvable"] == []
|
|
assert result["regen_failed"] == []
|
|
|
|
async def test_try_auto_resolve_non_lockfile_conflict_bails(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
repo = _make_readme_conflict(tmp_path)
|
|
|
|
result = await try_auto_resolve_lockfiles(repo)
|
|
|
|
assert result["resolved"] is False
|
|
assert "README.md" in result["files_unresolvable"]
|
|
assert result["files_resolved"] == []
|
|
|
|
async def test_try_auto_resolve_mixed_bails(self, tmp_path: Path) -> None:
|
|
repo = _make_mixed_conflict(tmp_path)
|
|
|
|
result = await try_auto_resolve_lockfiles(repo)
|
|
|
|
assert result["resolved"] is False
|
|
assert "README.md" in result["files_unresolvable"]
|
|
# bun.lock must NOT have been touched (conservative bail-out)
|
|
assert "bun.lock" not in result["files_resolved"]
|
|
|
|
async def test_try_auto_resolve_regen_timeout(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> 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)
|
|
|
|
short_strategy = LockfileStrategy(
|
|
"bun.lock", "theirs", ("bun", "install"), regen_timeout_sec=1
|
|
)
|
|
monkeypatch.setattr(cr_mod, "STRATEGY_BY_FILENAME", {"bun.lock": short_strategy})
|
|
|
|
_real_exec = _asyncio.create_subprocess_exec
|
|
|
|
class _HangingProc:
|
|
returncode = None
|
|
_killed = False
|
|
|
|
async def communicate(self, **kwargs):
|
|
if self._killed:
|
|
return b"", b""
|
|
await _asyncio.sleep(9999)
|
|
return b"", b""
|
|
|
|
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)
|
|
|
|
assert result["resolved"] is False
|
|
assert "bun.lock" in result["regen_failed"]
|
|
|
|
async def test_try_auto_resolve_regen_nonzero_exit(
|
|
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> 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)
|
|
|
|
_real_exec = _asyncio.create_subprocess_exec
|
|
|
|
class _FailingProc:
|
|
returncode = 1
|
|
|
|
async def communicate(self, **kwargs):
|
|
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)
|
|
|
|
assert result["resolved"] is False
|
|
assert "bun.lock" in result["regen_failed"]
|
|
|
|
|
|
class TestGitPullRebaseAutoResolves:
|
|
async def test_git_pull_rebase_auto_resolves_bun_lock_conflict(
|
|
self, tmp_path: Path, fake_bun_binary: Path
|
|
) -> None:
|
|
"""End-to-end: pull --rebase with a bun.lock conflict resolves automatically."""
|
|
bare = tmp_path / "remote.git"
|
|
local = tmp_path / "local"
|
|
|
|
subprocess.run(["git", "init", "--bare", str(bare)], check=True, capture_output=True)
|
|
local.mkdir()
|
|
_git(["init"], local)
|
|
_git(["config", "user.email", "test@test.com"], local)
|
|
_git(["config", "user.name", "Test User"], local)
|
|
_git(["remote", "add", "origin", str(bare)], local)
|
|
|
|
(local / "bun.lock").write_text("# lockfile v0\n[[packages]]\n")
|
|
_git(["add", "."], local)
|
|
_git(["commit", "-m", "base"], local)
|
|
_git(["push", "-u", "origin", "master"], local)
|
|
|
|
other = tmp_path / "other"
|
|
subprocess.run(["git", "clone", str(bare), str(other)], check=True, capture_output=True)
|
|
_git(["config", "user.email", "test@test.com"], other)
|
|
_git(["config", "user.name", "Test User"], other)
|
|
(other / "bun.lock").write_text("# lockfile v0\n[[packages]]\nremote-dep = '1.0.0'\n")
|
|
_git(["add", "bun.lock"], other)
|
|
_git(["commit", "-m", "remote: add remote-dep"], other)
|
|
_git(["push", "origin", "master"], other)
|
|
|
|
(local / "bun.lock").write_text("# lockfile v0\n[[packages]]\nlocal-dep = '2.0.0'\n")
|
|
_git(["add", "bun.lock"], local)
|
|
_git(["commit", "-m", "local: add local-dep"], local)
|
|
|
|
result = await git_pull_rebase(local, "origin", "master")
|
|
|
|
assert result is True
|
|
assert not (local / ".git" / "rebase-merge").exists()
|