62 lines
2.2 KiB
Python
62 lines
2.2 KiB
Python
|
"""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}"}
|