feat: add AutoMerge with E2E-conditional merge logic
This commit is contained in:
parent
db6e9b4a41
commit
0c4c22be5a
61
agent/auto_merge.py
Normal file
61
agent/auto_merge.py
Normal file
@ -0,0 +1,61 @@
|
||||
"""E2E 조건부 자동 머지 로직.
|
||||
|
||||
autonomous 모드에서 조건을 검증하여 PR을 자동 머지한다.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def should_auto_merge(
|
||||
auto_merge: bool, require_e2e: bool, max_files_changed: int,
|
||||
blocked_paths: list[str], changed_files: list[str],
|
||||
tests_passed: bool, e2e_passed: bool,
|
||||
) -> bool:
|
||||
if not auto_merge:
|
||||
return False
|
||||
if not tests_passed:
|
||||
return False
|
||||
if require_e2e and not e2e_passed:
|
||||
return False
|
||||
if len(changed_files) > max_files_changed:
|
||||
return False
|
||||
for f in changed_files:
|
||||
for blocked in blocked_paths:
|
||||
if f == blocked or f.endswith("/" + blocked):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
class AutoMergeChecker:
|
||||
def __init__(
|
||||
self, auto_merge: bool = False, require_e2e: bool = False,
|
||||
max_files_changed: int = 10, blocked_paths: list[str] | None = None,
|
||||
):
|
||||
self._auto_merge = auto_merge
|
||||
self._require_e2e = require_e2e
|
||||
self._max_files_changed = max_files_changed
|
||||
self._blocked_paths = blocked_paths or []
|
||||
|
||||
async def try_merge(
|
||||
self, owner: str, repo: str, pr_number: int,
|
||||
changed_files: list[str], tests_passed: bool, e2e_passed: bool,
|
||||
) -> dict:
|
||||
can_merge = should_auto_merge(
|
||||
auto_merge=self._auto_merge, require_e2e=self._require_e2e,
|
||||
max_files_changed=self._max_files_changed, blocked_paths=self._blocked_paths,
|
||||
changed_files=changed_files, tests_passed=tests_passed, e2e_passed=e2e_passed,
|
||||
)
|
||||
if not can_merge:
|
||||
return {"merged": False, "reason": "conditions not met"}
|
||||
try:
|
||||
from agent.utils.gitea_client import get_gitea_client
|
||||
client = get_gitea_client()
|
||||
await client.merge_pull_request(owner=owner, repo=repo, pr_number=pr_number)
|
||||
logger.info("Auto-merged PR #%d on %s/%s", pr_number, owner, repo)
|
||||
return {"merged": True, "reason": "all conditions met"}
|
||||
except Exception as e:
|
||||
logger.exception("Auto-merge failed for PR #%d", pr_number)
|
||||
return {"merged": False, "reason": f"merge failed: {e}"}
|
||||
70
tests/test_auto_merge.py
Normal file
70
tests/test_auto_merge.py
Normal file
@ -0,0 +1,70 @@
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
from agent.auto_merge import should_auto_merge, AutoMergeChecker
|
||||
|
||||
|
||||
def test_should_not_merge_when_disabled():
|
||||
result = should_auto_merge(
|
||||
auto_merge=False, require_e2e=False, max_files_changed=10,
|
||||
blocked_paths=[".env"], changed_files=["backend/app/main.py"],
|
||||
tests_passed=True, e2e_passed=True,
|
||||
)
|
||||
assert result is False
|
||||
|
||||
|
||||
def test_should_merge_when_all_conditions_met():
|
||||
result = should_auto_merge(
|
||||
auto_merge=True, require_e2e=True, max_files_changed=10,
|
||||
blocked_paths=[".env", "quant.md"],
|
||||
changed_files=["backend/app/main.py", "backend/tests/test_main.py"],
|
||||
tests_passed=True, e2e_passed=True,
|
||||
)
|
||||
assert result is True
|
||||
|
||||
|
||||
def test_should_not_merge_when_tests_fail():
|
||||
result = should_auto_merge(
|
||||
auto_merge=True, require_e2e=False, max_files_changed=10,
|
||||
blocked_paths=[], changed_files=["a.py"],
|
||||
tests_passed=False, e2e_passed=False,
|
||||
)
|
||||
assert result is False
|
||||
|
||||
|
||||
def test_should_not_merge_when_e2e_required_but_failed():
|
||||
result = should_auto_merge(
|
||||
auto_merge=True, require_e2e=True, max_files_changed=10,
|
||||
blocked_paths=[], changed_files=["a.py"],
|
||||
tests_passed=True, e2e_passed=False,
|
||||
)
|
||||
assert result is False
|
||||
|
||||
|
||||
def test_should_not_merge_when_too_many_files():
|
||||
files = [f"file{i}.py" for i in range(15)]
|
||||
result = should_auto_merge(
|
||||
auto_merge=True, require_e2e=False, max_files_changed=10,
|
||||
blocked_paths=[], changed_files=files,
|
||||
tests_passed=True, e2e_passed=True,
|
||||
)
|
||||
assert result is False
|
||||
|
||||
|
||||
def test_should_not_merge_when_blocked_path_modified():
|
||||
result = should_auto_merge(
|
||||
auto_merge=True, require_e2e=False, max_files_changed=10,
|
||||
blocked_paths=[".env", "quant.md"],
|
||||
changed_files=["backend/app/main.py", ".env"],
|
||||
tests_passed=True, e2e_passed=True,
|
||||
)
|
||||
assert result is False
|
||||
|
||||
|
||||
def test_should_merge_without_e2e_when_not_required():
|
||||
result = should_auto_merge(
|
||||
auto_merge=True, require_e2e=False, max_files_changed=10,
|
||||
blocked_paths=[], changed_files=["a.py"],
|
||||
tests_passed=True, e2e_passed=False,
|
||||
)
|
||||
assert result is True
|
||||
Loading…
x
Reference in New Issue
Block a user