460 lines
20 KiB
Python
460 lines
20 KiB
Python
"""KIN-103 — Regression + unit tests for worktrees_enabled feature.
|
||
|
||
Covers:
|
||
1. core/worktree.py — create_worktree, merge_worktree, cleanup_worktree, ensure_gitignore
|
||
2. agents/runner.py run_pipeline — worktree lifecycle (create, merge, cleanup)
|
||
- eligible role + worktrees_enabled=True → worktree created
|
||
- non-eligible role → no worktree
|
||
- worktrees_enabled=False → no worktree
|
||
- success path → merge + cleanup called
|
||
- failure path → cleanup called without merge
|
||
- create_worktree returns None → graceful fallback, no crash
|
||
- non-git directory (create_worktree fails) → pipeline continues normally
|
||
"""
|
||
|
||
import subprocess
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock, call, patch
|
||
|
||
import pytest
|
||
|
||
from core.db import init_db
|
||
from core import models
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Shared fixtures
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
@pytest.fixture
|
||
def conn():
|
||
c = init_db(":memory:")
|
||
models.create_project(c, "wt1", "WT Project", "/projects/wtproject",
|
||
tech_stack=["python"])
|
||
models.create_task(c, "WT1-001", "wt1", "Fix worktree bug",
|
||
brief={"route_type": "debug"})
|
||
yield c
|
||
c.close()
|
||
|
||
|
||
@pytest.fixture
|
||
def conn_wt_enabled(conn):
|
||
"""Project with worktrees_enabled=1."""
|
||
conn.execute("UPDATE projects SET worktrees_enabled=1 WHERE id='wt1'")
|
||
conn.commit()
|
||
return conn
|
||
|
||
|
||
def _ok_run():
|
||
m = MagicMock()
|
||
m.returncode = 0
|
||
m.stdout = ""
|
||
m.stderr = ""
|
||
return m
|
||
|
||
|
||
def _fail_run(stderr="git error"):
|
||
m = MagicMock()
|
||
m.returncode = 1
|
||
m.stdout = ""
|
||
m.stderr = stderr
|
||
return m
|
||
|
||
|
||
def _claude_ok():
|
||
import json
|
||
m = MagicMock()
|
||
m.returncode = 0
|
||
m.stdout = json.dumps({"result": "done"})
|
||
m.stderr = ""
|
||
return m
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 1. core/worktree.py — create_worktree
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestCreateWorktree:
|
||
@patch("subprocess.run")
|
||
def test_create_worktree_success_returns_path(self, mock_run, tmp_path):
|
||
"""create_worktree вызывает git worktree add и возвращает путь."""
|
||
from core.worktree import create_worktree
|
||
mock_run.return_value = _ok_run()
|
||
|
||
result = create_worktree(str(tmp_path), "WT1-001", "backend_dev")
|
||
|
||
assert result is not None
|
||
assert "WT1-001" in result
|
||
assert "backend_dev" in result
|
||
|
||
@patch("subprocess.run")
|
||
def test_create_worktree_calls_git_worktree_add(self, mock_run, tmp_path):
|
||
"""create_worktree вызывает git worktree add с корректными аргументами."""
|
||
from core.worktree import create_worktree
|
||
mock_run.return_value = _ok_run()
|
||
|
||
create_worktree(str(tmp_path), "WT1-001", "debugger")
|
||
|
||
cmd = mock_run.call_args[0][0]
|
||
assert "worktree" in cmd
|
||
assert "add" in cmd
|
||
|
||
@patch("subprocess.run")
|
||
def test_create_worktree_git_fail_returns_none(self, mock_run, tmp_path):
|
||
"""Если git возвращает non-zero returncode — create_worktree возвращает None."""
|
||
from core.worktree import create_worktree
|
||
mock_run.return_value = _fail_run("fatal: not a git repo")
|
||
|
||
result = create_worktree(str(tmp_path), "WT1-001", "debugger")
|
||
|
||
assert result is None
|
||
|
||
@patch("subprocess.run", side_effect=Exception("git not found"))
|
||
def test_create_worktree_exception_returns_none(self, mock_run, tmp_path):
|
||
"""При исключении create_worktree возвращает None, не поднимает."""
|
||
from core.worktree import create_worktree
|
||
|
||
result = create_worktree(str(tmp_path), "WT1-001", "debugger")
|
||
|
||
assert result is None
|
||
|
||
@patch("subprocess.run")
|
||
def test_create_worktree_sanitizes_step_name(self, mock_run, tmp_path):
|
||
"""Слэши и пробелы в step_name заменяются на _ в branch name (decision #596)."""
|
||
from core.worktree import create_worktree
|
||
mock_run.return_value = _ok_run()
|
||
|
||
result = create_worktree(str(tmp_path), "WT1-001", "role/with spaces")
|
||
|
||
assert result is not None
|
||
assert "/" not in Path(result).name
|
||
assert "role_with_spaces" in Path(result).name
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 2. core/worktree.py — merge_worktree
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestMergeWorktree:
|
||
@patch("subprocess.run")
|
||
def test_merge_worktree_success_returns_dict(self, mock_run, tmp_path):
|
||
"""Успешный merge возвращает {success: True, conflicts: [], merged_files: [...]}."""
|
||
from core.worktree import merge_worktree
|
||
|
||
worktree_path = tmp_path / ".kin_worktrees" / "WT1-001-debugger"
|
||
worktree_path.mkdir(parents=True)
|
||
|
||
# Fix B adds git add -A + git commit before merge
|
||
mock_run.side_effect = [
|
||
_ok_run(), # git add -A
|
||
_ok_run(), # git commit -m
|
||
_ok_run(), # git merge --no-ff
|
||
_ok_run(), # git diff HEAD~1 HEAD --name-only
|
||
]
|
||
|
||
result = merge_worktree(str(worktree_path), str(tmp_path))
|
||
|
||
assert result["success"] is True
|
||
assert result["conflicts"] == []
|
||
|
||
@patch("subprocess.run")
|
||
def test_merge_worktree_conflict_returns_conflicts(self, mock_run, tmp_path):
|
||
"""При конфликте merge возвращает {success: False, conflicts: [...]}."""
|
||
from core.worktree import merge_worktree
|
||
|
||
worktree_path = tmp_path / ".kin_worktrees" / "WT1-001-debugger"
|
||
worktree_path.mkdir(parents=True)
|
||
|
||
conflict_diff = MagicMock()
|
||
conflict_diff.returncode = 0
|
||
conflict_diff.stdout = "src/app.py\nsrc/utils.py\n"
|
||
conflict_diff.stderr = ""
|
||
|
||
abort_result = _ok_run()
|
||
|
||
merge_fail = _fail_run("CONFLICT (content)")
|
||
|
||
# Fix B adds git add -A + git commit before merge
|
||
mock_run.side_effect = [_ok_run(), _ok_run(), merge_fail, conflict_diff, abort_result]
|
||
|
||
result = merge_worktree(str(worktree_path), str(tmp_path))
|
||
|
||
assert result["success"] is False
|
||
assert "src/app.py" in result["conflicts"]
|
||
|
||
@patch("subprocess.run", side_effect=Exception("git gone"))
|
||
def test_merge_worktree_exception_returns_failure(self, mock_run, tmp_path):
|
||
"""При исключении merge возвращает {success: False}, не поднимает."""
|
||
from core.worktree import merge_worktree
|
||
|
||
worktree_path = tmp_path / ".kin_worktrees" / "WT1-001-debugger"
|
||
worktree_path.mkdir(parents=True)
|
||
|
||
result = merge_worktree(str(worktree_path), str(tmp_path))
|
||
|
||
assert result["success"] is False
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 3. core/worktree.py — cleanup_worktree
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestCleanupWorktree:
|
||
@patch("subprocess.run")
|
||
def test_cleanup_calls_worktree_remove_and_branch_delete(self, mock_run, tmp_path):
|
||
"""cleanup_worktree вызывает git worktree remove и git branch -D."""
|
||
from core.worktree import cleanup_worktree
|
||
|
||
worktree_path = tmp_path / ".kin_worktrees" / "WT1-001-debugger"
|
||
mock_run.return_value = _ok_run()
|
||
|
||
cleanup_worktree(str(worktree_path), str(tmp_path))
|
||
|
||
assert mock_run.call_count == 2
|
||
cmds = [call[0][0] for call in mock_run.call_args_list]
|
||
worktree_remove_cmd = cmds[0]
|
||
branch_delete_cmd = cmds[1]
|
||
assert "worktree" in worktree_remove_cmd and "remove" in worktree_remove_cmd
|
||
assert "branch" in branch_delete_cmd and "-D" in branch_delete_cmd
|
||
|
||
@patch("subprocess.run", side_effect=Exception("permission denied"))
|
||
def test_cleanup_worktree_exception_does_not_raise(self, mock_run, tmp_path):
|
||
"""cleanup_worktree никогда не поднимает исключение."""
|
||
from core.worktree import cleanup_worktree
|
||
|
||
worktree_path = tmp_path / ".kin_worktrees" / "WT1-001-debugger"
|
||
|
||
# Must not raise
|
||
cleanup_worktree(str(worktree_path), str(tmp_path))
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 4. core/worktree.py — ensure_gitignore
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestEnsureGitignore:
|
||
def test_ensure_gitignore_creates_file_if_not_exists(self, tmp_path):
|
||
"""ensure_gitignore создаёт .gitignore с .kin_worktrees/ если файла нет."""
|
||
from core.worktree import ensure_gitignore
|
||
|
||
ensure_gitignore(str(tmp_path))
|
||
|
||
gitignore = tmp_path / ".gitignore"
|
||
assert gitignore.exists()
|
||
assert ".kin_worktrees/" in gitignore.read_text()
|
||
|
||
def test_ensure_gitignore_appends_entry_if_file_exists(self, tmp_path):
|
||
"""ensure_gitignore добавляет .kin_worktrees/ в существующий .gitignore."""
|
||
from core.worktree import ensure_gitignore
|
||
|
||
gitignore = tmp_path / ".gitignore"
|
||
gitignore.write_text("*.pyc\n__pycache__/\n")
|
||
|
||
ensure_gitignore(str(tmp_path))
|
||
|
||
content = gitignore.read_text()
|
||
assert ".kin_worktrees/" in content
|
||
assert "*.pyc" in content # existing entries preserved
|
||
|
||
def test_ensure_gitignore_no_duplicate_if_entry_exists(self, tmp_path):
|
||
"""ensure_gitignore не дублирует запись если уже есть."""
|
||
from core.worktree import ensure_gitignore
|
||
|
||
gitignore = tmp_path / ".gitignore"
|
||
gitignore.write_text(".kin_worktrees/\n")
|
||
|
||
ensure_gitignore(str(tmp_path))
|
||
|
||
content = gitignore.read_text()
|
||
assert content.count(".kin_worktrees/") == 1
|
||
|
||
def test_ensure_gitignore_exception_does_not_raise(self, tmp_path):
|
||
"""ensure_gitignore не поднимает исключение при любых ошибках."""
|
||
from core.worktree import ensure_gitignore
|
||
|
||
# Read-only path trick: pass a file as path (not a dir)
|
||
fake_path = tmp_path / "not_a_dir.txt"
|
||
fake_path.write_text("content")
|
||
|
||
# Must not raise — function is defensive
|
||
try:
|
||
ensure_gitignore(str(fake_path))
|
||
except Exception as e:
|
||
pytest.fail(f"ensure_gitignore raised unexpectedly: {e}")
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# 5. runner — worktree lifecycle в run_pipeline
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
class TestRunPipelineWorktrees:
|
||
@patch("agents.runner.subprocess.run")
|
||
@patch("core.worktree.cleanup_worktree")
|
||
@patch("core.worktree.merge_worktree",
|
||
return_value={"success": True, "conflicts": [], "merged_files": []})
|
||
@patch("core.worktree.create_worktree", return_value="/tmp/fake-worktree")
|
||
@patch("core.worktree.ensure_gitignore")
|
||
def test_worktrees_enabled_eligible_role_creates_worktree(
|
||
self, mock_gitignore, mock_create, mock_merge, mock_cleanup, mock_runner_run,
|
||
conn_wt_enabled
|
||
):
|
||
"""worktrees_enabled=True + роль debugger → create_worktree вызывается."""
|
||
from agents.runner import run_pipeline
|
||
|
||
mock_runner_run.return_value = _claude_ok()
|
||
|
||
result = run_pipeline(conn_wt_enabled, "WT1-001",
|
||
[{"role": "debugger", "brief": "fix"}])
|
||
|
||
assert result["success"] is True
|
||
mock_create.assert_called_once()
|
||
call_args = mock_create.call_args[0]
|
||
assert "WT1-001" in call_args[1]
|
||
assert "debugger" in call_args[2]
|
||
|
||
@patch("agents.runner.subprocess.run")
|
||
@patch("core.worktree.subprocess.run")
|
||
def test_worktrees_disabled_no_worktree_created(
|
||
self, mock_wt_run, mock_runner_run, conn
|
||
):
|
||
"""worktrees_enabled=False → create_worktree НЕ вызывается."""
|
||
from agents.runner import run_pipeline
|
||
|
||
mock_runner_run.return_value = _claude_ok()
|
||
mock_wt_run.return_value = _ok_run()
|
||
|
||
with patch("core.worktree._git", return_value="git"):
|
||
result = run_pipeline(conn, "WT1-001",
|
||
[{"role": "debugger", "brief": "fix"}])
|
||
|
||
assert result["success"] is True
|
||
# git worktree add must NOT have been called
|
||
cmds = [c[0][0] for c in mock_wt_run.call_args_list]
|
||
worktree_add_calls = [cmd for cmd in cmds if "worktree" in cmd and "add" in cmd]
|
||
assert len(worktree_add_calls) == 0
|
||
|
||
@patch("agents.runner.subprocess.run")
|
||
@patch("core.worktree.subprocess.run")
|
||
def test_non_eligible_role_no_worktree(
|
||
self, mock_wt_run, mock_runner_run, conn_wt_enabled
|
||
):
|
||
"""Роль tester не входит в _WORKTREE_ROLES → worktree не создаётся."""
|
||
from agents.runner import run_pipeline
|
||
|
||
mock_runner_run.return_value = _claude_ok()
|
||
mock_wt_run.return_value = _ok_run()
|
||
|
||
with patch("core.worktree._git", return_value="git"):
|
||
result = run_pipeline(conn_wt_enabled, "WT1-001",
|
||
[{"role": "tester", "brief": "test"}])
|
||
|
||
assert result["success"] is True
|
||
cmds = [c[0][0] for c in mock_wt_run.call_args_list]
|
||
worktree_add_calls = [cmd for cmd in cmds if "worktree" in cmd and "add" in cmd]
|
||
assert len(worktree_add_calls) == 0
|
||
|
||
@patch("agents.runner.subprocess.run")
|
||
@patch("core.worktree.create_worktree", return_value=None)
|
||
@patch("core.worktree.ensure_gitignore")
|
||
def test_create_worktree_returns_none_fallback_no_crash(
|
||
self, mock_gitignore, mock_create, mock_runner_run, conn_wt_enabled
|
||
):
|
||
"""Если create_worktree возвращает None (не-git dir) — pipeline продолжается нормально."""
|
||
from agents.runner import run_pipeline
|
||
|
||
mock_runner_run.return_value = _claude_ok()
|
||
|
||
result = run_pipeline(conn_wt_enabled, "WT1-001",
|
||
[{"role": "debugger", "brief": "fix"}])
|
||
|
||
assert result["success"] is True
|
||
assert result["steps_completed"] == 1
|
||
|
||
@patch("agents.runner.subprocess.run")
|
||
@patch("core.worktree.cleanup_worktree")
|
||
@patch("core.worktree.merge_worktree", return_value={"success": True, "conflicts": [], "merged_files": []})
|
||
@patch("core.worktree.create_worktree", return_value="/tmp/WT1-001-debugger")
|
||
@patch("core.worktree.ensure_gitignore")
|
||
def test_successful_step_calls_merge_then_cleanup(
|
||
self, mock_gitignore, mock_create, mock_merge, mock_cleanup, mock_runner_run, conn_wt_enabled
|
||
):
|
||
"""После успешного шага: merge_worktree вызывается, затем cleanup_worktree."""
|
||
from agents.runner import run_pipeline
|
||
|
||
mock_runner_run.return_value = _claude_ok()
|
||
|
||
result = run_pipeline(conn_wt_enabled, "WT1-001",
|
||
[{"role": "debugger", "brief": "fix"}])
|
||
|
||
assert result["success"] is True
|
||
mock_merge.assert_called_once()
|
||
mock_cleanup.assert_called_once()
|
||
|
||
@patch("agents.runner.subprocess.run")
|
||
@patch("core.worktree.cleanup_worktree")
|
||
@patch("core.worktree.merge_worktree")
|
||
@patch("core.worktree.create_worktree", return_value="/tmp/WT1-001-debugger")
|
||
@patch("core.worktree.ensure_gitignore")
|
||
def test_failed_step_calls_cleanup_without_merge(
|
||
self, mock_gitignore, mock_create, mock_merge, mock_cleanup, mock_runner_run, conn_wt_enabled
|
||
):
|
||
"""После неудачного шага: cleanup_worktree вызывается, merge_worktree — нет."""
|
||
from agents.runner import run_pipeline
|
||
|
||
fail = MagicMock()
|
||
fail.returncode = 1
|
||
fail.stdout = ""
|
||
fail.stderr = "compilation error"
|
||
mock_runner_run.return_value = fail
|
||
|
||
result = run_pipeline(conn_wt_enabled, "WT1-001",
|
||
[{"role": "debugger", "brief": "fix"}])
|
||
|
||
assert result["success"] is False
|
||
mock_merge.assert_not_called()
|
||
mock_cleanup.assert_called_once()
|
||
|
||
@patch("agents.runner.subprocess.run")
|
||
@patch("core.worktree.cleanup_worktree")
|
||
@patch("core.worktree.merge_worktree",
|
||
return_value={"success": False, "conflicts": ["src/app.py"], "merged_files": []})
|
||
@patch("core.worktree.create_worktree", return_value="/tmp/WT1-001-debugger")
|
||
@patch("core.worktree.ensure_gitignore")
|
||
def test_merge_conflict_sets_task_blocked(
|
||
self, mock_gitignore, mock_create, mock_merge, mock_cleanup, mock_runner_run, conn_wt_enabled
|
||
):
|
||
"""При конфликте merge: задача переходит в blocked с корректной причиной."""
|
||
from agents.runner import run_pipeline
|
||
|
||
mock_runner_run.return_value = _claude_ok()
|
||
|
||
result = run_pipeline(conn_wt_enabled, "WT1-001",
|
||
[{"role": "debugger", "brief": "fix"}])
|
||
|
||
# pipeline fails due to merge conflict
|
||
assert result["success"] is False
|
||
task = models.get_task(conn_wt_enabled, "WT1-001")
|
||
assert task["status"] == "blocked"
|
||
assert "src/app.py" in (task.get("blocked_reason") or "")
|
||
# cleanup still called after conflict
|
||
mock_cleanup.assert_called_once()
|
||
|
||
@patch("agents.runner.subprocess.run")
|
||
@patch("core.worktree.cleanup_worktree", side_effect=Exception("disk error"))
|
||
@patch("core.worktree.merge_worktree",
|
||
return_value={"success": True, "conflicts": [], "merged_files": []})
|
||
@patch("core.worktree.create_worktree", return_value="/tmp/WT1-001-debugger")
|
||
@patch("core.worktree.ensure_gitignore")
|
||
def test_cleanup_exception_does_not_propagate(
|
||
self, mock_gitignore, mock_create, mock_merge, mock_cleanup, mock_runner_run, conn_wt_enabled
|
||
):
|
||
"""Исключение в cleanup_worktree не блокирует pipeline — он завершается успешно."""
|
||
from agents.runner import run_pipeline
|
||
|
||
mock_runner_run.return_value = _claude_ok()
|
||
|
||
# Must not raise
|
||
result = run_pipeline(conn_wt_enabled, "WT1-001",
|
||
[{"role": "debugger", "brief": "fix"}])
|
||
|
||
assert result["success"] is True
|