278 lines
11 KiB
Python
278 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:
|
|
repo = _make_bun_lock_conflict(tmp_path)
|
|
|
|
# Stub bun to hang (sleeping long enough to trigger timeout)
|
|
bin_dir = tmp_path / "stub-bin"
|
|
bin_dir.mkdir()
|
|
import os, stat as _stat
|
|
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
|
|
import auto_commit_service.git.conflict_resolution as cr_mod
|
|
original_strategies = cr_mod.LOCKFILE_STRATEGIES
|
|
from auto_commit_service.git.conflict_resolution import LockfileStrategy
|
|
monkeypatch.setattr(
|
|
cr_mod,
|
|
"LOCKFILE_STRATEGIES",
|
|
(LockfileStrategy("bun.lock", "theirs", ("bun", "install"), regen_timeout_sec=1),),
|
|
)
|
|
monkeypatch.setattr(
|
|
cr_mod,
|
|
"STRATEGY_BY_FILENAME",
|
|
{"bun.lock": LockfileStrategy("bun.lock", "theirs", ("bun", "install"), regen_timeout_sec=1)},
|
|
)
|
|
|
|
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:
|
|
repo = _make_bun_lock_conflict(tmp_path)
|
|
|
|
bin_dir = tmp_path / "stub-bin"
|
|
bin_dir.mkdir()
|
|
import os, stat as _stat
|
|
bun = bin_dir / "bun"
|
|
bun.write_text("#!/bin/sh\nexit 1\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']}")
|
|
|
|
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."""
|
|
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"
|
|
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()
|